fix(workflow): sanitize condition expression building and cache assigner node inputs

- Sanitize condition expression construction in graph_builder.py using json.dumps to prevent potential injection vulnerabilities.
- Cache input data prior to assigner node execution to ensure variable values are correctly captured before processing.

api/app/core/workflow/engine/graph_builder.py

@@ -2,6 +2,7 @@
 # Author: Eternity
 # @Email: 1533512157@qq.com
 # @Time : 2026/2/10 13:33
+import json
 import logging
 import re
 import uuid
@@ -141,9 +142,10 @@ class GraphBuilder:
 
         for node_info in source_nodes:
             if self.get_node_type(node_info["id"]) in BRANCH_NODES:
-                branch_nodes.append(
+                if node_info.get("branch") is not None:
-                    (node_info["id"], node_info["branch"])
+                    branch_nodes.append(
-                )
+                        (node_info["id"], node_info["branch"])
+                    )
             else:
                 if self.get_node_type(node_info["id"]) in (NodeType.END, NodeType.OUTPUT):
                     output_nodes.append(node_info["id"])
@@ -314,9 +316,12 @@ class GraphBuilder:
                 for idx in range(len(related_edge)):
                     # Generate a condition expression for each edge
                     # Used later to determine which branch to take based on the node's output
-                    # Assumes node output `node.<node_id>.output` matches the edge's label
+                    # For LLM nodes, use branch_signal field for routing (output is dynamic text)
-                    # For example, if node.123.output == 'CASE1', take the branch labeled 'CASE1'
+                    # For other branch nodes (e.g. HTTP), use output field
-                    related_edge[idx]['condition'] = f"node['{node_id}']['output'] == '{related_edge[idx]['label']}'"
+                    route_field = "branch_signal" if node_type == NodeType.LLM else "output"
+                    related_edge[idx]['condition'] = (
+                        f"node[{json.dumps(node_id)}][{json.dumps(route_field)}] == {json.dumps(related_edge[idx]['label'])}"
+                    )
 
             if node_instance:
                 # Wrap node's run method to avoid closure issues

api/app/core/workflow/nodes/assigner/node.py

@@ -18,10 +18,17 @@ class AssignerNode(BaseNode):
         super().__init__(node_config, workflow_config, down_stream_nodes)
         self.variable_updater = True
         self.typed_config: AssignerNodeConfig | None = None
+        self._input_data: dict[str, Any] | None = None
 
     def _output_types(self) -> dict[str, VariableType]:
         return {}
 
+    def _extract_input(self, state: WorkflowState, variable_pool: VariablePool) -> dict[str, Any]:
+        """提取节点输入，如果有缓存的执行前数据则使用缓存"""
+        if self._input_data is not None:
+            return self._input_data
+        return {"config": self._resolve_config(self.config, variable_pool)}
+
     async def execute(self, state: WorkflowState, variable_pool: VariablePool) -> Any:
         """
         Execute the assignment operation defined by this node.
@@ -34,6 +41,9 @@ class AssignerNode(BaseNode):
         Returns:
             None or the result of the assignment operation.
         """
+        # 在执行前提取并缓存输入数据（捕获执行前的变量值）
+        self._input_data = {"config": self._resolve_config(self.config, variable_pool)}
+        
         # Initialize a variable pool for accessing conversation, node, and system variables
         self.typed_config = AssignerNodeConfig(**self.config)
         logger.info(f"节点 {self.node_id} 开始执行")

api/app/core/workflow/nodes/base_node.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import re
 import time
 import uuid
 from abc import ABC, abstractmethod
@@ -22,6 +23,9 @@ from app.services.multimodal_service import MultimodalService
 
 logger = logging.getLogger(__name__)
 
+# 匹配模板变量 {{xxx}} 的正则
+_TEMPLATE_PATTERN = re.compile(r"\{\{.*?\}\}")
+
 
 class NodeExecutionError(Exception):
     """节点执行失败异常。
@@ -503,10 +507,29 @@ class BaseNode(ABC):
             variable_pool: The variable pool used for reading and writing variables.
 
         Returns:
-            A dictionary containing the node's input data.
+            A dictionary containing the node's input data with all template
+            variables resolved to their actual runtime values.
         """
