fix: 修复子应用表单调用无法调用问题 (#1741)

Author: shaohuzhang1

apps/application/flow/i_step_node.py

@@ -40,6 +40,10 @@ def write_context(step_variable: Dict, global_variable: Dict, node, workflow):
     node.context['run_time'] = time.time() - node.context['start_time']
 
 
+def is_interrupt(node, step_variable: Dict, global_variable: Dict):
+    return node.type == 'form-node' and not node.context.get('is_submit', False)
+
+
 class WorkFlowPostHandler:
     def __init__(self, chat_info, client_id, client_type):
         self.chat_info = chat_info
@@ -57,7 +61,7 @@ def handler(self, chat_id,
         answer_tokens = sum([row.get('answer_tokens') for row in details.values() if
                              'answer_tokens' in row and row.get('answer_tokens') is not None])
         answer_text_list = workflow.get_answer_text_list()
-        answer_text = '\n\n'.join(answer_text_list)
+        answer_text = '\n\n'.join(answer['content'] for answer in answer_text_list)
         if workflow.chat_record is not None:
             chat_record = workflow.chat_record
             chat_record.answer_text = answer_text
@@ -91,17 +95,26 @@ def handler(self, chat_id,
 
 class NodeResult:
     def __init__(self, node_variable: Dict, workflow_variable: Dict,
-                 _write_context=write_context):
+                 _write_context=write_context, _is_interrupt=is_interrupt):
         self._write_context = _write_context
         self.node_variable = node_variable
         self.workflow_variable = workflow_variable
+        self._is_interrupt = _is_interrupt
 
     def write_context(self, node, workflow):
         return self._write_context(self.node_variable, self.workflow_variable, node, workflow)
 
     def is_assertion_result(self):
         return 'branch_id' in self.node_variable
 
+    def is_interrupt_exec(self, current_node):
+        """
+        是否中断执行
+        @param current_node:
+        @return:
+        """
+        return self._is_interrupt(current_node, self.node_variable, self.workflow_variable)
+
 
 class ReferenceAddressSerializer(serializers.Serializer):
     node_id = serializers.CharField(required=True, error_messages=ErrMessage.char("节点id"))
@@ -139,14 +152,18 @@ def save_context(self, details, workflow_manage):
         pass
 
     def get_answer_text(self):
-        return self.answer_text
+        if self.answer_text is None:
+            return None
+        return {'content': self.answer_text, 'runtime_node_id': self.runtime_node_id,
+                'chat_record_id': self.workflow_params['chat_record_id']}
 
-    def __init__(self, node, workflow_params, workflow_manage, up_node_id_list=None):
+    def __init__(self, node, workflow_params, workflow_manage, up_node_id_list=None,
+                 get_node_params=lambda node: node.properties.get('node_data')):
         # 当前步骤上下文,用于存储当前步骤信息
         self.status = 200
         self.err_message = ''
         self.node = node
-        self.node_params = node.properties.get('node_data')
+        self.node_params = get_node_params(node)
         self.workflow_params = workflow_params
         self.workflow_manage = workflow_manage
         self.node_params_serializer = None

apps/application/flow/step_node/application_node/i_application_node.py

@@ -14,6 +14,8 @@ class ApplicationNodeSerializer(serializers.Serializer):
     user_input_field_list = serializers.ListField(required=False, error_messages=ErrMessage.uuid("用户输入字段"))
     image_list = serializers.ListField(required=False, error_messages=ErrMessage.list("图片"))
     document_list = serializers.ListField(required=False, error_messages=ErrMessage.list("文档"))
+    child_node = serializers.DictField(required=False, allow_null=True, error_messages=ErrMessage.dict("子节点"))
+    node_data = serializers.DictField(required=False, allow_null=True, error_messages=ErrMessage.dict("表单数据"))
 
 
 class IApplicationNode(INode):
@@ -55,5 +57,5 @@ def _run(self):
                             message=str(question), **kwargs)
 
     def execute(self, application_id, message, chat_id, chat_record_id, stream, re_chat, client_id, client_type,
-                app_document_list=None, app_image_list=None, **kwargs) -> NodeResult:
+                app_document_list=None, app_image_list=None, child_node=None, node_data=None, **kwargs) -> NodeResult:
         pass

