diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py index e238984b5ed..e874f33f421 100644 --- a/apps/application/serializers/common.py +++ b/apps/application/serializers/common.py @@ -121,6 +121,28 @@ def set_record(self, tool_record): ) +def load_debug_workflow_context(chat_record_id): + """ + 按记录 id 解析历史工作流 context:优先 Redis 调试缓存 DEBUG_WORKFLOW_CONTEXT,其次 DB ChatRecord.workflow_context。 + 属业务层逻辑(依赖 ChatRecord),供基于 ChatRecord 的续跑场景(应用对话、子应用节点)复用, + 作为 WorkflowManage.from_context 的 get_context 回调传入;引擎本身不关心 context 来源。 + """ + try: + cache_key = Cache_Version.DEBUG_WORKFLOW_CONTEXT.get_key(chat_record_id=str(chat_record_id)) + context_data = cache.get(cache_key) + 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 + return context_data + except Exception: + import traceback + + traceback.print_exc() + return None + + def resolve_chat_user(chat_user_id, chat_user_type, asker=None): """ 根据对话用户 id / 类型解析出对话用户信息。 diff --git a/apps/application/workflow/i_node.py b/apps/application/workflow/i_node.py index 19431851073..772020bcc14 100644 --- a/apps/application/workflow/i_node.py +++ b/apps/application/workflow/i_node.py @@ -8,6 +8,7 @@ """ import time +import traceback from enum import Enum from typing import Optional, Type, Callable @@ -99,6 +100,7 @@ def run(self): except CancelledException: self.complete(Status.CANCELLED) except Exception as e: + traceback.print_exc() self.complete(Status.FAIL, error=e) def _run(self): diff --git a/apps/application/workflow/loop_workflow_manage.py b/apps/application/workflow/loop_workflow_manage.py index 5952c995376..16d103c6ce7 100644 --- a/apps/application/workflow/loop_workflow_manage.py +++ b/apps/application/workflow/loop_workflow_manage.py @@ -1,11 +1,12 @@ # coding=utf-8 """ - @project: MaxKB - @Author:虎虎虎 - @file: loop_workflow_manage.py - @date:2026/7/2 10:00 - @desc: +@project: MaxKB +@Author:虎虎虎 +@file: loop_workflow_manage.py +@date:2026/7/2 10:00 +@desc: """ + from typing import Dict, Optional, Callable from application.workflow.common import Workflow, WorkflowType, Node @@ -14,21 +15,20 @@ class LoopWorkFlowManage(WorkflowManage): - - def __init__(self, - workflow: Workflow, - parameters: Dict, - workflow_type: WorkflowType, - call_back: CallBack, - get_start_node: Callable[[Workflow, WorkflowManage], INode], - parent_workflow_manage: WorkflowManage, - loop_context: Dict = None): + def __init__( + self, + workflow: Workflow, + parameters: Dict, + workflow_type: WorkflowType, + call_back: CallBack, + get_start_node: Callable[[Workflow, WorkflowManage], INode], + parent_workflow_manage: WorkflowManage, + ): self.parent_workflow_manage = parent_workflow_manage - self.loop_context = loop_context or {} super().__init__(workflow, parameters, workflow_type, call_back, get_start_node) def get_parameters(self): - return {**self.parameters, **self.loop_context} + return self.parameters def get_parent_context(self, node_id, key): return self.parent_workflow_manage.get_context(node_id, key) @@ -38,7 +38,8 @@ def generate_prompt(self, 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') + + prompt_template = PromptTemplate.from_template(prompt, template_format="jinja2") return prompt_template.format(context=context) def get_reference_field(self, node_id, fields): @@ -55,3 +56,28 @@ def get_reference_field(self, node_id, fields): # 从父工作流获取 return self.parent_workflow_manage.get_reference_field(node_id, fields) + + @classmethod + def from_context( + cls, get_context, workflow, parameters, workflow_type, call_back, get_start_node, parent_workflow_manage=None + ): + try: + context = get_context() + + instance = cls( + workflow=workflow, + parameters=parameters, + workflow_type=workflow_type, + call_back=call_back, + get_start_node=get_start_node, + parent_workflow_manage=parent_workflow_manage, + ) + if context: + instance.context = context + + return instance + except Exception: + import traceback + + traceback.print_exc() + return None diff --git a/apps/application/workflow/nodes/application_node/application_node.py b/apps/application/workflow/nodes/application_node/application_node.py index 85a0b6c7454..c5396382277 100644 --- a/apps/application/workflow/nodes/application_node/application_node.py +++ b/apps/application/workflow/nodes/application_node/application_node.py @@ -213,8 +213,10 @@ def get_start_node_fn(wf, wm): # 表单提交:从历史 context 恢复子应用;否则全新运行 if is_submit: + from application.serializers.common import load_debug_workflow_context + sub_manage = WorkflowManage.from_context( - chat_record_id=sub_chat_record_id, + get_context=lambda: load_debug_workflow_context(sub_chat_record_id), workflow=sub_workflow, parameters=sub_parameters, workflow_type=WorkflowType.APPLICATION, diff --git a/apps/application/workflow/nodes/loop_node/loop_node.py b/apps/application/workflow/nodes/loop_node/loop_node.py index e5266bee511..03c65d4bc3d 100644 --- a/apps/application/workflow/nodes/loop_node/loop_node.py +++ b/apps/application/workflow/nodes/loop_node/loop_node.py @@ -14,11 +14,8 @@ from application.workflow.common import WorkflowType, new_instance from application.workflow.i_node import INode, Signal - -from application.workflow.message.struct.content import NodeInfo, Position -from application.workflow.message.struct.text_content import TextContent +from application.workflow.message.struct.content import Position from application.workflow.status import Status - from common.exception.app_exception import AppApiException MAX_LOOP_COUNT = 500 @@ -61,6 +58,8 @@ class LoopNode(INode): serializer_class = LoopNodeSerializer supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] type = "loop-node" + _workflow_params = None + _iterator = None def _run(self): self.execute() @@ -81,7 +80,7 @@ def execute(self): 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}) + self.data["params"] = {"loop_type": loop_type, "array": array, "number": number} # 根据 start_index 构建迭代器 if loop_type == "ARRAY": @@ -90,12 +89,8 @@ def execute(self): iterator = _generate_while_loop(number or MAX_LOOP_COUNT, start_index=start_index) else: iterator = _generate_loop_number(number, start_index=start_index) - - 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.data["loop_body"] = loop_body self._iterator = iterator self._run_next() @@ -104,35 +99,31 @@ def _run_next(self): try: item, index = next(self._iterator) except StopIteration: - self.write_context("answer", self._answer_text) - self.write_context("run_time", time.time() - self.data.get("start_time", time.time())) + self.data["run_time"] = time.time() - self.data.get("start_time", time.time()) self.complete(Status.SUCCESS) return - loop_context = {"index": index, "item": item} - workflow = new_instance(self._loop_body, self.get_workflow_type()) + workflow = new_instance(self.data["loop_body"], self.get_workflow_type()) chunk_list = [] def on_next(wf_manage, content): chunk_list.append(content) - if hasattr(content, "content"): - self._answer_text += content.content content.position = Position(self.get_node_id(), index, content.position) self.write(content) def on_complete(wf_manage, error): loop_details_list = self.data.setdefault("loop_details_list", []) loop_details_list.append(wf_manage.get_details()) - self._loop_node_data.append(wf_manage.context) - self._loop_answer_data.append([c.to_dict() for c in chunk_list]) - self.write_context("loop_node_data", self._loop_node_data) - self.write_context("loop_answer_data", self._loop_answer_data) self.write_context("index", index) self.write_context("item", item) + last_context = self.workflow_manage.get_context(self.node.id, "last_context") + if last_context: + self.write_context("last_context", {**last_context, **wf_manage.context}) + else: + self.write_context("last_context", wf_manage.context) 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.data["run_time"] = time.time() - self.data.get("start_time", time.time()) self.complete(Status.SUCCESS) return @@ -173,40 +164,41 @@ def get_start_node_fn(wf, wf_manage): start_node = wf.get_node("loop-start-node") return loop_start_class(start_node, wf_manage, lambda n: n.properties.get("node_data", {})) + def get_context(): + last_context = self.workflow_manage.get_context(self.node.id, "last_context") or {} + if last_context: + return last_context + return {} + # 构建子工作流参数,第一次迭代传入 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( + loop_workflow_params["index"] = index + loop_workflow_params["item"] = item + loop_manage = LoopWorkFlowManage.from_context( workflow=workflow, parameters=loop_workflow_params, workflow_type=self.get_workflow_type(), call_back=call_back, get_start_node=get_start_node_fn, parent_workflow_manage=self.workflow_manage, - loop_context=loop_context, + get_context=get_context, ) - loop_manage.start_node.workflow_manage = loop_manage loop_manage.run() def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs): details = super().get_details(index, position, old_details, **kwargs) details.update( - { - "params": self.get_context("params"), - "index": self.get_context("index"), - "item": self.get_context("item"), - "answer": self.get_context("answer"), - } + {"params": self.data.get("params"), "index": self.get_context("index"), "item": self.get_context("item")} ) loop_details = [] position_index = 0 loop_position_index = 0 if old_details and position: - for index, item in enumerate(old_details.get("iteration_details") or []): + for index, item in enumerate(old_details.get("children") or []): loop_position_index = index loop_details.append(item) current_details = loop_details[loop_position_index] diff --git a/apps/application/workflow/nodes/loop_start_node/loop_start_node.py b/apps/application/workflow/nodes/loop_start_node/loop_start_node.py index 53943087a54..d0b25af9188 100644 --- a/apps/application/workflow/nodes/loop_start_node/loop_start_node.py +++ b/apps/application/workflow/nodes/loop_start_node/loop_start_node.py @@ -23,12 +23,18 @@ class LoopStartNode(INode): type = "loop-start-node" def execute(self): - loop_context = getattr(self.workflow_manage, "loop_context", {}) - index = loop_context.get("index", 0) - item = loop_context.get("item", None) - - self.write_context("index", index) - self.write_context("item", item) + loop = self.workflow_manage.context.get("loop") + if loop is None: + self.write_context("loop", {}) + parameters = self.workflow_manage.get_parameters() + if parameters is not None: + index = parameters.get("index", 0) + item = parameters.get("item", 0) + self.write_context("index", index) + self.write_context("item", item) + else: + self.write_context("index", 0) + self.write_context("item", 0) def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs): details = super().get_details(index, position, old_details, **kwargs) diff --git a/apps/application/workflow/workflow_manage.py b/apps/application/workflow/workflow_manage.py index 9f150bb2eab..11afd733472 100644 --- a/apps/application/workflow/workflow_manage.py +++ b/apps/application/workflow/workflow_manage.py @@ -246,27 +246,16 @@ def get_reference_field(self, node_id, fields): 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 - + def from_context(cls, get_context, workflow, parameters, workflow_type, call_back, get_start_node): + """ + 恢复 WorkflowManage:调用 get_context() 拿到历史 context 并用它重建实例。 + context 从何而来(DB、缓存或其它)由调用方通过 get_context 决定,引擎不关心其业务来源; + get_context 返回空或抛异常则返回 None,调用方可据此回退为全新执行。 + """ 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 实例 + context = get_context() + if not context: + return None instance = cls( workflow=workflow, parameters=parameters, @@ -274,12 +263,10 @@ def from_context(cls, chat_record_id, workflow, parameters, workflow_type, call_ call_back=call_back, get_start_node=get_start_node, ) - # 恢复全局 context - instance.context = context_data - + instance.context = context return instance - except Exception as e: + except Exception: import traceback traceback.print_exc() diff --git a/apps/chat/serializers/chat.py b/apps/chat/serializers/chat.py index 31e5e49ab90..7cc33297c83 100644 --- a/apps/chat/serializers/chat.py +++ b/apps/chat/serializers/chat.py @@ -36,7 +36,7 @@ ) from application.serializers.application import ApplicationOperateSerializer from application.serializers.application_chat import ChatCountSerializer -from application.serializers.common import resolve_chat_user, resolve_chat_user_group +from application.serializers.common import load_debug_workflow_context, resolve_chat_user, resolve_chat_user_group from chat.serializers.chat_history import ChatHistory from application.workflow.common import WorkflowType, new_instance from application.workflow.message.aggregator import AggregationManager @@ -356,7 +356,7 @@ def get_start_node_fn(wf, wm): # Form 提交(有 position 和 chat_record_id):从历史 context 恢复 if position and chat_record_id: work_flow_manage = WorkflowManage.from_context( - chat_record_id=chat_record_id, + get_context=lambda: load_debug_workflow_context(chat_record_id), workflow=workflow, parameters=parameters, workflow_type=WorkflowType.APPLICATION,