-        # Default implementation returns the node configuration
+        return {"config": self._resolve_config(self.config, variable_pool)}
-        return {"config": self.config}
+
+    @staticmethod
+    def _resolve_config(config: Any, variable_pool: VariablePool) -> Any:
+        """递归解析 config 中的模板变量，将 {{xxx}} 替换为实际值。
+
+        Args:
+            config: 节点的原始配置（可能包含模板变量）。
+            variable_pool: 变量池，用于解析模板变量。
+
+        Returns:
+            解析后的配置，所有字符串中的 {{变量}} 已被替换为真实值。
+        """
+        if isinstance(config, str) and _TEMPLATE_PATTERN.search(config):
+            return BaseNode._render_template(config, variable_pool, strict=False)
+        elif isinstance(config, dict):
+            return {k: BaseNode._resolve_config(v, variable_pool) for k, v in config.items()}
+        elif isinstance(config, list):
+            return [BaseNode._resolve_config(item, variable_pool) for item in config]
+        return config
 
     def _extract_output(self, business_result: Any) -> Any:
         """Extracts the actual output from the business result.

api/app/core/workflow/nodes/document_extractor/node.py

@@ -121,7 +121,10 @@ class DocExtractorNode(BaseNode):
         return business_result
 
     def _extract_input(self, state: WorkflowState, variable_pool: VariablePool) -> dict[str, Any]:
-        return {"file_selector": self.config.get("file_selector")}
+        file_selector = self.config.get("file_selector", "")
+        # 将变量选择器（如 sys.files）解析为实际值
+        resolved = self.get_variable(file_selector, variable_pool, strict=False, default=file_selector)
+        return {"file_selector": resolved}
 
     async def execute(self, state: WorkflowState, variable_pool: VariablePool) -> Any:
         config = DocExtractorNodeConfig(**self.config)

api/app/core/workflow/nodes/enums.py

@@ -31,7 +31,7 @@ class NodeType(StrEnum):
     NOTES = "notes"
 
 
-BRANCH_NODES = frozenset({NodeType.IF_ELSE, NodeType.HTTP_REQUEST, NodeType.QUESTION_CLASSIFIER})
+BRANCH_NODES = frozenset({NodeType.IF_ELSE, NodeType.HTTP_REQUEST, NodeType.QUESTION_CLASSIFIER, NodeType.LLM})
 
 
 class ComparisonOperator(StrEnum):

api/app/core/workflow/nodes/llm/config.py

@@ -6,6 +6,7 @@ import uuid
 from pydantic import BaseModel, Field, field_validator
 
 from app.core.workflow.nodes.base_config import BaseNodeConfig, VariableDefinition
+from app.core.workflow.nodes.enums import HttpErrorHandle
 from app.core.workflow.variable.base_variable import VariableType
 
 
@@ -49,6 +50,20 @@ class MemoryWindowSetting(BaseModel):
     )
 
 
+class LLMErrorHandleConfig(BaseModel):
+    """LLM 异常处理配置"""
+
+    method: HttpErrorHandle = Field(
+        default=HttpErrorHandle.NONE,
+        description="异常处理策略：'none' 抛出异常, 'default' 返回默认值, 'branch' 走异常分支",
+    )
+
+    output: str = Field(
+        default="",
+        description="LLM 异常时返回的默认输出文本（method=default 时生效）",
+    )
+
+
 class LLMNodeConfig(BaseNodeConfig):
     """LLM 节点配置
     
@@ -152,6 +167,11 @@ class LLMNodeConfig(BaseNodeConfig):
         description="输出变量定义（自动生成，通常不需要修改）"
     )
 
+    error_handle: LLMErrorHandleConfig = Field(
+        default_factory=LLMErrorHandleConfig,
+        description="LLM 异常处理配置",
+    )
+
     @field_validator("messages", "prompt")
     @classmethod
     def validate_input_mode(cls, v):

api/app/core/workflow/nodes/llm/node.py

@@ -15,6 +15,7 @@ from app.core.models import RedBearLLM, RedBearModelConfig
 from app.core.workflow.engine.state_manager import WorkflowState
 from app.core.workflow.engine.variable_pool import VariablePool
 from app.core.workflow.nodes.base_node import BaseNode
+from app.core.workflow.nodes.enums import HttpErrorHandle
 from app.core.workflow.nodes.llm.config import LLMNodeConfig
 from app.core.workflow.variable.base_variable import VariableType
 from app.db import get_db_context
@@ -76,7 +77,7 @@ class LLMNode(BaseNode):
         self.messages = []
 
     def _output_types(self) -> dict[str, VariableType]:
-        return {"output": VariableType.STRING}
+        return {"output": VariableType.STRING, "branch_signal": VariableType.STRING}
 
     def _render_context(self, message: str, variable_pool: VariablePool):
         context = f"<context>{self._render_template(self.typed_config.context, variable_pool)}</context>"