apps/application/flow/step_node/application_node/impl/base_application_node.py

@@ -2,19 +2,25 @@
 import json
 import time
 import uuid
-from typing import List, Dict
+from typing import Dict
+
 from application.flow.i_step_node import NodeResult, INode
 from application.flow.step_node.application_node.i_application_node import IApplicationNode
 from application.models import Chat
-from common.handle.impl.response.openai_to_response import OpenaiToResponse
 
 
 def string_to_uuid(input_str):
     return str(uuid.uuid5(uuid.NAMESPACE_DNS, input_str))
 
 
+def _is_interrupt_exec(node, node_variable: Dict, workflow_variable: Dict):
+    return node_variable.get('is_interrupt_exec', False)
+
+
 def _write_context(node_variable: Dict, workflow_variable: Dict, node: INode, workflow, answer: str):
     result = node_variable.get('result')
+    node.context['child_node'] = node_variable['child_node']
+    node.context['is_interrupt_exec'] = node_variable['is_interrupt_exec']
     node.context['message_tokens'] = result.get('usage', {}).get('prompt_tokens', 0)
     node.context['answer_tokens'] = result.get('usage', {}).get('completion_tokens', 0)
     node.context['answer'] = answer
@@ -36,17 +42,34 @@ def write_context_stream(node_variable: Dict, workflow_variable: Dict, node: INo
     response = node_variable.get('result')
     answer = ''
     usage = {}
+    node_child_node = {}
+    is_interrupt_exec = False
     for chunk in response:
         # 先把流转成字符串
         response_content = chunk.decode('utf-8')[6:]
         response_content = json.loads(response_content)
-        choices = response_content.get('choices')
-        if choices and isinstance(choices, list) and len(choices) > 0:
-            content = choices[0].get('delta', {}).get('content', '')
-            answer += content
-            yield content
+        content = response_content.get('content', '')
+        runtime_node_id = response_content.get('runtime_node_id', '')
+        chat_record_id = response_content.get('chat_record_id', '')
+        child_node = response_content.get('child_node')
+        node_type = response_content.get('node_type')
+        real_node_id = response_content.get('real_node_id')
+        node_is_end = response_content.get('node_is_end', False)
+        if node_type == 'form-node':
+            is_interrupt_exec = True
+        answer += content
+        node_child_node = {'runtime_node_id': runtime_node_id, 'chat_record_id': chat_record_id,
+                           'child_node': child_node}
+        yield {'content': content,
+               'node_type': node_type,
+               'runtime_node_id': runtime_node_id, 'chat_record_id': chat_record_id,
+               'child_node': child_node,
+               'real_node_id': real_node_id,
+               'node_is_end': node_is_end}
         usage = response_content.get('usage', {})
     node_variable['result'] = {'usage': usage}
+    node_variable['is_interrupt_exec'] = is_interrupt_exec
+    node_variable['child_node'] = node_child_node
     _write_context(node_variable, workflow_variable, node, workflow, answer)
 
 
@@ -64,6 +87,11 @@ def write_context(node_variable: Dict, workflow_variable: Dict, node: INode, wor
 
 
 class BaseApplicationNode(IApplicationNode):
+    def get_answer_text(self):
+        if self.answer_text is None:
+            return None
+        return {'content': self.answer_text, 'runtime_node_id': self.runtime_node_id,
+                'chat_record_id': self.workflow_params['chat_record_id'], 'child_node': self.context.get('child_node')}
 
     def save_context(self, details, workflow_manage):
         self.context['answer'] = details.get('answer')
@@ -72,7 +100,7 @@ def save_context(self, details, workflow_manage):
         self.answer_text = details.get('answer')
 
     def execute(self, application_id, message, chat_id, chat_record_id, stream, re_chat, client_id, client_type,
-                app_document_list=None, app_image_list=None,
+                app_document_list=None, app_image_list=None, child_node=None, node_data=None,
                 **kwargs) -> NodeResult:
         from application.serializers.chat_message_serializers import ChatMessageSerializer
         # 生成嵌入应用的chat_id
@@ -85,6 +113,14 @@ def execute(self, application_id, message, chat_id, chat_record_id, stream, re_c
             app_document_list = []
         if app_image_list is None:
             app_image_list = []
+        runtime_node_id = None
+        record_id = None
+        child_node_value = None
+        if child_node is not None:
+            runtime_node_id = child_node.get('runtime_node_id')
+            record_id = child_node.get('chat_record_id')
+            child_node_value = child_node.get('child_node')
+
         response = ChatMessageSerializer(
             data={'chat_id': current_chat_id, 'message': message,
                   're_chat': re_chat,
@@ -94,16 +130,20 @@ def execute(self, application_id, message, chat_id, chat_record_id, stream, re_c
                   'client_type': client_type,
                   'document_list': app_document_list,
                   'image_list': app_image_list,
-                  'form_data': kwargs}).chat(base_to_response=OpenaiToResponse())
+                  'runtime_node_id': runtime_node_id,
+                  'chat_record_id': record_id,
+                  'child_node': child_node_value,
+                  'node_data': node_data,
+                  'form_data': kwargs}).chat()
         if response.status_code == 200:
             if stream:
                 content_generator = response.streaming_content
                 return NodeResult({'result': content_generator, 'question': message}, {},
-                                  _write_context=write_context_stream)
+                                  _write_context=write_context_stream, _is_interrupt=_is_interrupt_exec)
             else:
                 data = json.loads(response.content)
                 return NodeResult({'result': data, 'question': message}, {},
-                                  _write_context=write_context)
+                                  _write_context=write_context, _is_interrupt=_is_interrupt_exec)
 
     def get_details(self, index: int, **kwargs):
         global_fields = []

