fix: The form carried in the loop node cannot retrieve historical data

Author: shaohuzhang1

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 / 类型解析出对话用户信息。

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):

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

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,

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]

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)

apps/application/workflow/workflow_manage.py

@@ -246,40 +246,27 @@ 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,
                 workflow_type=workflow_type,
                 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()

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,