@@ -239,7 +240,7 @@ class LLMNode(BaseNode):
 
         return llm
 
-    async def execute(self, state: WorkflowState, variable_pool: VariablePool) -> AIMessage:
+    async def execute(self, state: WorkflowState, variable_pool: VariablePool):
         """非流式执行 LLM 调用
         
         Args:
@@ -247,28 +248,36 @@ class LLMNode(BaseNode):
             variable_pool: 变量池
         
         Returns:
-            LLM 响应消息
+            dict: {"llm_result": AIMessage, "branch_signal": "SUCCESS"} on success,
+                  {"llm_result": None, "branch_signal": "ERROR"} on branch error
         """
-        # self.typed_config = LLMNodeConfig(**self.config)
+        try:
-        llm = await self._prepare_llm(state, variable_pool, False)
+            # self.typed_config = LLMNodeConfig(**self.config)
+            llm = await self._prepare_llm(state, variable_pool, False)
 
-        logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（非流式）")
+            logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（非流式）")
 
-        # 调用 LLM（支持字符串或消息列表）
+            # 调用 LLM（支持字符串或消息列表）
-        response = await llm.ainvoke(self.messages)
+            response = await llm.ainvoke(self.messages)
-        # 提取内容
+            # 提取内容
-        if hasattr(response, 'content'):
+            if hasattr(response, 'content'):
-            content = self.process_model_output(response.content)
+                content = self.process_model_output(response.content)
-        else:
+            else:
-            content = str(response)
+                content = str(response)
 
-        logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(content)}")
+            logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(content)}")
 
-        # 返回 AIMessage（包含响应元数据）
+            # 返回 AIMessage（包含响应元数据）
-        return AIMessage(content=content, response_metadata={
+            return {
-            **response.response_metadata,
+                "llm_result": AIMessage(content=content, response_metadata={
-            "token_usage": getattr(response, 'usage_metadata', None) or response.response_metadata.get('token_usage')
+                    **response.response_metadata,
-        })
+                    "token_usage": getattr(response, 'usage_metadata', None) or response.response_metadata.get('token_usage')
+                }),
+                "branch_signal": "SUCCESS",
+            }
+        except Exception as e:
+            logger.error(f"节点 {self.node_id} LLM 调用失败: {e}")
+            return self._handle_llm_error(e)
 
     def _extract_input(self, state: WorkflowState, variable_pool: VariablePool) -> dict[str, Any]:
         """提取输入数据（用于记录）"""
@@ -286,16 +295,36 @@ class LLMNode(BaseNode):
             }
         }
 
-    def _extract_output(self, business_result: Any) -> str:
+    def _extract_output(self, business_result: Any) -> dict:
-        """从 AIMessage 中提取文本内容"""
+        """从业务结果中提取输出变量
+        
+        支持新旧两种格式：
+        - 新格式：{"llm_result": AIMessage, "branch_signal": "SUCCESS"}
+        - 旧格式：AIMessage（向后兼容）
+        """
+        if isinstance(business_result, dict) and "branch_signal" in business_result:
+            llm_result = business_result.get("llm_result")
+            if isinstance(llm_result, AIMessage):
+                return {
+                    "output": llm_result.content,
+                    "branch_signal": business_result["branch_signal"],
+                }
+            return {
+                "output": str(llm_result) if llm_result else "",
+                "branch_signal": business_result["branch_signal"],
+            }
+        # 旧格式向后兼容
         if isinstance(business_result, AIMessage):
-            return business_result.content
+            return {"output": business_result.content, "branch_signal": "SUCCESS"}
-        return str(business_result)
+        return {"output": str(business_result), "branch_signal": "SUCCESS"}
 
     def _extract_token_usage(self, business_result: Any) -> dict[str, int] | None:
-        """从 AIMessage 中提取 token 使用情况"""
+        """从业务结果中提取 token 使用情况"""
-        if isinstance(business_result, AIMessage) and hasattr(business_result, 'response_metadata'):
+        llm_result = business_result
-            usage = business_result.response_metadata.get('token_usage')
+        if isinstance(business_result, dict):
+            llm_result = business_result.get("llm_result", business_result)
+        if isinstance(llm_result, AIMessage) and hasattr(llm_result, 'response_metadata'):
+            usage = llm_result.response_metadata.get('token_usage')
             if usage:
                 return {
                     "prompt_tokens": usage.get('input_tokens', 0),
@@ -304,6 +333,44 @@ class LLMNode(BaseNode):
                 }
         return None
 
