diff --git a/api/app/core/workflow/adapters/base_adapter.py b/api/app/core/workflow/adapters/base_adapter.py index 49321b89..2e24d085 100644 --- a/api/app/core/workflow/adapters/base_adapter.py +++ b/api/app/core/workflow/adapters/base_adapter.py @@ -9,7 +9,7 @@ from typing import Any from pydantic import BaseModel, Field -from app.core.workflow.adapters.errors import ExceptionDefineition +from app.core.workflow.adapters.errors import ExceptionDefinition from app.schemas.workflow_schema import ( EdgeDefinition, NodeDefinition, @@ -40,8 +40,8 @@ class WorkflowParserResult(BaseModel): edges: list[EdgeDefinition] = Field(default_factory=list) nodes: list[NodeDefinition] = Field(default_factory=list) variables: list[VariableDefinition] = Field(default_factory=list) - warnings: list[ExceptionDefineition] = Field(default_factory=list) - errors: list[ExceptionDefineition] = Field(default_factory=list) + warnings: list[ExceptionDefinition] = Field(default_factory=list) + errors: list[ExceptionDefinition] = Field(default_factory=list) class WorkflowImportResult(BaseModel): @@ -51,8 +51,8 @@ class WorkflowImportResult(BaseModel): edges: list[EdgeDefinition] = Field(default_factory=list) nodes: list[NodeDefinition] = Field(default_factory=list) variables: list[VariableDefinition] = Field(default_factory=list) - warnings: list[ExceptionDefineition] = Field(default_factory=list) - errors: list[ExceptionDefineition] = Field(default_factory=list) + warnings: list[ExceptionDefinition] = Field(default_factory=list) + errors: list[ExceptionDefinition] = Field(default_factory=list) class BasePlatformAdapter(ABC): diff --git a/api/app/core/workflow/adapters/dify/converter.py b/api/app/core/workflow/adapters/dify/converter.py index 467beb07..4fa9508b 100644 --- a/api/app/core/workflow/adapters/dify/converter.py +++ b/api/app/core/workflow/adapters/dify/converter.py @@ -9,9 +9,9 @@ from urllib.parse import quote from app.core.workflow.adapters.base_converter import BaseConverter from app.core.workflow.adapters.errors import ( - UnsupportVariableType, - UnknowModelWarning, - ExceptionDefineition, + UnsupportedVariableType, + UnknownModelWarning, + ExceptionDefinition, ExceptionType ) from app.core.workflow.nodes.assigner.config import AssignmentItem @@ -54,7 +54,7 @@ from app.core.workflow.nodes.http_request.config import ( HttpFormData, HttpTimeOutConfig, HttpRetryConfig, - HttpErrorDefaultTamplete, + HttpErrorDefaultTemplate, HttpErrorHandleConfig ) from app.core.workflow.nodes.if_else.config import ConditionDetail, ConditionBranchConfig @@ -108,7 +108,7 @@ class DifyConverter(BaseConverter): try: return config.model_validate(value) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.CONFIG, node_id=node_id, node_name=node_name, @@ -138,7 +138,7 @@ class DifyConverter(BaseConverter): var_selector = mapping.get(var_selector, var_selector) return var_selector - def _process_list_variable_litearl(self, variable_selector: list) -> str | None: + def _process_list_variable_literal(self, variable_selector: list) -> str | None: if not self.process_var_selector(".".join(variable_selector)): return None return "{{" + self.process_var_selector(".".join(variable_selector)) + "}}" @@ -269,7 +269,7 @@ class DifyConverter(BaseConverter): var_type = self.variable_type_map(var["type"]) if not var_type: self.errors.append( - UnsupportVariableType( + UnsupportedVariableType( scope=node["id"], name=var["variable"], var_type=var["type"], @@ -281,7 +281,7 @@ class DifyConverter(BaseConverter): if var_type in ["file", "array[file]"]: self.errors.append( - ExceptionDefineition( + ExceptionDefinition( type=ExceptionType.VARIABLE, node_id=node["id"], node_name=node_data["title"], @@ -311,7 +311,7 @@ class DifyConverter(BaseConverter): def convert_question_classifier_node_config(self, node: dict) -> dict: node_data = node["data"] self.warnings.append( - UnknowModelWarning( + UnknownModelWarning( node_id=node["id"], node_name=node_data["title"], model_name=node_data["model"].get("name") @@ -327,7 +327,7 @@ class DifyConverter(BaseConverter): ) result = QuestionClassifierNodeConfig.model_construct( - input_variable=self._process_list_variable_litearl(node_data.get("query_variable_selector")), + input_variable=self._process_list_variable_literal(node_data.get("query_variable_selector")), user_supplement_prompt=self.trans_variable_format(node_data.get("instructions", "")), categories=categories, ).model_dump() @@ -337,13 +337,13 @@ class DifyConverter(BaseConverter): def convert_llm_node_config(self, node: dict) -> dict: node_data = node["data"] self.warnings.append( - UnknowModelWarning( + UnknownModelWarning( node_id=node["id"], node_name=node_data["title"], model_name=node_data["model"].get("name") ) ) - context = self._process_list_variable_litearl(node_data["context"]["variable_selector"]) + context = self._process_list_variable_literal(node_data["context"]["variable_selector"]) memory = MemoryWindowSetting( enable=bool(node_data.get("memory")), enable_window=bool(node_data.get("memory", {}).get("window", {}).get("enabled", False)), @@ -367,7 +367,7 @@ class DifyConverter(BaseConverter): ) ) vision = node_data["vision"]["enabled"] - vision_input = self._process_list_variable_litearl( + vision_input = self._process_list_variable_literal( node_data["vision"]["configs"]["variable_selector"] ) if vision else None result = LLMNodeConfig.model_construct( @@ -433,7 +433,7 @@ class DifyConverter(BaseConverter): conditions.append( LoopConditionDetail.model_construct( operator=self.convert_compare_operator(condition["comparison_operator"]), - left=self._process_list_variable_litearl(condition["variable_selector"]), + left=self._process_list_variable_literal(condition["variable_selector"]), right=self.trans_variable_format( right_value ) if isinstance(right_value, str) and self.is_variable(right_value) else self.convert_variable_type( @@ -453,7 +453,7 @@ class DifyConverter(BaseConverter): right_input_type = variable["value_type"] right_value_type = self.variable_type_map(variable["var_type"]) if right_input_type == ValueInputType.VARIABLE: - right_value = self._process_list_variable_litearl(variable.get("value", "")) + right_value = self._process_list_variable_literal(variable.get("value", "")) else: right_value = self.convert_variable_type(right_value_type, variable.get("value", "")) loop_variables.append( @@ -475,10 +475,10 @@ class DifyConverter(BaseConverter): def convert_iteration_node_config(self, node: dict) -> dict: node_data = node["data"] result = IterationNodeConfig.model_construct( - input=self._process_list_variable_litearl(node_data["iterator_selector"]), + input=self._process_list_variable_literal(node_data["iterator_selector"]), parallel=node_data["is_parallel"], parallel_count=node_data["parallel_nums"], - output=self._process_list_variable_litearl(node_data["output_selector"]), + output=self._process_list_variable_literal(node_data["output_selector"]), output_type=self.variable_type_map(node_data.get("output_type")), flatten=node_data["flatten_output"], ).model_dump() @@ -494,8 +494,8 @@ class DifyConverter(BaseConverter): continue assignments.append( AssignmentItem( - variable_selector=self._process_list_variable_litearl(assignment["variable_selector"]), - value=self._process_list_variable_litearl( + variable_selector=self._process_list_variable_literal(assignment["variable_selector"]), + value=self._process_list_variable_literal( assignment["value"] ) if assignment["input_type"] == ValueInputType.VARIABLE else assignment["value"], operation=self.convert_assignment_operator(assignment["operation"]) @@ -514,7 +514,7 @@ class DifyConverter(BaseConverter): input_variables.append( InputVariable.model_construct( name=input_variable["variable"], - variable=self._process_list_variable_litearl(input_variable["value_selector"]), + variable=self._process_list_variable_literal(input_variable["value_selector"]), ) ) @@ -570,7 +570,7 @@ class DifyConverter(BaseConverter): else: if node_data["body"]["data"]: body_content = (node_data["body"]["data"][0].get("value") or - self._process_list_variable_litearl(node_data["body"]["data"][0].get("file"))) + self._process_list_variable_literal(node_data["body"]["data"][0].get("file"))) else: body_content = "" @@ -585,7 +585,7 @@ class DifyConverter(BaseConverter): self.trans_variable_format(key_value[0]) ] = self.trans_variable_format(key_value[1]) else: - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( type=ExceptionType.CONFIG, node_id=node["id"], node_name=node_data["title"], @@ -603,7 +603,7 @@ class DifyConverter(BaseConverter): self.trans_variable_format(key_value[0]) ] = self.trans_variable_format(key_value[1]) else: - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( type=ExceptionType.CONFIG, node_id=node["id"], node_name=node_data["title"], @@ -625,7 +625,7 @@ class DifyConverter(BaseConverter): default_header = var["value"] elif var["key"] == "status_code": default_status_code = var["value"] - default_value = HttpErrorDefaultTamplete( + default_value = HttpErrorDefaultTemplate( body=default_body, headers=default_header, status_code=default_status_code, @@ -668,7 +668,7 @@ class DifyConverter(BaseConverter): for variable in node_data["variables"]: mapping.append(VariablesMappingConfig.model_construct( name=variable["variable"], - value=self._process_list_variable_litearl(variable["value_selector"]) + value=self._process_list_variable_literal(variable["value_selector"]) )) result = JinjaRenderNodeConfig.model_construct( template=node_data["template"], @@ -679,14 +679,14 @@ class DifyConverter(BaseConverter): def convert_knowledge_node_config(self, node: dict) -> dict: node_data = node["data"] - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( node_id=node["id"], node_name=node_data["title"], type=ExceptionType.CONFIG, detail=f"Please reconfigure the Knowledge Retrieval node.", )) result = KnowledgeRetrievalNodeConfig.model_construct( - query=self._process_list_variable_litearl(node_data["query_variable_selector"]), + query=self._process_list_variable_literal(node_data["query_variable_selector"]), ).model_dump() self.config_validate(node["id"], node["data"]["title"], KnowledgeRetrievalNodeConfig, result) @@ -695,7 +695,7 @@ class DifyConverter(BaseConverter): def convert_parameter_extractor_node_config(self, node: dict) -> dict: node_data = node["data"] self.warnings.append( - UnknowModelWarning( + UnknownModelWarning( node_id=node["id"], node_name=node_data["title"], model_name=node_data["model"].get("name") @@ -712,7 +712,7 @@ class DifyConverter(BaseConverter): ) ) result = ParameterExtractorNodeConfig.model_construct( - text=self._process_list_variable_litearl(node_data["query"]), + text=self._process_list_variable_literal(node_data["query"]), params=params, prompt=node_data.get("instruction") ).model_dump() @@ -727,14 +727,14 @@ class DifyConverter(BaseConverter): group_type = {} if not advanced_settings or not advanced_settings["group_enabled"]: group_variables = [ - self._process_list_variable_litearl(variable) + self._process_list_variable_literal(variable) for variable in node_data["variables"] ] group_type["output"] = node_data["output_type"] else: for group in advanced_settings["groups"]: group_variables[group["group_name"]] = [ - self._process_list_variable_litearl(variable) + self._process_list_variable_literal(variable) for variable in group["variables"] ] group_type[group["group_name"]] = group["output_type"] @@ -751,7 +751,7 @@ class DifyConverter(BaseConverter): def convert_tool_node_config(self, node: dict) -> dict: node_data = node["data"] - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( node_id=node["id"], node_name=node_data["title"], type=ExceptionType.CONFIG, diff --git a/api/app/core/workflow/adapters/dify/dify_adapter.py b/api/app/core/workflow/adapters/dify/dify_adapter.py index 10397ad0..abd95408 100644 --- a/api/app/core/workflow/adapters/dify/dify_adapter.py +++ b/api/app/core/workflow/adapters/dify/dify_adapter.py @@ -12,7 +12,7 @@ from app.core.workflow.adapters.base_adapter import ( WorkflowParserResult ) from app.core.workflow.adapters.dify.converter import DifyConverter -from app.core.workflow.adapters.errors import ExceptionDefineition, ExceptionType +from app.core.workflow.adapters.errors import ExceptionDefinition, ExceptionType from app.core.workflow.nodes.enums import NodeType from app.schemas.workflow_schema import ( NodeDefinition, @@ -85,7 +85,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): if not all(field in self.config for field in require_fields): return False if self.config.get("app", {}).get("mode") == "workflow": - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.PLATFORM, detail="workflow mode is not supported" )) @@ -111,12 +111,12 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): edge = self._convert_edge(edge) if edge: self.edges.append(edge) - # + for variable in self.config.get("workflow").get("conversation_variables"): con_var = self._convert_variable(variable) if variable: self.conv_variables.append(con_var) - # + # for variables in config.get("workflow").get("environment_variables"): # variable = self._convert_variable(variables) # conv_variables.append(variable) @@ -152,7 +152,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): "y": node["position"]["y"] + position["y"] } self.errors.append( - ExceptionDefineition( + ExceptionDefinition( type=ExceptionType.NODE, node_id=node_id, detail="parent cycle node not found" @@ -189,7 +189,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): node_data = node["data"] converter = self.get_node_convert(node_type) if node_type == NodeType.UNKNOWN: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.NODE, node_id=node["id"], node_name=node["data"]["title"], @@ -197,7 +197,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): )) return converter(node) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.NODE, node_id=node["id"], node_name=node["data"]["title"], @@ -207,7 +207,6 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): def _convert_edge(self, edge: dict[str, Any]) -> EdgeDefinition | None: try: - source = edge["source"] target = edge["target"] label = None @@ -230,7 +229,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): label=label, ) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.EDGE, detail=f"convert edge error - {e}", )) @@ -246,7 +245,7 @@ class DifyAdapter(BasePlatformAdapter, DifyConverter): description=variable.get("description") ) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.VARIABLE, name=variable.get("name"), detail=f"convert variable error - {e}", diff --git a/api/app/core/workflow/adapters/errors.py b/api/app/core/workflow/adapters/errors.py index c0340a5e..cb743c68 100644 --- a/api/app/core/workflow/adapters/errors.py +++ b/api/app/core/workflow/adapters/errors.py @@ -18,7 +18,7 @@ class ExceptionType(StrEnum): UNKNOWN = "unknown" -class ExceptionDefineition(BaseModel): +class ExceptionDefinition(BaseModel): type: ExceptionType detail: str @@ -29,7 +29,7 @@ class ExceptionDefineition(BaseModel): name: str | None = None -class UnknowModelWarning(ExceptionDefineition): +class UnknownModelWarning(ExceptionDefinition): type: ExceptionType = ExceptionType.NODE def __init__(self, node_id, node_name, model_name): @@ -40,36 +40,36 @@ class UnknowModelWarning(ExceptionDefineition): ) -class UnknowError(ExceptionDefineition): +class UnknownError(ExceptionDefinition): type: ExceptionType = ExceptionType.UNKNOWN def __init__(self, detail: str, **kwargs): super().__init__(detail=detail, **kwargs) -class UnsupportPlatform(ExceptionDefineition): +class UnsupportedPlatform(ExceptionDefinition): type: ExceptionType = ExceptionType.PLATFORM def __init__(self, platform: str): - super().__init__(detail=f"Unsupport platform {platform}") + super().__init__(detail=f"Unsupported platform {platform}") -class UnsupportVariableType(ExceptionDefineition): +class UnsupportedVariableType(ExceptionDefinition): type: ExceptionType = ExceptionType.VARIABLE def __init__(self, scope, name, var_type: str, **kwargs): - super().__init__(scope=scope, name=name, detail=f"Unsupport variable type：[{var_type}]", **kwargs) + super().__init__(scope=scope, name=name, detail=f"Unsupported variable type: [{var_type}]", **kwargs) -class InvalidConfiguration(ExceptionDefineition): +class InvalidConfiguration(ExceptionDefinition): type: ExceptionType = ExceptionType.CONFIG def __init__(self): super().__init__(detail="Invalid workflow configuration format") -class UnsupportNodeType(ExceptionDefineition): +class UnsupportedNodeType(ExceptionDefinition): type: ExceptionType = ExceptionType.NODE def __init__(self, node_id: str, node_type: str): - super().__init__(node_id=node_id, detail=f"Unsupport node Type {node_type}") + super().__init__(node_id=node_id, detail=f"Unsupported node type {node_type}") diff --git a/api/app/core/workflow/adapters/memory_bear/memory_bear_adapter.py b/api/app/core/workflow/adapters/memory_bear/memory_bear_adapter.py index 3516cb58..a2608a01 100644 --- a/api/app/core/workflow/adapters/memory_bear/memory_bear_adapter.py +++ b/api/app/core/workflow/adapters/memory_bear/memory_bear_adapter.py @@ -11,7 +11,7 @@ from app.core.workflow.adapters.base_adapter import ( BasePlatformAdapter, WorkflowParserResult ) -from app.core.workflow.adapters.errors import ExceptionDefineition, ExceptionType, UnsupportNodeType +from app.core.workflow.adapters.errors import ExceptionDefinition, ExceptionType, UnsupportedNodeType from app.core.workflow.adapters.memory_bear.memory_bear_converter import MemoryBearConverter from app.core.workflow.nodes.enums import NodeType from app.schemas.workflow_schema import ExecutionConfig, NodeDefinition, EdgeDefinition, VariableDefinition @@ -73,7 +73,7 @@ class MemoryBearAdapter(BasePlatformAdapter, MemoryBearConverter): try: node_type = self.map_node_type(node["type"]) if node_type == NodeType.UNKNOWN: - self.errors.append(UnsupportNodeType( + self.errors.append(UnsupportedNodeType( node_id=node_id, node_type=node["type"] )) @@ -85,7 +85,7 @@ class MemoryBearAdapter(BasePlatformAdapter, MemoryBearConverter): return NodeDefinition(**node) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.NODE, node_id=node_id, node_name=node_name, @@ -97,14 +97,14 @@ class MemoryBearAdapter(BasePlatformAdapter, MemoryBearConverter): def _convert_edge(self, edge: dict[str, Any], valid_node_ids: set) -> EdgeDefinition | None: try: if edge.get("source") not in valid_node_ids or edge.get("target") not in valid_node_ids: - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( type=ExceptionType.EDGE, detail=f"edge {edge.get('id')} skipped: source or target node not found" )) return None return EdgeDefinition(**edge) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.EDGE, detail=f"convert edge error - {e}" )) @@ -115,7 +115,7 @@ class MemoryBearAdapter(BasePlatformAdapter, MemoryBearConverter): try: return VariableDefinition(**variable) except Exception as e: - self.warnings.append(ExceptionDefineition( + self.warnings.append(ExceptionDefinition( type=ExceptionType.VARIABLE, name=variable.get("name"), detail=f"convert variable error - {e}" diff --git a/api/app/core/workflow/adapters/memory_bear/memory_bear_converter.py b/api/app/core/workflow/adapters/memory_bear/memory_bear_converter.py index 031c7025..e96e0bf2 100644 --- a/api/app/core/workflow/adapters/memory_bear/memory_bear_converter.py +++ b/api/app/core/workflow/adapters/memory_bear/memory_bear_converter.py @@ -1,6 +1,6 @@ # -*- coding: UTF-8 -*- from app.core.workflow.adapters.base_converter import BaseConverter -from app.core.workflow.adapters.errors import ExceptionDefineition, ExceptionType +from app.core.workflow.adapters.errors import ExceptionDefinition, ExceptionType from app.core.workflow.nodes.base_config import BaseNodeConfig from app.core.workflow.nodes.configs import ( StartNodeConfig, @@ -65,7 +65,7 @@ class MemoryBearConverter(BaseConverter): try: return config_cls.model_validate(value) except Exception as e: - self.errors.append(ExceptionDefineition( + self.errors.append(ExceptionDefinition( type=ExceptionType.CONFIG, node_id=node_id, node_name=node_name, diff --git a/api/app/core/workflow/engine/graph_builder.py b/api/app/core/workflow/engine/graph_builder.py index c5cf3324..29f46765 100644 --- a/api/app/core/workflow/engine/graph_builder.py +++ b/api/app/core/workflow/engine/graph_builder.py @@ -69,11 +69,12 @@ class GraphBuilder: for node in self.nodes if node.get("type") == "end" and node.get("id") in self.reachable_nodes ] + self._reverse_adj: dict[str, list[dict]] = defaultdict(list) + self._adj: dict[str, list[str]] = defaultdict(list) + self._build_reverse_adj() self.add_edges() # EDGES MUST BE ADDED AFTER NODES ARE ADDED. - self._reverse_adj: dict[str, list[dict]] = defaultdict(list) - self._build_reverse_adj() self._analyze_end_node_output() @property @@ -115,6 +116,7 @@ class GraphBuilder: self._reverse_adj[edge.get("target")].append({ "id": edge["source"], "branch": edge.get("label") }) + self._adj[edge.get("source")].append(edge["target"]) def _find_upstream_activation_dep( self, @@ -413,11 +415,12 @@ class GraphBuilder: # Add conditional edges for source_node, branches in conditional_edges.items(): def make_router(src, branch_list): - """reate a router function for each source node that routes to a NOP node for later merging.""" + """Create a router function for each source node that routes to a NOP node for later merging.""" def make_branch_node(node_name, targets): def node(s): - # NOTE: NOP NODE MUST NOT MODIFY STATE + # NOTE: NOP NODE USED FOR ROUTING ONLY. + # MUST NOT MUTATE STATE DIRECTLY; ONLY EMIT ACTIVATE SIGNALS. return { "activate": { node_id: s["activate"][node_name] @@ -504,11 +507,9 @@ class GraphBuilder: logger.debug(f"Added waiting edge: {sources} -> {target}") # Connect End nodes to the global END node - for end_node in self.end_nodes: - end_node_id = end_node.get("id") - if end_node_id: - self.graph.add_edge(end_node_id, END) - logger.debug(f"Added edge: {end_node_id} -> END") + for node in self.reachable_nodes: + if not self._adj[node]: + self.graph.add_edge(node, END) return def build(self) -> CompiledStateGraph: diff --git a/api/app/core/workflow/engine/result_builder.py b/api/app/core/workflow/engine/result_builder.py index e5a03c1c..be0c957a 100644 --- a/api/app/core/workflow/engine/result_builder.py +++ b/api/app/core/workflow/engine/result_builder.py @@ -2,6 +2,7 @@ # Author: Eternity # @Email: 1533512157@qq.com # @Time : 2026/2/10 13:33 +from app.core.workflow.engine.runtime_schema import ExecutionContext from app.core.workflow.engine.variable_pool import VariablePool @@ -9,6 +10,7 @@ class WorkflowResultBuilder: def build_final_output( self, result: dict, + execution_context: ExecutionContext, variable_pool: VariablePool, elapsed_time: float, final_output: str, @@ -26,6 +28,8 @@ class WorkflowResultBuilder: - "node_outputs" (dict): Outputs of executed nodes. - "messages" (list): Conversation messages exchanged during execution. - "error" (str, optional): Error message if any node failed. + execution_context (ExecutionContext): The execution context containing metadata like + execution ID, workspace ID, and user ID.) variable_pool (VariablePool): Variable Pool elapsed_time (float): Total execution time in seconds. final_output (Any): The aggregated or final output content of the workflow @@ -48,18 +52,23 @@ class WorkflowResultBuilder: """ node_outputs = result.get("node_outputs", {}) token_usage = self.aggregate_token_usage(node_outputs) - conversation_id = variable_pool.get_value("sys.conversation_id") + conversation_vars = {} + sys_vars = {} + + if variable_pool: + conversation_vars = variable_pool.get_all_conversation_vars() + sys_vars = variable_pool.get_all_system_vars() return { "status": "completed" if success else "failed", "output": final_output, "variables": { - "conv": variable_pool.get_all_conversation_vars(), - "sys": variable_pool.get_all_system_vars() + "conv": conversation_vars, + "sys": sys_vars }, "node_outputs": node_outputs, "messages": result.get("messages", []), - "conversation_id": conversation_id, + "conversation_id": execution_context.conversation_id, "elapsed_time": elapsed_time, "token_usage": token_usage, "error": result.get("error"), diff --git a/api/app/core/workflow/engine/runtime_schema.py b/api/app/core/workflow/engine/runtime_schema.py index 48eafaa9..036ce0e8 100644 --- a/api/app/core/workflow/engine/runtime_schema.py +++ b/api/app/core/workflow/engine/runtime_schema.py @@ -12,6 +12,7 @@ class ExecutionContext(BaseModel): execution_id: str workspace_id: str user_id: str + conversation_id: str memory_storage_type: str user_rag_memory_id: str checkpoint_config: RunnableConfig @@ -22,6 +23,7 @@ class ExecutionContext(BaseModel): execution_id: str, workspace_id: str, user_id: str, + conversation_id: str, memory_storage_type: str, user_rag_memory_id: str ): @@ -29,6 +31,7 @@ class ExecutionContext(BaseModel): execution_id=execution_id, workspace_id=workspace_id, user_id=user_id, + conversation_id=conversation_id, memory_storage_type=memory_storage_type, user_rag_memory_id=user_rag_memory_id, diff --git a/api/app/core/workflow/executor.py b/api/app/core/workflow/executor.py index 6a127e96..1170d66c 100644 --- a/api/app/core/workflow/executor.py +++ b/api/app/core/workflow/executor.py @@ -3,6 +3,7 @@ # @Email: 1533512157@qq.com # @Time : 2026/2/9 13:51 import datetime +import time import logging from typing import Any @@ -82,6 +83,7 @@ class WorkflowExecutor: CompiledStateGraph: The compiled and ready-to-run state graph. """ logger.info(f"Starting workflow graph build: execution_id={self.execution_context.execution_id}") + start_time = time.time() builder = GraphBuilder( self.workflow_config, stream=stream, @@ -96,7 +98,8 @@ class WorkflowExecutor: variable_pool=self.variable_pool, execution_id=self.execution_context.execution_id ) - logger.info(f"Workflow graph build completed: execution_id={self.execution_context.execution_id}") + logger.info(f"Workflow graph build completed: execution_id={self.execution_context.execution_id}, " + f"cost: {time.time() - start_time:.4f}s") return self.graph @@ -134,94 +137,12 @@ class WorkflowExecutor: return event.get("data") return self.result_builder.build_final_output( {"error": "Workflow execution did not end as expected"}, + self.execution_context, self.variable_pool, (datetime.datetime.now() - start).total_seconds(), "", success=False ) - # logger.info(f"Starting workflow execution: execution_id={self.execution_context.execution_id}") - # - # start_time = datetime.datetime.now() - # - # # Execute the workflow - # try: - # # Build the workflow graph - # graph = self.build_graph() - # - # # Initialize the variable pool with input data - # await self.variable_initializer.initialize( - # variable_pool=self.variable_pool, - # input_data=input_data, - # execution_context=self.execution_context - # ) - # initial_state = self.state_manager.create_initial_state( - # workflow_config=self.workflow_config, - # input_data=input_data, - # execution_context=self.execution_context, - # start_node_id=self.start_node_id - # ) - # - # result = await graph.ainvoke(initial_state, config=self.execution_context.checkpoint_config) - # - # # Aggregate output from all End nodes - # full_content = '' - # for end_id in self.stream_coordinator.end_outputs.keys(): - # full_content += self.variable_pool.get_value(f"{end_id}.output", default="", strict=False) - # - # # Append messages for user and assistant - # if input_data.get("files"): - # result["messages"].extend( - # [ - # { - # "role": "user", - # "content": input_data.get("message", '') - # }, - # { - # "role": "user", - # "content": input_data.get("files") - # }, - # { - # "role": "assistant", - # "content": full_content - # } - # ] - # ) - # else: - # result["messages"].extend( - # [ - # { - # "role": "user", - # "content": input_data.get("message", '') - # }, - # { - # "role": "assistant", - # "content": full_content - # } - # ] - # ) - # # Calculate elapsed time - # end_time = datetime.datetime.now() - # elapsed_time = (end_time - start_time).total_seconds() - # - # logger.info( - # f"Workflow execution completed: execution_id={self.execution_context.execution_id}, elapsed_time={elapsed_time:.2f}ms") - # - # return self.result_builder.build_final_output(result, self.variable_pool, elapsed_time, full_content) - # - # except Exception as e: - # end_time = datetime.datetime.now() - # elapsed_time = (end_time - start_time).total_seconds() - # - # logger.error(f"Workflow execution failed: execution_id={self.execution_context.execution_id}, error={e}", - # exc_info=True) - # return { - # "status": "failed", - # "error": str(e), - # "output": None, - # "node_outputs": {}, - # "elapsed_time": elapsed_time, - # "token_usage": None - # } async def execute_stream( self, @@ -255,7 +176,7 @@ class WorkflowExecutor: "data": { "execution_id": self.execution_context.execution_id, "workspace_id": self.execution_context.workspace_id, - "conversation_id": input_data.get("conversation_id"), + "conversation_id": self.execution_context.conversation_id, "timestamp": int(start_time.timestamp() * 1000) } } @@ -376,6 +297,7 @@ class WorkflowExecutor: "event": "workflow_end", "data": self.result_builder.build_final_output( result, + self.execution_context, self.variable_pool, elapsed_time, full_content, @@ -396,6 +318,7 @@ class WorkflowExecutor: "event": "workflow_end", "data": self.result_builder.build_final_output( result, + self.execution_context, self.variable_pool, elapsed_time, full_content, @@ -432,6 +355,7 @@ async def execute_workflow( execution_id=execution_id, workspace_id=workspace_id, user_id=user_id, + conversation_id=input_data.get("conversation_id"), memory_storage_type=memory_storage_type, user_rag_memory_id=user_rag_memory_id ) @@ -471,6 +395,7 @@ async def execute_workflow_stream( workspace_id=workspace_id, user_id=user_id, memory_storage_type=memory_storage_type, + conversation_id=input_data.get("conversation_id"), user_rag_memory_id=user_rag_memory_id ) executor = WorkflowExecutor( diff --git a/api/app/core/workflow/nodes/agent/node.py b/api/app/core/workflow/nodes/agent/node.py index 8959e27c..7b146a9c 100644 --- a/api/app/core/workflow/nodes/agent/node.py +++ b/api/app/core/workflow/nodes/agent/node.py @@ -64,9 +64,7 @@ class AgentNode(BaseNode): if not release: raise ValueError(f"Agent 不存在: {agent_id}") - - return release, message async def execute(self, state: WorkflowState, variable_pool: VariablePool) -> dict[str, Any]: diff --git a/api/app/core/workflow/nodes/base_node.py b/api/app/core/workflow/nodes/base_node.py index 7f2b8aa6..34b7dfa3 100644 --- a/api/app/core/workflow/nodes/base_node.py +++ b/api/app/core/workflow/nodes/base_node.py @@ -315,8 +315,8 @@ class BaseNode(ABC): elapsed_time = (time.time() - start_time) * 1000 - logger.info(f"Node {self.node_id} streaming execution finished, " - f"time elapsed: {elapsed_time:.2f}ms, chunks: {chunk_count}") + logger.debug(f"Node {self.node_id} streaming execution finished, " + f"time elapsed: {elapsed_time:.2f}ms, chunks: {chunk_count}") # Extract processed output (call subclass's _extract_output) extracted_output = self._extract_output(final_result) @@ -644,7 +644,7 @@ class BaseNode(ABC): if content.content_cache.get(f"{provider}_{ModelInfo.is_omni}"): return content.content_cache[f"{provider}_{ModelInfo.is_omni}"] with get_db_read() as db: - multimodel_service = MultimodalService(db, api_config=api_config) + multimodal_service = MultimodalService(db, api_config=api_config) file_obj = FileInput( type=content.type, url=content.url, @@ -653,7 +653,7 @@ class BaseNode(ABC): upload_file_id=uuid.UUID(content.file_id) if content.file_id else None, ) file_obj.set_content(content.get_content()) - message = await multimodel_service.process_files( + message = await multimodal_service.process_files( [file_obj], ) content.set_content(file_obj.get_content()) @@ -661,7 +661,7 @@ class BaseNode(ABC): content.content_cache[f"{provider}_{ModelInfo.is_omni}"] = message return message return None - raise TypeError(f'Unexpect input value type - {type(content)}') + raise TypeError(f'Unexpected input value type - {type(content)}') @staticmethod def process_model_output(content) -> str: diff --git a/api/app/core/workflow/nodes/end/config.py b/api/app/core/workflow/nodes/end/config.py index 5c2a6c2a..02df5091 100644 --- a/api/app/core/workflow/nodes/end/config.py +++ b/api/app/core/workflow/nodes/end/config.py @@ -1,9 +1,7 @@ """End 节点配置""" - from pydantic import Field -from app.core.workflow.nodes.base_config import BaseNodeConfig, VariableDefinition -from app.core.workflow.variable.base_variable import VariableType +from app.core.workflow.nodes.base_config import BaseNodeConfig class EndNodeConfig(BaseNodeConfig): diff --git a/api/app/core/workflow/nodes/end/node.py b/api/app/core/workflow/nodes/end/node.py index 2799316a..770cf328 100644 --- a/api/app/core/workflow/nodes/end/node.py +++ b/api/app/core/workflow/nodes/end/node.py @@ -36,8 +36,6 @@ class EndNode(BaseNode): Returns: 最终输出字符串 """ - logger.info(f"节点 {self.node_id} (End) 开始执行") - # 获取配置的输出模板 output_template = self.config.get("output") @@ -46,11 +44,4 @@ class EndNode(BaseNode): output = self._render_template(output_template, variable_pool, strict=False) else: output = "" - - # 统计信息（用于日志） - node_outputs = state.get("node_outputs", {}) - total_nodes = len(node_outputs) - - logger.info(f"节点 {self.node_id} (End) 执行完成，共执行 {total_nodes} 个节点") - return output diff --git a/api/app/core/workflow/nodes/enums.py b/api/app/core/workflow/nodes/enums.py index 43ab593b..5a603ac9 100644 --- a/api/app/core/workflow/nodes/enums.py +++ b/api/app/core/workflow/nodes/enums.py @@ -28,7 +28,7 @@ class NodeType(StrEnum): NOTES = "notes" -BRANCH_NODES = [NodeType.IF_ELSE, NodeType.HTTP_REQUEST, NodeType.QUESTION_CLASSIFIER] +BRANCH_NODES = frozenset({NodeType.IF_ELSE, NodeType.HTTP_REQUEST, NodeType.QUESTION_CLASSIFIER}) class ComparisonOperator(StrEnum): diff --git a/api/app/core/workflow/nodes/http_request/config.py b/api/app/core/workflow/nodes/http_request/config.py index fe38fafb..e1b84f0c 100644 --- a/api/app/core/workflow/nodes/http_request/config.py +++ b/api/app/core/workflow/nodes/http_request/config.py @@ -115,7 +115,7 @@ class HttpRetryConfig(BaseModel): ) -class HttpErrorDefaultTamplete(BaseModel): +class HttpErrorDefaultTemplate(BaseModel): body: str = Field( default="", description="Default body returned on HTTP error", @@ -143,7 +143,7 @@ class HttpErrorHandleConfig(BaseModel): description="Error handling strategy: 'none', 'default', or 'branch'", ) - default: HttpErrorDefaultTamplete | None = Field( + default: HttpErrorDefaultTemplate | None = Field( default=None, description="Default response template for error handling", ) diff --git a/api/app/core/workflow/nodes/http_request/node.py b/api/app/core/workflow/nodes/http_request/node.py index 23378c83..8aa8726e 100644 --- a/api/app/core/workflow/nodes/http_request/node.py +++ b/api/app/core/workflow/nodes/http_request/node.py @@ -16,7 +16,7 @@ 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 HttpRequestMethod, HttpErrorHandle, HttpAuthType, HttpContentType from app.core.workflow.nodes.http_request.config import HttpRequestNodeConfig, HttpRequestNodeOutput -from app.core.workflow.utils.file_processer import mime_to_file_type +from app.core.workflow.utils.file_processor import mime_to_file_type from app.core.workflow.variable.base_variable import VariableType, FileObject from app.core.workflow.variable.variable_objects import FileVariable, ArrayVariable from app.schemas import FileType, TransferMethod diff --git a/api/app/core/workflow/nodes/start/node.py b/api/app/core/workflow/nodes/start/node.py index a9618f7b..58567e6a 100644 --- a/api/app/core/workflow/nodes/start/node.py +++ b/api/app/core/workflow/nodes/start/node.py @@ -62,7 +62,6 @@ class StartNode(BaseNode): 包含系统参数、会话变量和自定义变量的字典 """ self.typed_config = StartNodeConfig(**self.config) - logger.info(f"节点 {self.node_id} (Start) 开始执行") # 处理自定义变量（传入 pool 避免重复创建） custom_vars = self._process_custom_variables(variable_pool) @@ -77,9 +76,9 @@ class StartNode(BaseNode): **custom_vars # 自定义变量作为节点输出的一部分 } - logger.info( - f"节点 {self.node_id} (Start) 执行完成，" - f"输出了 {len(custom_vars)} 个自定义变量" + logger.debug( + f"Node {self.node_id} (Start) execution completed, " + f"outputting {len(custom_vars)} custom variables" ) return result diff --git a/api/app/core/workflow/utils/file_processer.py b/api/app/core/workflow/utils/file_processor.py similarity index 100% rename from api/app/core/workflow/utils/file_processer.py rename to api/app/core/workflow/utils/file_processor.py diff --git a/api/app/core/workflow/validator.py b/api/app/core/workflow/validator.py index fe4aea19..683ccb98 100644 --- a/api/app/core/workflow/validator.py +++ b/api/app/core/workflow/validator.py @@ -6,6 +6,7 @@ import copy import logging +from collections import defaultdict, deque from typing import Any, Union, TYPE_CHECKING from app.core.workflow.nodes.enums import NodeType @@ -119,7 +120,6 @@ class WorkflowValidator: errors = [] graphs = cls.get_subgraph(workflow_config) - logger.info(graphs) for index, graph in enumerate(graphs): nodes = graph.get("nodes", []) edges = graph.get("edges", []) @@ -204,18 +204,18 @@ class WorkflowValidator: Returns: 可达节点 ID 集合 """ + adj = defaultdict(list) + for edge in edges: + adj[edge["source"]].append(edge["target"]) + reachable = {start_id} - queue = [start_id] - + queue = deque([start_id]) while queue: - current = queue.pop(0) - for edge in edges: - if edge.get("source") == current: - target = edge.get("target") - if target and target not in reachable: - reachable.add(target) - queue.append(target) - + current = queue.popleft() + for target in adj[current]: + if target not in reachable: + reachable.add(target) + queue.append(target) return reachable @staticmethod diff --git a/api/app/core/workflow/variable/variable_objects.py b/api/app/core/workflow/variable/variable_objects.py index 5e8e3f1e..79e023c1 100644 --- a/api/app/core/workflow/variable/variable_objects.py +++ b/api/app/core/workflow/variable/variable_objects.py @@ -54,7 +54,7 @@ class DictVariable(BaseVariable): def valid_value(self, value) -> dict: if not isinstance(value, dict): - raise TypeError(f"Value must be a dict - {type(value)}:{value}") + raise TypeError(f"Value must be a dict - {type(value)}:{value}") return value def to_literal(self) -> str: diff --git a/api/app/services/workflow_import_service.py b/api/app/services/workflow_import_service.py index 2b36c5ea..fd8f25f3 100644 --- a/api/app/services/workflow_import_service.py +++ b/api/app/services/workflow_import_service.py @@ -12,7 +12,7 @@ from app.aioRedis import aio_redis_set, aio_redis_get from app.core.config import settings from app.core.exceptions import BusinessException from app.core.workflow.adapters.base_adapter import WorkflowImportResult, WorkflowParserResult -from app.core.workflow.adapters.errors import UnsupportPlatform, InvalidConfiguration +from app.core.workflow.adapters.errors import UnsupportedPlatform, InvalidConfiguration from app.core.workflow.adapters.registry import PlatformAdapterRegistry from app.schemas import AppCreate from app.schemas.workflow_schema import WorkflowConfigCreate @@ -46,7 +46,7 @@ class WorkflowImportService: success=False, temp_id=None, workflow_id=None, - errors=[UnsupportPlatform(platform=platform)] + errors=[UnsupportedPlatform(platform=platform)] ) adapter = self.registry.get_adapter(platform, config)