From 9cb5db09524727cc17b32338b593961585b9b257 Mon Sep 17 00:00:00 2001 From: shaohuzhang1 Date: Wed, 15 Jul 2026 19:32:33 +0800 Subject: [PATCH] feat: Add form nodes, interrupt forms, recall --- .../0014_chatrecord_workflow_context.py | 18 ++ apps/application/models/application_chat.py | 1 + apps/application/serializers/common.py | 8 + .../workflow/loop_workflow_manage.py | 16 ++ .../workflow/message/struct/content.py | 13 +- .../workflow/message/struct/form_content.py | 24 ++ .../message/struct/reasoning_content.py | 6 +- .../workflow/message/struct/text_content.py | 6 +- .../workflow/message/struct/tool_content.py | 7 +- apps/application/workflow/nodes/__init__.py | 14 +- .../nodes/ai_chat_node/ai_chat_node.py | 26 +- .../workflow/nodes/form_node/__init__.py | 9 + .../workflow/nodes/form_node/form_node.py | 179 ++++++++++++++ .../workflow/nodes/loop_node/loop_node.py | 60 +++-- .../workflow/nodes/reply_node/reply_node.py | 5 +- apps/application/workflow/workflow_manage.py | 62 ++++- apps/chat/serializers/chat.py | 229 +++++++++++++++--- apps/common/constants/cache_version.py | 2 + 18 files changed, 618 insertions(+), 67 deletions(-) create mode 100644 apps/application/migrations/0014_chatrecord_workflow_context.py create mode 100644 apps/application/workflow/message/struct/form_content.py create mode 100644 apps/application/workflow/nodes/form_node/__init__.py create mode 100644 apps/application/workflow/nodes/form_node/form_node.py diff --git a/apps/application/migrations/0014_chatrecord_workflow_context.py b/apps/application/migrations/0014_chatrecord_workflow_context.py new file mode 100644 index 00000000000..d8d16cfdfbb --- /dev/null +++ b/apps/application/migrations/0014_chatrecord_workflow_context.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.14 on 2026-07-14 08:49 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('application', '0013_application_long_term_enable_and_more'), + ] + + operations = [ + migrations.AddField( + model_name='chatrecord', + name='workflow_context', + field=models.JSONField(blank=True, default=dict, null=True, verbose_name='工作流上下文'), + ), + ] diff --git a/apps/application/models/application_chat.py b/apps/application/models/application_chat.py index e9a3efbc697..6f0a5bbcdb7 100644 --- a/apps/application/models/application_chat.py +++ b/apps/application/models/application_chat.py @@ -108,6 +108,7 @@ class ChatRecord(AppModelMixin): index = models.IntegerField(verbose_name="对话下标") source = models.JSONField(verbose_name="来源", default=dict) ip_address = models.CharField(max_length=128, verbose_name="ip地址", default='') + workflow_context = models.JSONField(verbose_name="工作流上下文", default=dict, null=True, blank=True) def get_human_message(self): if 'problem_padding' in self.details: diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py index b41ce79e470..4c02955f453 100644 --- a/apps/application/serializers/common.py +++ b/apps/application/serializers/common.py @@ -312,6 +312,14 @@ def append_chat_record(self, chat_record: ChatRecord): break if is_save: self.chat_record_list.append(chat_record) + + # 调试模式：把 workflow_context 存到 Redis + if self.debug and chat_record.workflow_context: + from django.core.cache import cache + from common.constants.cache_version import Cache_Version + cache_key = Cache_Version.DEBUG_WORKFLOW_CONTEXT.get_key(chat_record_id=str(chat_record.id)) + cache.set(cache_key, chat_record.workflow_context, timeout=3600) + if not self.debug: if not QuerySet(Chat).filter(id=self.chat_id).exists(): Chat(id=self.chat_id, application_id=self.application_id, abstract=chat_record.problem_text[0:1024], diff --git a/apps/application/workflow/loop_workflow_manage.py b/apps/application/workflow/loop_workflow_manage.py index 4f78a73d9e4..5952c995376 100644 --- a/apps/application/workflow/loop_workflow_manage.py +++ b/apps/application/workflow/loop_workflow_manage.py @@ -35,7 +35,23 @@ def get_parent_context(self, node_id, key): def generate_prompt(self, prompt): prompt = self.workflow.reset_prompt(prompt) + prompt = self.parent_workflow_manage.workflow.reset_prompt(prompt) context = {**self.context, **self.parent_workflow_manage.context} from langchain_core.prompts import PromptTemplate prompt_template = PromptTemplate.from_template(prompt, template_format='jinja2') return prompt_template.format(context=context) + + def get_reference_field(self, node_id, fields): + """ + 获取引用字段，先从当前工作流获取，获取不到再从父工作流获取 + @param node_id: 节点id + @param fields: 字段 + @return: 引用数据 + """ + # 先从当前工作流获取 + result = super().get_reference_field(node_id, fields) + if result is not None: + return result + + # 从父工作流获取 + return self.parent_workflow_manage.get_reference_field(node_id, fields) diff --git a/apps/application/workflow/message/struct/content.py b/apps/application/workflow/message/struct/content.py index 438db668894..2763c2093e0 100644 --- a/apps/application/workflow/message/struct/content.py +++ b/apps/application/workflow/message/struct/content.py @@ -6,6 +6,9 @@ @date：2026/6/30 15:38 @desc: """ +from enum import Enum +from typing import Optional + from application.workflow.content_type import ContentType from application.workflow.status import Status @@ -17,10 +20,18 @@ def __init__(self, _id: str, name: str, status: Status): self.status = status +class Position: + def __init__(self, _id: str, index: Optional[int] = None, children: Optional['Position'] = None): + self.id = _id + self.index = index + self.children = children + + class Content: - def __init__(self, _id, status: Status, _type: ContentType, node_info: NodeInfo, **kwargs): + def __init__(self, _id, status: Status, _type: ContentType, node_info: NodeInfo, position: Position, **kwargs): self.id = _id self.status = status self.type = _type self.node_info = node_info + self.position = position self.extra = kwargs diff --git a/apps/application/workflow/message/struct/form_content.py b/apps/application/workflow/message/struct/form_content.py new file mode 100644 index 00000000000..1cc5e27dff0 --- /dev/null +++ b/apps/application/workflow/message/struct/form_content.py @@ -0,0 +1,24 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： form_content.py + @date：2026/7/6 15:30 + @desc: +""" +from typing import List, Dict, Optional + +from application.workflow.content_type import ContentType +from application.workflow.message.struct.content import Content, NodeInfo, Position +from application.workflow.status import Status + + +class FormContent(Content): + def __init__(self, _id, form_field_list: List[Dict], form_content_format: str, + is_submit: bool, status: Status, node_info: NodeInfo, position: Position, + form_data: Optional[Dict] = None, **kwargs): + self.form_field_list = form_field_list + self.form_content_format = form_content_format + self.is_submit = is_submit + self.form_data = form_data or {} + super().__init__(_id, status, ContentType.FORM, node_info, position, **kwargs) diff --git a/apps/application/workflow/message/struct/reasoning_content.py b/apps/application/workflow/message/struct/reasoning_content.py index 533323e3720..6a5129f4e24 100644 --- a/apps/application/workflow/message/struct/reasoning_content.py +++ b/apps/application/workflow/message/struct/reasoning_content.py @@ -7,11 +7,11 @@ @desc: """ from application.workflow.content_type import ContentType -from application.workflow.message.struct.content import Content, NodeInfo +from application.workflow.message.struct.content import Content, NodeInfo, Position from application.workflow.status import Status class ReasoningContent(Content): - def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, **kwargs): + def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, position: Position, **kwargs): self.content = content - super().__init__(_id, status, ContentType.REASONING, node_info, **kwargs) + super().__init__(_id, status, ContentType.REASONING, node_info, position, **kwargs) diff --git a/apps/application/workflow/message/struct/text_content.py b/apps/application/workflow/message/struct/text_content.py index 55287ff94cf..c2f49f677b7 100644 --- a/apps/application/workflow/message/struct/text_content.py +++ b/apps/application/workflow/message/struct/text_content.py @@ -7,11 +7,11 @@ @desc: """ from application.workflow.content_type import ContentType -from application.workflow.message.struct.content import Content, NodeInfo +from application.workflow.message.struct.content import Content, NodeInfo, Position from application.workflow.status import Status class TextContent(Content): - def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, **kwargs): + def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, position: Position, **kwargs): self.content = content - super().__init__(_id, status, ContentType.TEXT, node_info, **kwargs) + super().__init__(_id, status, ContentType.TEXT, node_info, position, **kwargs) diff --git a/apps/application/workflow/message/struct/tool_content.py b/apps/application/workflow/message/struct/tool_content.py index 95a8f9d129f..e873a545411 100644 --- a/apps/application/workflow/message/struct/tool_content.py +++ b/apps/application/workflow/message/struct/tool_content.py @@ -7,13 +7,14 @@ @desc: """ from application.workflow.content_type import ContentType -from application.workflow.message.struct.content import Content, NodeInfo +from application.workflow.message.struct.content import Content, NodeInfo, Position from application.workflow.status import Status class ToolContent(Content): - def __init__(self, _id, tool_name: str, arguments: str, result: str, status: Status, node_info: NodeInfo, **kwargs): + def __init__(self, _id, tool_name: str, arguments: str, result: str, status: Status, node_info: NodeInfo, + position: Position, **kwargs): self.content = tool_name self.arguments = arguments self.result = result - super().__init__(_id, status, ContentType.TOOL, node_info, **kwargs) + super().__init__(_id, status, ContentType.TOOL, node_info, position, **kwargs) diff --git a/apps/application/workflow/nodes/__init__.py b/apps/application/workflow/nodes/__init__.py index fdda9f7b8c1..3f6eeb92644 100644 --- a/apps/application/workflow/nodes/__init__.py +++ b/apps/application/workflow/nodes/__init__.py @@ -47,14 +47,26 @@ def get_node_class(_type, workflow_type): return node_class -def get_start_node(workflow, workflow_manage, workflow_type): +def get_start_node(workflow, workflow_manage, workflow_type, position=None): """ 获取开始节点实例 @param workflow: 工作流对象 @param workflow_manage 工作流管理器 @param workflow_type: 工作流类型 + @param position: 位置信息（可选） @return: 开始节点实例 """ + # 如果有 position，根据 position 确定开始节点 + if position and position.get('id'): + node_id = position.get('id') + node = workflow.get_node(node_id) + if node: + node_class = get_node_class(node.type, workflow_type) + def get_node_parameters(n): + return n.properties.get('node_data', {}) + return node_class(node, workflow_manage, get_node_parameters) + + # 默认返回开始节点 start_node = workflow.get_node('start-node') if start_node is None: raise ValueError("开始节点不存在") diff --git a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py index 46e35f91ac1..d7d2720e00b 100644 --- a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py +++ b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py @@ -21,7 +21,7 @@ from application.models import Application, ApplicationAccessToken, ApplicationApiKey from application.workflow.common import WorkflowType from application.workflow.i_node import INode -from application.workflow.message.struct.content import NodeInfo +from application.workflow.message.struct.content import NodeInfo, Position from application.workflow.message.struct.reasoning_content import ReasoningContent from application.workflow.message.struct.text_content import TextContent from application.workflow.message.struct.tool_content import ToolContent @@ -261,7 +261,7 @@ def execute(self): if is_result: answer = self.get_context('answer') node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) - self.write(TextContent(text_content_id, answer, Status.SUCCESS, node_info)) + self.write(TextContent(text_content_id, answer, Status.SUCCESS, node_info, Position(self.get_node_id()))) def _generate_prompt_question(self, prompt, model, vision, image_list, video_list): images = [] @@ -301,11 +301,17 @@ def _stream_response(self, response, chat_model, message_list, question, if reasoning_content_chunk is None: reasoning_content_chunk = '' reasoning_content += reasoning_content_chunk - + reasoning_end = False if content_chunk: - self.write(TextContent(text_content_id, content_chunk, Status.RUNNING, node_info)) + if not reasoning_end: + self.write( + ReasoningContent(reasoning_content_id, '', Status.SUCCESS, node_info, + Position(self.get_node_id()))) + self.write(TextContent(text_content_id, content_chunk, Status.RUNNING, node_info, + Position(self.get_node_id()))) if reasoning_content_chunk and model_setting.get('reasoning_content_enable', False): - self.write(ReasoningContent(reasoning_content_id, reasoning_content_chunk, Status.RUNNING, node_info)) + self.write(ReasoningContent(reasoning_content_id, reasoning_content_chunk, Status.RUNNING, node_info, + Position(self.get_node_id()))) reasoning_end = reasoning.get_end_reasoning_content() answer += reasoning_end.get('content') @@ -313,9 +319,11 @@ def _stream_response(self, response, chat_model, message_list, question, if not response_reasoning_content: reasoning_content_chunk = reasoning_end.get('reasoning_content') if reasoning_end.get('content'): - self.write(TextContent(text_content_id, reasoning_end.get('content'), Status.RUNNING, node_info)) + self.write(TextContent(text_content_id, reasoning_end.get('content'), Status.RUNNING, node_info, + Position(self.get_node_id()))) if reasoning_content_chunk and model_setting.get('reasoning_content_enable', False): - self.write(ReasoningContent(reasoning_content_id, reasoning_content_chunk, Status.RUNNING, node_info)) + self.write(ReasoningContent(reasoning_content_id, reasoning_content_chunk, Status.RUNNING, node_info, + Position(self.get_node_id()))) self._write_final_context(chat_model, message_list, question, answer, reasoning_content) @@ -475,6 +483,7 @@ def _handle_mcp( chunk.content, Status.RUNNING, node_info, + Position(self.get_node_id()) )) continue @@ -484,7 +493,8 @@ def _handle_mcp( answer += chunk.content if hasattr(chunk, 'content') else str(chunk) if chunk.content: - self.write(TextContent(text_content_id, chunk.content, Status.RUNNING, node_info)) + self.write(TextContent(text_content_id, chunk.content, Status.RUNNING, node_info, + Position(self.get_node_id()))) self._write_final_context(chat_model, message_list, question.content, answer, '') return True diff --git a/apps/application/workflow/nodes/form_node/__init__.py b/apps/application/workflow/nodes/form_node/__init__.py new file mode 100644 index 00000000000..261db05100c --- /dev/null +++ b/apps/application/workflow/nodes/form_node/__init__.py @@ -0,0 +1,9 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： __init__.py + @date：2026/7/6 15:30 + @desc: +""" +from .form_node import FormNode diff --git a/apps/application/workflow/nodes/form_node/form_node.py b/apps/application/workflow/nodes/form_node/form_node.py new file mode 100644 index 00000000000..6ccb950b66b --- /dev/null +++ b/apps/application/workflow/nodes/form_node/form_node.py @@ -0,0 +1,179 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： form_node.py + @date：2026/7/6 15:30 + @desc: +""" +import copy +import re + +import uuid_utils.compat as uuid +from rest_framework import serializers + +from django.utils.translation import gettext_lazy as _ + +from application.workflow.common import WorkflowType +from application.workflow.i_node import INode, Signal +from application.workflow.message.struct.content import NodeInfo, Position +from application.workflow.message.struct.form_content import FormContent +from application.workflow.status import Status + +_TEMPLATE_RE = re.compile(r'\{\{([^.\s}]+)\.([^.\s}]+)\}\}') + +_MULTI_SELECT_TYPES = {'MultiSelect', 'MultiRow'} + + +def _get_default_option(option_list, _type, value_field): + try: + if option_list and isinstance(option_list, list) and len(option_list) > 0: + default_value_list = [o.get(value_field) for o in option_list if o.get('default')] + if len(default_value_list) == 0: + return [option_list[0].get(value_field)] if _type in _MULTI_SELECT_TYPES else option_list[0].get( + value_field) + else: + return default_value_list if _type in _MULTI_SELECT_TYPES else default_value_list[0] + except Exception: + pass + return [] + + +class FormNodeSerializer(serializers.Serializer): + form_field_list = serializers.ListField(required=True, label=_("Form Configuration")) + form_content_format = serializers.CharField(required=True, label=_('Form output content')) + form_data = serializers.DictField(required=False, allow_null=True, label=_("Form Data")) + is_result = serializers.BooleanField(required=False, label=_('Whether to return content')) + + +class FormNode(INode): + serializer_class = FormNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = 'form-node' + + def execute(self): + node_params = self.get_parameters() + workflow_params = self.workflow_manage.get_parameters() + + # 判断是否是表单提交 + position = workflow_params.get('position') or {} + is_form_submit = position.get('id') == self.node.id + + form_field_list = node_params.get('form_field_list', []) + form_content_format = node_params.get('form_content_format', '') + + if is_form_submit: + # 表单提交：从 workflow_params 获取前端提交的 form_data + form_data = workflow_params.get('form_data') or {} + is_submit = True + # 复用前端传来的 chunk_id + chunk_id = workflow_params.get('chunk_id') or str(uuid.uuid7()) + else: + # 首次执行：从节点参数获取 + form_data = node_params.get('form_data') + is_submit = form_data is not None + # 生成新 chunk_id + chunk_id = str(uuid.uuid7()) + + # 写入 context + self.write_context('is_submit', is_submit) + self.write_context('form_content_format', form_content_format) + + if is_submit: + self.write_context('form_data', form_data) + for key in form_data: + self.write_context(key, form_data.get(key)) + + form_field_list = [self._reset_field(field) for field in form_field_list] + self.write_context('form_field_list', form_field_list) + + # 输出表单内容 + node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) + self.write(FormContent( + chunk_id, form_field_list, form_content_format, + is_submit, Status.SUCCESS, node_info, Position(self.get_node_id()), + form_data=form_data, + )) + + # 如果未提交，中断工作流等待用户提交 + if not is_submit: + self.complete(Status.SUCCESS, signal=Signal.FORM) + return + + # 已提交，继续执行后续节点 + self.complete(Status.SUCCESS) + + def _generate_prompt(self, prompt): + try: + return self.workflow_manage.generate_prompt(prompt) + except Exception: + return prompt + + def _reset_field(self, field): + field = copy.copy(field) + for f in ['field', 'label', 'default_value']: + _value = field.get(f) + if _value is None: + continue + if isinstance(_value, str): + field[f] = self._generate_prompt(_value) + elif f == 'label' and isinstance(_value, dict): + _label_value = _value.get('label') + _value['label'] = self._generate_prompt(_label_value) + tooltip = _value.get('attrs', {}).get('tooltip') + if tooltip is not None: + _value['attrs']['tooltip'] = self._generate_prompt(tooltip) + + input_type = field.get('input_type') + if input_type in {'SingleSelect', 'MultiSelect', 'RadioCard', 'RadioRow', 'MultiRow'}: + if field.get('assignment_method') == 'ref_variables': + option_list_ref = field.get('option_list') + if option_list_ref and len(option_list_ref) >= 2: + option_list = self.workflow_manage.get_reference_field( + option_list_ref[0], option_list_ref[1:]) + option_list = option_list if isinstance(option_list, list) else [] + field['option_list'] = option_list + field['default_value'] = _get_default_option( + option_list, input_type, field.get('value_field')) + + if input_type == 'JsonInput': + if field.get('default_value_assignment_method') == 'ref_variables': + default_ref = field.get('default_value') + if default_ref and isinstance(default_ref, list) and len(default_ref) >= 2: + field['default_value'] = self.workflow_manage.get_reference_field( + default_ref[0], default_ref[1:]) + + self._reset_visibility_rules(field) + return field + + def _reset_visibility_rules(self, field): + visibility_rules = field.get('visibility_rules') + if not visibility_rules or not isinstance(visibility_rules.get('conditions'), list): + return + for cond in visibility_rules['conditions']: + cond_field = cond.get('field') + if not cond_field or len(cond_field) < 2 or not cond_field[0] or not cond_field[1]: + continue + if cond_field[0] != self.node.id: + cond['_left'] = self.workflow_manage.get_reference_field(cond_field[0], cond_field[1:]) + cond_value = cond.get("value") + if isinstance(cond_value, str) and _TEMPLATE_RE.search(cond_value): + cond['value'] = self._render_cond_value(cond_value) + + def _render_cond_value(self, value): + def replacer(match): + node_display = match.group(1) + field_name = match.group(2) + workflow = self.workflow_manage.workflow + for f in workflow.node_field_list: + if f.node_name == node_display and f.value == field_name: + if f.node_id == self.node.id: + return match.group(0) + ref = self.workflow_manage.get_reference_field(f.node_id, [field_name]) + return str(ref) if ref is not None else '' + return match.group(0) + + try: + return _TEMPLATE_RE.sub(replacer, value) + except Exception: + return value diff --git a/apps/application/workflow/nodes/loop_node/loop_node.py b/apps/application/workflow/nodes/loop_node/loop_node.py index 8aeb3f158b1..6c53190fb19 100644 --- a/apps/application/workflow/nodes/loop_node/loop_node.py +++ b/apps/application/workflow/nodes/loop_node/loop_node.py @@ -14,7 +14,7 @@ from application.workflow.common import WorkflowType, new_instance from application.workflow.i_node import INode, Signal -from application.workflow.message.struct.content import NodeInfo +from application.workflow.message.struct.content import NodeInfo, Position from application.workflow.message.struct.text_content import TextContent from application.workflow.status import Status @@ -44,16 +44,16 @@ def is_valid(self, *, raise_exception=False): raise AppApiException(500, message) -def _generate_loop_number(number): - return iter([(i, i) for i in range(number)]) +def _generate_loop_number(number, start_index=0): + return iter([(i, i) for i in range(start_index, number)]) -def _generate_loop_array(array): - return iter([(item, i) for i, item in enumerate(array)]) +def _generate_loop_array(array, start_index=0): + return iter([(item, i) for i, item in enumerate(array) if i >= start_index]) -def _generate_while_loop(number): - return iter([(i, i) for i in range(number)]) +def _generate_while_loop(number, start_index=0): + return iter([(i, i) for i in range(start_index, number)]) class LoopNode(INode): @@ -73,21 +73,26 @@ def execute(self): number = node_params.get('number') loop_body = node_params.get('loop_body') + # 从 position 获取 start_index + position = workflow_params.get('position') or {} + start_index = position.get('index') or 0 if position.get('id') == self.node.id else 0 + if loop_type == 'ARRAY' and isinstance(array, list) and len(array) >= 2: array = self.workflow_manage.get_reference_field(array[0], array[1:]) self.write_context('params', {'loop_type': loop_type, 'array': array, 'number': number}) + # 根据 start_index 构建迭代器 if loop_type == 'ARRAY': - iterator = _generate_loop_array(array) + iterator = _generate_loop_array(array, start_index=start_index) elif loop_type == 'LOOP': - iterator = _generate_while_loop(number or MAX_LOOP_COUNT) + iterator = _generate_while_loop(number or MAX_LOOP_COUNT, start_index=start_index) else: - iterator = _generate_loop_number(number) + iterator = _generate_loop_number(number, start_index=start_index) - self._loop_node_data = [] - self._loop_answer_data = [] - self._answer_text = '' + self._loop_node_data = self.get_context('loop_node_data') or [] + self._loop_answer_data = self.get_context('loop_answer_data') or [] + self._answer_text = self.get_context('answer') or '' self._workflow_params = workflow_params self._loop_body = loop_body self._iterator = iterator @@ -111,7 +116,8 @@ def on_next(wf_manage, content): chunk_list.append(content) if hasattr(content, 'content'): self._answer_text += content.content - self.write(content) + content.position = Position(self.get_node_id(), index, content.position) + self.write(content) def on_complete(wf_manage, error): self._loop_node_data.append(wf_manage.context) @@ -121,7 +127,7 @@ def on_complete(wf_manage, error): self.write_context('index', index) self.write_context('item', item) - if wf_manage.signal == Signal.BREAK: + if wf_manage.signal == Signal.BREAK or wf_manage.signal == Signal.FORM: self.write_context('answer', self._answer_text) self.write_context('run_time', time.time() - self.data.get('start_time', time.time())) self.complete(Status.SUCCESS) @@ -144,13 +150,35 @@ def on_complete(wf_manage, error): loop_start_class = get_node_class('loop-start-node', self.get_workflow_type()) + # 获取 position，当前 index 和 position.index 一致时传入 children + position = self._workflow_params.get('position') or {} + child_position = None + if position.get('id') == self.node.id and position.get('index') == index: + child_position = position.get('children') + def get_start_node_fn(wf, wf_manage): + # 如果有 child_position，从指定节点开始 + if child_position and child_position.get('id'): + node_id = child_position.get('id') + node = wf.get_node(node_id) + if node: + node_class = get_node_class(node.type, self.get_workflow_type()) + return node_class(node, wf_manage, lambda n: n.properties.get('node_data', {})) + + # 默认从 loop-start-node 开始 start_node = wf.get_node('loop-start-node') return loop_start_class(start_node, wf_manage, lambda n: n.properties.get('node_data', {})) + # 构建子工作流参数，第一次迭代传入 child_position + loop_workflow_params = dict(self._workflow_params) + if child_position: + loop_workflow_params['position'] = child_position + else: + loop_workflow_params.pop('position', None) + loop_manage = LoopWorkFlowManage( workflow=workflow, - parameters=self._workflow_params, + parameters=loop_workflow_params, workflow_type=self.get_workflow_type(), call_back=call_back, get_start_node=get_start_node_fn, diff --git a/apps/application/workflow/nodes/reply_node/reply_node.py b/apps/application/workflow/nodes/reply_node/reply_node.py index 3c27efa263b..4c0da9ff029 100644 --- a/apps/application/workflow/nodes/reply_node/reply_node.py +++ b/apps/application/workflow/nodes/reply_node/reply_node.py @@ -13,7 +13,7 @@ from application.workflow.common import WorkflowType from application.workflow.i_node import INode -from application.workflow.message.struct.content import NodeInfo +from application.workflow.message.struct.content import NodeInfo, Position from application.workflow.message.struct.text_content import TextContent from application.workflow.status import Status @@ -33,7 +33,6 @@ class ReplyNode(INode): def execute(self): node_params = self.get_parameters() - workflow_params = self.get_workflow_parameters() chunk_id = uuid.uuid7() reply_type = node_params.get('reply_type') @@ -50,7 +49,7 @@ def execute(self): if is_result: node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) - self.write(TextContent(str(chunk_id), result, Status.SUCCESS, node_info)) + self.write(TextContent(str(chunk_id), result, Status.SUCCESS, node_info, Position(self.get_node_id()))) def _generate_reply_content(self, prompt): if prompt is None: diff --git a/apps/application/workflow/workflow_manage.py b/apps/application/workflow/workflow_manage.py index cd978e541e0..157aa5001b2 100644 --- a/apps/application/workflow/workflow_manage.py +++ b/apps/application/workflow/workflow_manage.py @@ -55,6 +55,7 @@ def __init__(self, self.parameters = parameters self.context = {} self.nodes = [] + self.node_dict = {} self.signal = None self.start_node = get_start_node(workflow, self) @@ -64,6 +65,7 @@ def run(self): @return: None """ self.nodes.append(self.start_node) + self.node_dict = {node.node.id: node for node in self.nodes} self._run_async(self.start_node) def _run(self, node): @@ -77,11 +79,26 @@ def next_nodes(self, nodes: Optional[List[Node]]): """ if nodes is None or len(nodes) == 0: return + # 需要校验是否可执行 + for n in nodes: + condition = n.properties.get("condition") + if condition == 'AND': + up_nodes = self.workflow.get_up_nodes(n.id) + # 如果是AND就是前面所有节点都执行结束 + unfinished = {Status.BEFORE_RUNNING, Status.RUNNING} + end = all( + [self.node_dict.get(node.id) and self.node_dict.get(node.id).status not in unfinished for node in + up_nodes]) + if not end: + return + with self._lock: from application.workflow.nodes import get_node_class instances = [get_node_class(n.type, self.workflow_type)(n, self, get_node_parameters) for n in nodes] self.nodes.extend(instances) + for node in instances: + self.node_dict[node.node.id] = node for inst in instances: self._run_async(inst) @@ -124,7 +141,10 @@ def get_context(self, node_id, key): @param key: key @return: 数据 """ - return self.context.setdefault(node_id).get(key) + node_context = self.context.get(node_id) + if node_context is None: + return None + return node_context.get(key) def _run_async(self, node): t = threading.Thread(target=lambda: self._run(node)) @@ -187,3 +207,43 @@ def get_reference_field(self, node_id, fields): else: return None return obj + + @classmethod + def from_context(cls, chat_record_id, workflow, parameters, workflow_type, + call_back, get_start_node): + """从历史 context 恢复 WorkflowManage""" + from application.models import ChatRecord + from django.core.cache import cache + from common.constants.cache_version import Cache_Version + + try: + context_data = None + + # 先从 Redis 查（调试模式） + cache_key = Cache_Version.DEBUG_WORKFLOW_CONTEXT.get_key(chat_record_id=str(chat_record_id)) + context_data = cache.get(cache_key) + + # Redis 没有，从数据库查 + if not context_data: + chat_record = ChatRecord.objects.filter(id=chat_record_id).first() + if not chat_record or not chat_record.workflow_context: + return None + context_data = chat_record.workflow_context + + # 创建 WorkflowManage 实例 + instance = cls( + workflow=workflow, + parameters=parameters, + workflow_type=workflow_type, + call_back=call_back, + get_start_node=get_start_node + ) + + # 恢复全局 context + instance.context = context_data + + return instance + except Exception as e: + import traceback + traceback.print_exc() + return None diff --git a/apps/chat/serializers/chat.py b/apps/chat/serializers/chat.py index 58f23136638..248aa7af46a 100644 --- a/apps/chat/serializers/chat.py +++ b/apps/chat/serializers/chat.py @@ -24,10 +24,15 @@ BaseGenerateHumanMessageStep from application.chat_pipeline.step.reset_problem_step.impl.base_reset_problem_step import BaseResetProblemStep from application.chat_pipeline.step.search_dataset_step.impl.base_search_dataset_step import BaseSearchDatasetStep -from application.flow.common import Answer, Workflow -from application.flow.i_step_node import WorkFlowPostHandler +from application.flow.common import Answer from application.flow.tools import to_stream_response_simple -from application.flow.workflow_manage import WorkflowManage +from application.workflow.common import WorkflowType, new_instance +from application.workflow.workflow_manage import WorkflowManage, CallBack +from application.workflow.nodes import get_start_node +from application.workflow.message.struct.text_content import TextContent +from application.workflow.message.struct.reasoning_content import ReasoningContent +from application.workflow.message.struct.tool_content import ToolContent +from application.workflow.message.struct.form_content import FormContent from application.models import Application, ApplicationTypeChoices, \ ChatUserType, ApplicationChatUserStats, ApplicationAccessToken, ChatRecord, Chat, ApplicationVersion from application.serializers.application import ApplicationOperateSerializer @@ -385,12 +390,14 @@ def get_chat_record(chat_info, chat_record_id): chat_record = QuerySet(ChatRecord).filter(id=chat_record_id, chat_id=chat_info.chat_id).first() if chat_record is None: raise ChatException(500, _("Conversation record does not exist")) - + return chat_record chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first() return chat_record def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response): + import queue + message = instance.get('message') re_chat = instance.get('re_chat') stream = instance.get('stream') @@ -406,38 +413,204 @@ def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response): other_list = instance.get('other_list') workspace_id = chat_info.application.workspace_id chat_record_id = instance.get('chat_record_id') + position = instance.get('position') + chunk_id = instance.get('chunk_id') debug = self.data.get('debug', False) - chat_record = None history_chat_record = chat_info.chat_record_list if chat_record_id is not None: chat_record = self.get_chat_record(chat_info, chat_record_id) if chat_record: history_chat_record = [r for r in chat_info.chat_record_list if str(r.id) != chat_record_id] + work_flow = chat_info.application.work_flow - work_flow_manage = WorkflowManage(Workflow.new_instance(work_flow), - {'history_chat_record': history_chat_record, 'question': message, - 'chat_id': chat_info.chat_id, 'chat_record_id': str( - uuid.uuid7()) if chat_record_id is None else str(chat_record_id), - 'stream': stream, - 're_chat': re_chat, - 'chat_user_id': chat_user_id, - 'chat_user_type': chat_user_type, - 'ip_address': ip_address, - 'source': source, - 'workspace_id': workspace_id, - 'debug': debug, - 'chat_user': chat_info.get_chat_user(), - 'chat_user_group': chat_info.get_chat_user_group(), - 'application_id': str(chat_info.application_id)}, - WorkFlowPostHandler(chat_info), - base_to_response, form_data, image_list, document_list, audio_list, - video_list, - other_list, - instance.get('runtime_node_id'), - instance.get('node_data'), chat_record, instance.get('child_node')) + workflow = new_instance(work_flow, WorkflowType.APPLICATION) + + chat_record_id_str = str(uuid.uuid7()) if chat_record_id is None else str(chat_record_id) + + parameters = { + 'history_chat_record': history_chat_record, + 'question': message, + 'chat_id': chat_info.chat_id, + 'chat_record_id': chat_record_id_str, + 'stream': stream, + 're_chat': re_chat, + 'chat_user_id': chat_user_id, + 'chat_user_type': chat_user_type, + 'ip_address': ip_address, + 'source': source, + 'workspace_id': workspace_id, + 'debug': debug, + 'chat_user': chat_info.get_chat_user(), + 'chat_user_group': chat_info.get_chat_user_group(), + 'application_id': str(chat_info.application_id), + 'form_data': form_data or {}, + 'position': position, + 'chunk_id': chunk_id, + 'image_list': image_list or [], + 'document_list': document_list or [], + 'audio_list': audio_list or [], + 'video_list': video_list or [], + 'other_list': other_list or [], + } + + result_queue = queue.Queue() + answer_text = '' + reasoning_text = '' + + def on_next(wf_manage, content): + nonlocal answer_text, reasoning_text + if isinstance(content, TextContent): + answer_text += content.content + result_queue.put(('chunk', { + 'content': [{ + 'id': content.id, + 'type': 'TEXT', + 'content': content.content, + }] + })) + elif isinstance(content, ReasoningContent): + reasoning_text += content.content + result_queue.put(('chunk', { + 'content': [{ + 'id': content.id, + 'type': 'REASONING', + 'content': content.content, + 'status': content.status.value if content.status else None, + }] + })) + elif isinstance(content, ToolContent): + result_queue.put(('chunk', { + 'content': [{ + 'id': content.id, + 'type': 'TOOL', + 'content': content.content, + 'arguments': content.arguments, + 'result': content.result, + 'status': content.status.value if content.status else None, + }] + })) + elif isinstance(content, FormContent): + def position_to_dict(pos): + if pos is None: + return None + return { + 'id': pos.id, + 'index': pos.index, + 'children': position_to_dict(pos.children) + } + + result_queue.put(('chunk', { + 'content': [{ + 'id': content.id, + 'type': 'FORM', + 'form_field_list': content.form_field_list, + 'form_content_format': content.form_content_format, + 'is_submit': content.is_submit, + 'form_data': content.form_data, + 'status': content.status.value if content.status else None, + 'position': position_to_dict(content.position), + 'chat_record_id': chat_record_id_str, + }] + })) + + def on_complete(wf_manage, error): + if error: + result_queue.put(('error', error)) + else: + self._save_chat_record(chat_info, chat_info.chat_id, chat_record_id_str, + message, answer_text, reasoning_text, wf_manage) + result_queue.put(('done', None)) + + call_back = CallBack(on_next, on_complete) + + def get_start_node_fn(wf, wm): + return get_start_node(wf, wm, WorkflowType.APPLICATION, position) + + # 判断是否是 Form 提交（有 position 和 chat_record_id） + if position and chat_record_id: + # 从历史 context 恢复 + work_flow_manage = WorkflowManage.from_context( + chat_record_id=chat_record_id, + workflow=workflow, + parameters=parameters, + workflow_type=WorkflowType.APPLICATION, + call_back=call_back, + get_start_node=get_start_node_fn + ) + if work_flow_manage is None: + # 恢复失败，回退到正常流程 + work_flow_manage = WorkflowManage(workflow, parameters, WorkflowType.APPLICATION, + call_back, get_start_node_fn) + else: + # 正常创建新的 WorkflowManage + work_flow_manage = WorkflowManage(workflow, parameters, WorkflowType.APPLICATION, + call_back, get_start_node_fn) + + work_flow_manage.start_node.workflow_manage = work_flow_manage + chat_info.set_chat(message) - r = work_flow_manage.run() - return r + + if stream: + def generate(): + work_flow_manage.run() + while True: + msg_type, data = result_queue.get() + if msg_type == 'done': + yield 'data: [DONE]\n\n' + break + if msg_type == 'error': + yield 'data: ' + json.dumps({ + 'chat_id': str(chat_info.chat_id), + 'chat_record_id': chat_record_id_str, + 'content': [{'type': 'FAILURE', 'content': str(data)}] + }, ensure_ascii=False) + '\n\n' + yield 'data: [DONE]\n\n' + break + if msg_type == 'chunk': + data['chat_id'] = str(chat_info.chat_id) + data['chat_record_id'] = chat_record_id_str + yield 'data: ' + json.dumps(data, ensure_ascii=False) + '\n\n' + + return to_stream_response_simple(generate()) + else: + work_flow_manage.run() + while True: + msg_type, data = result_queue.get() + if msg_type == 'done': + break + if msg_type == 'error': + raise data + return base_to_response.to_block_response( + chat_info.chat_id, chat_record_id_str, answer_text, True, 0, 0) + + def _save_chat_record(self, chat_info, chat_id, chat_record_id, question, answer, + reasoning_content, wf_manage): + context = wf_manage.context + message_tokens = sum( + v.get('message_tokens', 0) for v in context.values() if isinstance(v, dict) and 'message_tokens' in v) + answer_tokens = sum( + v.get('answer_tokens', 0) for v in context.values() if isinstance(v, dict) and 'answer_tokens' in v) + + answer_text_list = [[Answer(answer, 'ai-chat-node', 'ai-chat-node', 'ai-chat-node', {}, + 'ai-chat-node', reasoning_content).to_dict()]] + + chat_record = ChatRecord( + id=chat_record_id, + chat_id=chat_id, + problem_text=question, + answer_text=answer, + details={}, + message_tokens=message_tokens, + answer_tokens=answer_tokens, + answer_text_list=answer_text_list, + run_time=0, + index=len(chat_info.chat_record_list) + 1, + ip_address=chat_info.ip_address, + source=chat_info.source, + workflow_context=dict(context) if context else None + ) + chat_info.append_chat_record(chat_record) + chat_info.set_cache() def is_valid_chat_user(self): chat_user_id = self.data.get('chat_user_id') diff --git a/apps/common/constants/cache_version.py b/apps/common/constants/cache_version.py index 0aed4715e2e..5642b7929ae 100644 --- a/apps/common/constants/cache_version.py +++ b/apps/common/constants/cache_version.py @@ -41,6 +41,8 @@ class Cache_Version(Enum): TOOL_WORKFLOW_EXECUTE = "TOOL_WORKFLOW_EXECUTE", lambda key: key + DEBUG_WORKFLOW_CONTEXT = "DEBUG_WORKFLOW_CONTEXT", lambda chat_record_id: chat_record_id + def get_version(self): return self.value[0]