+    def _handle_llm_error(self, error: Exception) -> dict:
+        """处理 LLM 调用异常，根据 error_handle 配置决定行为
+        
+        Args:
+            error: LLM 调用中捕获的异常
+        
+        Returns:
+            dict: {"llm_result": None, "branch_signal": "ERROR"} for branch mode,
+                  or default output for default mode
+        
+        Raises:
+            原异常（当 error_handle.method 为 NONE 时）
+        """
+        if self.typed_config is None:
+            raise error
+
+        match self.typed_config.error_handle.method:
+            case HttpErrorHandle.NONE:
+                raise error
+            case HttpErrorHandle.DEFAULT:
+                logger.warning(
+                    f"节点 {self.node_id}: LLM 调用失败，返回默认输出"
+                )
+                default_output = self.typed_config.error_handle.output or ""
+                return {
+                    "llm_result": AIMessage(content=default_output, response_metadata={}),
+                    "branch_signal": "SUCCESS",
+                }
+            case HttpErrorHandle.BRANCH:
+                logger.warning(
+                    f"节点 {self.node_id}: LLM 调用失败，切换到异常处理分支"
+                )
+                return {
+                    "llm_result": None,
+                    "branch_signal": "ERROR",
+                }
+        raise error
+
     async def execute_stream(self, state: WorkflowState, variable_pool: VariablePool):
         """流式执行 LLM 调用
         
@@ -316,54 +383,58 @@ class LLMNode(BaseNode):
         """
         self.typed_config = LLMNodeConfig(**self.config)
 
-        llm = await self._prepare_llm(state, variable_pool, True)
+        try:
+            llm = await self._prepare_llm(state, variable_pool, True)
 
-        logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（流式）")
+            logger.info(f"节点 {self.node_id} 开始执行 LLM 调用（流式）")
-        # logger.debug(f"LLM 配置: streaming={getattr(llm._model, 'streaming', 'unknown')}")
 
-        # 累积完整响应
+            # 累积完整响应
-        full_response = ""
+            full_response = ""
-        chunk_count = 0
+            chunk_count = 0
 
-        # 调用 LLM（流式，支持字符串或消息列表）
+            # 调用 LLM（流式，支持字符串或消息列表）
-        last_meta_data = {}
+            last_meta_data = {}
-        last_usage_metadata = {}
+            last_usage_metadata = {}
-        async for chunk in llm.astream(self.messages):
+            async for chunk in llm.astream(self.messages):
-            if hasattr(chunk, 'content'):
+                if hasattr(chunk, 'content'):
-                content = self.process_model_output(chunk.content)
+                    content = self.process_model_output(chunk.content)
-            else:
+                else:
-                content = str(chunk)
+                    content = str(chunk)
-            if hasattr(chunk, 'response_metadata') and chunk.response_metadata:
+                if hasattr(chunk, 'response_metadata') and chunk.response_metadata:
-                last_meta_data = chunk.response_metadata
+                    last_meta_data = chunk.response_metadata
-            if hasattr(chunk, 'usage_metadata') and chunk.usage_metadata:
+                if hasattr(chunk, 'usage_metadata') and chunk.usage_metadata:
-                last_usage_metadata = chunk.usage_metadata
+                    last_usage_metadata = chunk.usage_metadata
 
-            # 只有当内容不为空时才处理
+                # 只有当内容不为空时才处理
-            if content:
+                if content:
-                full_response += content
+                    full_response += content
-                chunk_count += 1
+                    chunk_count += 1
 
-                # 流式返回每个文本片段
+                    # 流式返回每个文本片段
-                yield {
+                    yield {
-                    "__final__": False,
+                        "__final__": False,
-                    "chunk": content
+                        "chunk": content
-                }
+                    }
 
-        yield {
+            yield {
-            "__final__": False,
+                "__final__": False,
-            "chunk": "",
+                "chunk": "",
-            "done": True
+                "done": True
-        }
-        logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(full_response)}, 总 chunks: {chunk_count}")
-
-        # 构建完整的 AIMessage（包含元数据）
-        final_message = AIMessage(
-            content=full_response,
-            response_metadata={
-                **last_meta_data,
-                "token_usage": last_usage_metadata or last_meta_data.get('token_usage')
             }
-        )
+            logger.info(f"节点 {self.node_id} LLM 调用完成，输出长度: {len(full_response)}, 总 chunks: {chunk_count}")
 