apps/application/flow/step_node/form_node/i_form_node.py

@@ -17,6 +17,7 @@
 class FormNodeParamsSerializer(serializers.Serializer):
     form_field_list = serializers.ListField(required=True, error_messages=ErrMessage.list("表单配置"))
     form_content_format = serializers.CharField(required=True, error_messages=ErrMessage.char('表单输出内容'))
+    form_data = serializers.DictField(required=False, allow_null=True, error_messages=ErrMessage.dict("表单数据"))
 
 
 class IFormNode(INode):
@@ -29,5 +30,5 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
     def _run(self):
         return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
 
-    def execute(self, form_field_list, form_content_format, **kwargs) -> NodeResult:
+    def execute(self, form_field_list, form_content_format, form_data, **kwargs) -> NodeResult:
         pass

apps/application/flow/step_node/form_node/impl/base_form_node.py

@@ -42,7 +42,12 @@ def save_context(self, details, workflow_manage):
             for key in form_data:
                 self.context[key] = form_data[key]
 
-    def execute(self, form_field_list, form_content_format, **kwargs) -> NodeResult:
+    def execute(self, form_field_list, form_content_format, form_data, **kwargs) -> NodeResult:
+        if form_data is not None:
+            self.context['is_submit'] = True
+            self.context['form_data'] = form_data
+        else:
+            self.context['is_submit'] = False
         form_setting = {"form_field_list": form_field_list, "runtime_node_id": self.runtime_node_id,
                         "chat_record_id": self.flow_params_serializer.data.get("chat_record_id"),
                         "is_submit": self.context.get("is_submit", False)}
@@ -63,7 +68,8 @@ def get_answer_text(self):
         form = f'<form_rander>{json.dumps(form_setting)}</form_rander>'
         prompt_template = PromptTemplate.from_template(form_content_format, template_format='jinja2')
         value = prompt_template.format(form=form)
-        return value
+        return {'content': value, 'runtime_node_id': self.runtime_node_id,
+                'chat_record_id': self.workflow_params['chat_record_id']}
 
     def get_details(self, index: int, **kwargs):
         form_content_format = self.context.get('form_content_format')