-        # yield 完成标记
+            # 构建完整的 AIMessage（包含元数据）
-        yield {"__final__": True, "result": final_message}
+            final_message = AIMessage(
+                content=full_response,
+                response_metadata={
+                    **last_meta_data,
+                    "token_usage": last_usage_metadata or last_meta_data.get('token_usage')
+                }
+            )
+
+            # yield 完成标记
+            yield {"__final__": True, "result": {"llm_result": final_message, "branch_signal": "SUCCESS"}}
+        except Exception as e:
+            logger.error(f"节点 {self.node_id} LLM 流式调用失败: {e}")
+            error_result = self._handle_llm_error(e)
+            yield {"__final__": True, "result": error_result}

api/app/services/app_dsl_service.py

@@ -102,6 +102,11 @@ class AppDslService:
                     {**r, "_ref": self._agent_ref(r.get("target_agent_id"))} for r in (cfg["routing_rules"] or [])
                 ]
             return enriched
+        if app_type == AppType.WORKFLOW:
+            enriched = {**cfg}
+            if "nodes" in cfg:
+                enriched["nodes"] = self._enrich_workflow_nodes(cfg["nodes"])
+            return enriched
         return cfg
 
     def _export_draft(self, app: App, meta: dict, app_meta: dict) -> tuple[str, str]:
@@ -110,7 +115,7 @@ class AppDslService:
             config_data = {
                 "variables": config.variables if config else [],
                 "edges": config.edges if config else [],
-                "nodes": config.nodes if config else [],
+                "nodes": self._enrich_workflow_nodes(config.nodes) if config else [],
                 "features": config.features if config else {},
                 "execution_config": config.execution_config if config else {},
                 "triggers": config.triggers if config else [],
@@ -190,6 +195,23 @@ class AppDslService:
     def _enrich_tools(self, tools: list) -> list:
         return [{**t, "_ref": self._tool_ref(t.get("tool_id"))} for t in (tools or [])]
 
+    def _enrich_workflow_nodes(self, nodes: list) -> list:
+        """enrich 工作流节点中的模型引用，添加 name、provider、type 信息"""
+        from app.core.workflow.nodes.enums import NodeType
+        enriched_nodes = []
+        for node in (nodes or []):
+            node_type = node.get("type")
+            config = dict(node.get("config") or {})
+            
+            if node_type in (NodeType.LLM.value, NodeType.QUESTION_CLASSIFIER.value, NodeType.PARAMETER_EXTRACTOR.value):
+                model_id = config.get("model_id")
+                if model_id:
+                    config["model_ref"] = self._model_ref(model_id)
+                    del config["model_id"]
+            
+            enriched_nodes.append({**node, "config": config})
+        return enriched_nodes
+
     def _skill_ref(self, skill_id) -> Optional[dict]:
         if not skill_id:
             return None
@@ -620,16 +642,16 @@ class AppDslService:
                         warnings.append(f"[{node_label}] 知识库 '{kb_id}' 未匹配，已移除，请导入后手动配置")
                 config["knowledge_bases"] = resolved_kbs
             elif node_type in (NodeType.LLM.value, NodeType.QUESTION_CLASSIFIER.value, NodeType.PARAMETER_EXTRACTOR.value):
-                model_ref = config.get("model_id")
+                model_ref = config.get("model_ref") or config.get("model_id")
                 if model_ref:
                     ref_dict = None
                     if isinstance(model_ref, dict):
-                        ref_id = model_ref.get("id")
+                        ref_dict = {
-                        ref_name = model_ref.get("name")
+                            "id": model_ref.get("id"),
-                        if ref_id:
+                            "name": model_ref.get("name"),
-                            ref_dict = {"id": ref_id}
+                            "provider": model_ref.get("provider"),
-                        elif ref_name is not None:
+                            "type": model_ref.get("type")
-                            ref_dict = {"name": ref_name, "provider": model_ref.get("provider"), "type": model_ref.get("type")}
+                        }
                     elif isinstance(model_ref, str):
                         try:
                             uuid.UUID(model_ref)
@@ -640,12 +662,18 @@ class AppDslService:
                         resolved_model_id = self._resolve_model(ref_dict, tenant_id, warnings)
                         if resolved_model_id:
                             config["model_id"] = resolved_model_id
+                            if "model_ref" in config:
+                                del config["model_ref"]
                         else:
                             warnings.append(f"[{node_label}] 模型未匹配，已置空，请导入后手动配置")
                             config["model_id"] = None
+                            if "model_ref" in config:
+                                del config["model_ref"]
                     else:
                         warnings.append(f"[{node_label}] 模型未匹配，已置空，请导入后手动配置")
                         config["model_id"] = None
+                        if "model_ref" in config:
+                            del config["model_ref"]
             resolved_nodes.append({**node, "config": config})
         return resolved_nodes