"""Pipecat Flows-backed Workflow v3 brain.""" from __future__ import annotations import asyncio from collections.abc import Awaitable from copy import deepcopy from dataclasses import dataclass, replace from typing import Any from loguru import logger from models import AssistantConfig, RuntimeTool from db.session import SessionLocal from pipecat.flows import ( ContextStrategy, ContextStrategyConfig, FlowManager, FlowsFunctionSchema, NodeConfig, ) from pipecat.frames.frames import ( LLMRunFrame, LLMUpdateSettingsFrame, OutputTransportMessageUrgentFrame, ) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor from pipecat.services.llm_service import ( FunctionCallParams, FunctionCallResultProperties, ) from services.brains.base import ( BaseBrain, BrainRuntime, BrainSpec, SessionVariableUpdate, ) from services.action_runtime import ( ActionInvocationCancelled, ActionOutcome, ActionRunner, ActionStatus, ) from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction from services.knowledge import search as search_knowledge from services.message_stage import ( MessageDisplaySpec, MessageStageResult, MessageStageRunner, MessageStageSpec, ) from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool from services.workflow.agent import WorkflowAgentStage from services.workflow.models import RouteStatus, WorkflowRuntimeState, WorkflowStatus from services.workflow.output import WorkflowOutput from services.workflow.routing import WorkflowEdgeEvaluator from services.workflow_engine import WorkflowEngine from services.workflow_router import WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 @dataclass class _MessageContinuation: """Resume one Workflow path after its visible Message gate completes.""" token: int node_id: str context_messages: list[dict[str, str]] triggering_user_text: str task: asyncio.Task[None] | None = None class ConfiguredFlowManager(FlowManager): """Preserve Flow transitions while suppressing late async-tool replies.""" ENTRY_ACTION_TYPE = "workflow_function_transition_entry" async def _set_node(self, node_id: str, node_config: NodeConfig) -> None: """Notify Workflow only after FlowManager committed the active node.""" await super()._set_node(node_id, node_config) after_activation = node_config.get("workflow_after_activation") if callable(after_activation): await after_activation(node_id) async def _create_transition_func(self, name, handler): transition = await super()._create_transition_func(name, handler) if not getattr(handler, "_suppress_followup_llm", False): return transition async def configured_transition(params: FunctionCallParams) -> None: original_callback = params.result_callback async def result_callback(result, *, properties=None): if properties and properties.on_context_updated: # Deterministic Workflow transitions already use run_llm=False # and must retain their context-updated callback. configured_properties = properties elif properties: configured_properties = replace(properties, run_llm=False) else: configured_properties = FunctionCallResultProperties( run_llm=False ) await original_callback( result, properties=configured_properties, ) await transition(replace(params, result_callback=result_callback)) return configured_transition class WorkflowBrain(BaseBrain): spec = BrainSpec( type="workflow", supported_runtime_modes=frozenset({"pipeline"}), owns_context=True, ) def __init__(self, cfg_or_graph: AssistantConfig | dict[str, Any]): cfg = cfg_or_graph if isinstance(cfg_or_graph, AssistantConfig) else None graph = deepcopy(cfg.graph if cfg is not None else cfg_or_graph) if cfg is not None: # Graph v3 owns Workflow defaults. Keep older saved graphs compatible # by filling the new interaction settings from the assistant row. settings = graph.setdefault("settings", {}) settings.setdefault("enableInterrupt", cfg.enableInterrupt) settings.setdefault("turnConfig", deepcopy(cfg.turnConfig)) self._engine = WorkflowEngine(graph or {}) if not self._engine.has_graph() or not self._engine.start_id: raise ValueError("WorkflowBrain 缺少有效的 Start 节点") self._cfg = cfg self._store = DynamicVariableStore.from_config(cfg or AssistantConfig(type="workflow")) self._tools = ToolExecutor(self._store) self._actions = ActionRunner(self._tools) self._action_stages = ActionStageRunner(self._actions) self._message_stages = MessageStageRunner() self._tool_by_id: dict[str, RuntimeTool] = { tool.id: tool for tool in (cfg.tools if cfg else []) } self._runtime: BrainRuntime | None = None self._manager: FlowManager | None = None self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) self._edge_evaluator = WorkflowEdgeEvaluator( self._engine, self._store, self._router_for_node, ) self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id) self._turn_lock = asyncio.Lock() self._output: WorkflowOutput | None = None self._agent_stage: WorkflowAgentStage | None = None self._ended = False self._next_message_token = 1 self._pending_message: _MessageContinuation | None = None async def greeting(self, _cfg: AssistantConfig) -> str: """Workflow opening speech belongs to an explicit Message or Agent.""" return "" def system_prompt(self, cfg: AssistantConfig) -> str: return self._store.render(self._engine.global_prompt()) def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: from services.pipecat.service_factory import create_llm return create_llm(cfg) async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: if runtime.worker is None or runtime.context_aggregator is None: raise RuntimeError("WorkflowBrain 需要 PipelineWorker 和 context aggregator pair") self._cfg = cfg self._runtime = runtime self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store, client_tools=runtime.client_tools) self._actions = ActionRunner( self._tools, is_session_ending=lambda: runtime.call_end.ending, ) self._action_stages = ActionStageRunner(self._actions) self._message_stages = MessageStageRunner(runtime.client_tools) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) self._edge_evaluator = WorkflowEdgeEvaluator( self._engine, self._store, self._router_for_node, ) self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id) self._turn_lock = asyncio.Lock() self._output = WorkflowOutput(self._store, runtime) self._agent_stage = WorkflowAgentStage( cfg=cfg, engine=self._engine, store=self._store, runtime=runtime, ) self._ended = False self._next_message_token = 1 self._pending_message = None self._manager = ConfiguredFlowManager( worker=runtime.worker, llm=runtime.llm, context_aggregator=runtime.context_aggregator, transport=runtime.transport, global_functions=runtime.flow_global_functions, ) self._manager.state["variables"] = self._store.values async def on_connected(self, *, greeting_pending: bool = False) -> None: self._state.enter(self._engine.start_id, WorkflowStatus.STARTING) await self._emit_node_active(self._engine.start_id) await self._emit_variables( reason="initialized", node_id=self._engine.start_id, ) if self._manager is None: raise RuntimeError("Workflow FlowManager 尚未初始化") node_config = await self._initial_node_config() await self._activate_node_config(node_config, initialize=True) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") async def _initial_node_config(self) -> NodeConfig: """Only a default-only Start advances before the first user turn.""" outgoing = self._engine.outgoing(self._engine.start_id) has_condition = any( self._engine.edge_mode(edge) != "always" for edge in outgoing ) if has_condition: self._state.enter(self._engine.start_id, WorkflowStatus.WAITING_USER) return self._passive_node_config(self._engine.start_id) edge = next( ( candidate for candidate in outgoing if self._engine.edge_mode(candidate) == "always" ), None, ) return ( await self._follow_edge(edge) if edge else self._passive_node_config(self._engine.start_id) ) async def on_client_ready(self) -> None: """Replay state that may have been emitted before WebRTC data was ready.""" await self._require_output().mark_client_ready() current_node = ( str(self._manager.current_node) if self._manager and self._manager.current_node else self._state.current_node_id ) if current_node != self._state.current_node_id: self._state.current_node_id = current_node await self._emit_node_active(current_node, trace=False) await self._emit_variables( reason="client_ready", node_id=current_node, ) async def on_session_update( self, dynamic_variables: dict[str, Any], ) -> SessionVariableUpdate: if self._ended: raise ValueError("工作流会话已经结束") changed = self._store.assign_declared_many(dynamic_variables) current = self._state.current_node_id if changed and current and self._engine.node_type(current) == "agent": await self._refresh_agent_prompt(current) if changed: await self._emit_variables( reason="session_update", node_id=current or None, changed=changed, ) return SessionVariableUpdate( changed=changed, dynamic_variables=self._store.public_values(), ) def record_user_message(self, content: str) -> None: if content and not self._ended: self._store.record("user", content) async def on_user_turn_end(self, content: str) -> bool: """Route a complete user turn before the active stage may reply.""" if not content or self._ended: return True async with self._turn_lock: return await self._handle_user_turn_end(content) async def _handle_user_turn_end(self, content: str) -> bool: """Serialized implementation so one user turn cannot transition twice.""" self.record_user_message(content) self._state.begin_user_turn(content) manager = self._require_manager() current = self._state.current_node_id if not current: return True self._state.status = WorkflowStatus.ROUTING decision = await self._edge_evaluator.evaluate(current) if decision.status == RouteStatus.ERROR: await self._require_output().emit_error( decision.error or "工作流路由失败", node_id=current, code="workflow_routing_error", ) return await self._continue_current_node_after_no_transition(current) if decision.edge and manager.current_node == current: next_config = await self._follow_edge( decision.edge, triggering_user_text=content, ) await self._activate_node_config( next_config, triggering_user_text=content, ) return True return await self._continue_current_node_after_no_transition(current) async def _continue_current_node_after_no_transition( self, node_id: str, ) -> bool: if self._engine.node_type(node_id) != "agent": self._state.enter(node_id, WorkflowStatus.WAITING_USER) return True await self._refresh_agent_prompt(node_id) self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) await self._require_runtime().queue_frame(LLMRunFrame()) return True async def _select_edge( self, node_id: str, ) -> dict | None: """Compatibility helper used by automatic-node traversal and tests.""" decision = await self._edge_evaluator.evaluate(node_id) if decision.status == RouteStatus.ERROR: await self._require_output().emit_error( decision.error or "工作流路由失败", node_id=node_id, code="workflow_routing_error", ) return None return decision.edge async def on_assistant_text_end( self, _turn_id: str, content: str, interrupted: bool, ) -> None: if not content or interrupted or self._ended: return self._store.record("agent", content, completed_agent_turn=True) self._state.consume_user_turn() if self._engine.node_type(self._state.current_node_id) == "agent": self._state.status = WorkflowStatus.WAITING_USER async def _refresh_agent_prompt(self, node_id: str) -> None: await self._require_agent_stage().refresh_prompt(node_id) def _agent_role_message(self, node_id: str) -> str: return self._require_agent_stage().role_message(node_id) def _router_for_node(self, node_id: str) -> WorkflowLLMRouter: if self._agent_stage is None: return self._router return self._agent_stage.router_for_node(node_id, self._router) async def _apply_agent_stage(self, node_id: str) -> None: await self._require_agent_stage().apply(node_id) def _agent_config( self, node_id: str, leading_messages: list[dict[str, str]] | None = None, ) -> NodeConfig: stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] for tool_id in stage.tool_ids: tool = self._tool_by_id.get(str(tool_id)) if tool and tool.type in {"http", "mcp", "client"}: functions.append(self._flow_tool(tool, node_id)) knowledge_function = self._knowledge_function(node_id) if knowledge_function: functions.append(knowledge_function) if stage.vision_enabled and self._require_runtime().vision_function: functions.append(self._require_runtime().vision_function) return self._require_agent_stage().node_config( node_id, functions=functions, leading_messages=leading_messages, ) async def _after_node_activated( self, node_config: NodeConfig, *, triggering_user_text: str = "", ) -> None: """Run the entry behavior owned by the newly active node.""" node_id = str(node_config.get("name") or "") node_type = self._engine.node_type(node_id) if node_type == "message": await self._activate_message_continuation(node_id) return if node_type != "agent": if node_type == "start": self._state.enter(node_id, WorkflowStatus.WAITING_USER) return await self._emit_node_active(node_id) data = self._engine.data(node_id) entry_mode = str(data.get("entryMode") or "wait_user") should_run = entry_mode == "generate" or bool(triggering_user_text) if should_run: self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) await self._require_runtime().queue_frame(LLMRunFrame()) return self._state.enter(node_id, WorkflowStatus.WAITING_USER) async def _activate_node_config( self, node_config: NodeConfig, *, triggering_user_text: str = "", initialize: bool = False, ) -> None: """Install one node and dispatch its entry behavior exactly once.""" manager = self._require_manager() if initialize: await manager.initialize(node_config) else: await manager.set_node_from_config(node_config) await self._after_node_activated( node_config, triggering_user_text=triggering_user_text, ) async def _queue_visible_speech( self, text: str, *, source: str = "workflow-speech", node_id: str | None = None, ) -> Awaitable[None] | None: return await self._require_output().speak( text, source=source, node_id=node_id, ) def _passive_node_config( self, node_id: str, task_messages: list[dict[str, str]] | None = None, ) -> NodeConfig: """Keep a non-conversational terminal node active without ending the call.""" return { "name": node_id, "role_message": self._store.render(self._engine.global_prompt()), "task_messages": list(task_messages or []), "functions": [], "context_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND), "respond_immediately": False, } def _flow_tool(self, tool: RuntimeTool, node_id: str) -> FlowsFunctionSchema: properties, required = self._tools.schema_parts(tool) self._tools.register_secrets(tool) policy = policy_for_tool(tool) async def handler(args, _flow_manager): transition_id = self._state.transition_id try: result = await self._tools.execute(tool, dict(args or {})) except ToolExecutionError as exc: return {"status": "error", "message": str(exc)} if ( self._state.current_node_id != node_id or self._state.transition_id != transition_id ): return { "status": "stale", "message": "工具完成时当前 Agent 已经切换，结果不再触发路由。", } updated_variables = list(result.get("updated_variables") or []) if updated_variables: await self._emit_variables( reason="tool", node_id=node_id, changed=updated_variables, ) await self._refresh_agent_prompt(node_id) edge = self._engine.deterministic_edge( node_id, self._store, include_default=False, ) if edge: next_config = await self._follow_edge( edge, triggering_user_text=( self._state.pending_user_turn.text if self._state.pending_user_turn else "" ), ) return result, self._flow_managed_transition_config( next_config, triggering_user_text=( self._state.pending_user_turn.text if self._state.pending_user_turn else "" ), ) return result if not policy.runs_llm_after_result: setattr(handler, "_suppress_followup_llm", True) return FlowsFunctionSchema( name=tool.function_name, description=tool.description or f"调用 {tool.name}", properties=properties, required=required, handler=handler, cancel_on_interruption=policy.cancel_on_interruption, timeout_secs=( float(((tool.definition or {}).get("config") or {}).get("timeout_seconds") or 3) if tool.type == "client" and policy.response_wait_mode == "timeout" else None ), ) def _flow_managed_transition_config( self, node_config: NodeConfig, *, triggering_user_text: str, ) -> NodeConfig: """Let FlowManager finish a tool result before activating its target. Function-returned transitions are the one place where FlowManager must schedule the LLM run. Its pre-action only updates our explicit runtime state; it never performs a second LLM run. """ node_id = str(node_config.get("name") or "") node_type = self._engine.node_type(node_id) if node_type == "message": configured = dict(node_config) configured["workflow_after_activation"] = ( self._activate_message_continuation ) return configured if node_type != "agent": return node_config entry_mode = str( self._engine.data(node_id).get("entryMode") or "wait_user" ) should_run = entry_mode == "generate" or bool(triggering_user_text) configured = dict(node_config) configured["respond_immediately"] = should_run configured["pre_actions"] = [ { "type": ConfiguredFlowManager.ENTRY_ACTION_TYPE, "node_id": node_id, "should_run": should_run, "handler": self._activate_from_flow_transition, } ] return configured async def _activate_from_flow_transition( self, action: dict, _flow_manager: FlowManager, ) -> None: """Apply visible entry state without manually queueing an LLM run.""" node_id = str(action.get("node_id") or "") await self._emit_node_active(node_id) if action.get("should_run"): self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) else: self._state.enter(node_id, WorkflowStatus.WAITING_USER) def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None: stage = self._engine.agent_stage_config(node_id) knowledge_id = str(stage.knowledge_base_id or "") if not knowledge_id or stage.knowledge_mode != "on_demand": return None cfg = self._cfg or AssistantConfig(type="workflow") knowledge = cfg.workflow_knowledge_bases.get(knowledge_id) description = "在当前 Agent 绑定的知识库中检索资料。" if knowledge: description += f"知识库：{knowledge.name}。{knowledge.description}" async def handler(args, _flow_manager): query = str((args or {}).get("query") or "").strip() if not query: return {"status": "error", "message": "检索问题为空"} try: async with SessionLocal() as session: results = await search_knowledge( session, knowledge_id, query, top_k=stage.knowledge_top_n, score_threshold=stage.knowledge_score_threshold, ) return {"status": "ok", "results": results} except Exception as exc: # noqa: BLE001 - tool errors are returned to the LLM logger.warning(f"Workflow 知识库检索失败:{exc}") return {"status": "error", "message": "知识库检索暂时不可用"} return FlowsFunctionSchema( name="search_knowledge_base", description=description, properties={ "query": {"type": "string", "description": "完整问题或检索关键词"} }, required=["query"], handler=handler, cancel_on_interruption=True, ) async def _follow_edge( self, edge: dict, *, leading_messages: list[dict[str, str]] | None = None, triggering_user_text: str = "", ) -> NodeConfig: await self._begin_edge_transition(edge) context_messages = list(leading_messages or []) speech = self._engine.edge_transition_speech(edge) if speech: content = self._store.render(speech).strip() if content: await self._queue_visible_speech( content, source="workflow-edge-transition", node_id=str(edge.get("target") or "") or None, ) context_messages.append( {"role": "assistant", "content": content} ) return await self._resolve_path( str(edge.get("target") or ""), leading_messages=context_messages, triggering_user_text=triggering_user_text, ) async def _resolve_path( self, node_id: str, *, leading_messages: list[dict[str, str]] | None = None, triggering_user_text: str = "", ) -> NodeConfig: context_messages = list(leading_messages or []) for hop in range(MAX_AUTOMATIC_HOPS): self._state.automatic_hops = hop node_type = self._engine.node_type(node_id) if node_type == "agent": await self._apply_agent_stage(node_id) agent_messages = context_messages if ( triggering_user_text and self._engine.data(node_id).get("contextPolicy") == "fresh" ): agent_messages = [ {"role": "user", "content": triggering_user_text}, *context_messages, ] return self._agent_config(node_id, agent_messages) if node_type == "end": await self._enter_end(node_id) return self._passive_node_config(node_id, context_messages) if node_type == "action": outcome = await self._enter_action(node_id) if not outcome.should_route: return self._passive_node_config(node_id, context_messages) elif node_type == "message": self._prepare_message_continuation( node_id, context_messages=context_messages, triggering_user_text=triggering_user_text, ) return self._passive_node_config(node_id, context_messages) elif node_type == "handoff": await self._enter_handoff(node_id) elif node_type == "start": self._state.enter(node_id, WorkflowStatus.WAITING_USER) await self._emit_node_active(node_id) else: raise RuntimeError(f"工作流指向未知节点:{node_id}") if not self._engine.has_outgoing(node_id): return self._passive_node_config(node_id, context_messages) edge = await self._select_edge(node_id) if not edge: return self._passive_node_config(node_id, context_messages) await self._begin_edge_transition(edge) speech = self._engine.edge_transition_speech(edge) if speech: content = self._store.render(speech).strip() if content: target_id = str(edge.get("target") or "") await self._queue_visible_speech( content, source="workflow-edge-transition", node_id=target_id or None, ) context_messages.append( {"role": "assistant", "content": content} ) node_id = str(edge.get("target") or "") raise RuntimeError("工作流连续自动跳转超过安全上限") def _prepare_message_continuation( self, node_id: str, *, context_messages: list[dict[str, str]], triggering_user_text: str, ) -> None: """Save the path state without waiting inside the pipeline call stack.""" token = self._next_message_token self._next_message_token += 1 self._pending_message = _MessageContinuation( token=token, node_id=node_id, context_messages=[dict(message) for message in context_messages], triggering_user_text=triggering_user_text, ) self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) async def _activate_message_continuation(self, node_id: str) -> None: """Start a prepared Message only after FlowManager made it active.""" continuation = self._pending_message if ( continuation is None or continuation.node_id != node_id or continuation.task is not None ): return self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) await self._emit_node_active(node_id) runtime = self._require_runtime() if runtime.set_input_enabled is not None: runtime.set_input_enabled(False) continuation.task = asyncio.create_task( self._complete_message_continuation(continuation), name=f"workflow-message-{node_id}-{continuation.token}", ) async def _complete_message_continuation( self, continuation: _MessageContinuation, ) -> None: """Wait for playback/confirmation outside routing, then resume once.""" try: result = await self._enter_message( continuation.node_id, already_active=True, input_already_blocked=True, ) if not result.succeeded: if self._pending_message is continuation: self._pending_message = None return await self._resume_message_continuation(continuation, result) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - keep callback failures visible logger.exception(f"Message 节点续跑失败:{exc}") if self._pending_message is continuation: self._pending_message = None self._state.enter( continuation.node_id, WorkflowStatus.WAITING_USER, ) runtime = self._require_runtime() if runtime.set_input_enabled is not None: runtime.set_input_enabled(True) await self._require_output().emit_error( "Message 节点完成后无法继续工作流", node_id=continuation.node_id, code="workflow_message_resume_error", ) async def _resume_message_continuation( self, continuation: _MessageContinuation, result: MessageStageResult, ) -> None: """Advance from the completed Message without blocking media frames.""" async with self._turn_lock: manager = self._require_manager() if ( self._ended or self._pending_message is not continuation or self._state.current_node_id != continuation.node_id or str(manager.current_node or "") != continuation.node_id ): return self._pending_message = None context_messages = [ dict(message) for message in continuation.context_messages ] if result.speech: context_messages.append( {"role": "assistant", "content": result.speech} ) if not self._engine.has_outgoing(continuation.node_id): self._state.enter( continuation.node_id, WorkflowStatus.WAITING_USER, ) return edge = await self._select_edge(continuation.node_id) if not edge: self._state.enter( continuation.node_id, WorkflowStatus.WAITING_USER, ) return next_config = await self._follow_edge( edge, leading_messages=context_messages, triggering_user_text=continuation.triggering_user_text, ) await self._activate_node_config( next_config, triggering_user_text=continuation.triggering_user_text, ) async def _enter_action(self, node_id: str) -> ActionOutcome: self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_node_active(node_id) data = self._engine.data(node_id) runtime = self._require_runtime() invocation_id = self._actions.new_invocation_id() tool_id = str(data.get("toolId") or "") tool = self._tool_by_id.get(tool_id) try: stage_result = await self._action_stages.run( ActionStageSpec( actions=( StageAction( id=node_id, tool=tool, arguments=data.get("arguments") or {}, result_assignments=self._action_result_assignments(data), invocation_id=invocation_id, ), ), input_policy=( "block" if data.get("userInputPolicy") == "block" else "queue" ), ), set_input_enabled=runtime.set_input_enabled, on_started=lambda: self._emit_trace( "action_started", nodeId=node_id, invocationId=invocation_id, toolId=tool_id, toolType=tool.type if tool else None, ), ) outcome = stage_result.outcomes[0] updated_variables = list(outcome.updated_variables) if updated_variables: await self._emit_variables( reason="action", node_id=node_id, changed=updated_variables, ) except ActionInvocationCancelled as exc: outcome = exc.outcome self._set_last_action(outcome) await self._emit_action_outcome(node_id, outcome) raise self._set_last_action(outcome) await self._emit_action_outcome(node_id, outcome) return outcome async def _enter_message( self, node_id: str, *, already_active: bool = False, input_already_blocked: bool = False, ) -> MessageStageResult: if not already_active: self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) await self._emit_node_active(node_id) data = self._engine.data(node_id) runtime = self._require_runtime() speech = self._store.render(str(data.get("speech") or "")).strip() show_message = bool(data.get("showMessage", False)) require_confirmation = bool(data.get("requireConfirmation", False)) display = ( MessageDisplaySpec( title=self._store.render( str(data.get("title") or "重要提示") ).strip(), message=self._store.render( str(data.get("message") or "") ).strip(), confirm_label=self._store.render( str(data.get("confirmLabel") or "确认") ).strip(), ) if show_message else None ) result = await self._message_stages.run( MessageStageSpec( speech=speech, display=display, require_confirmation=require_confirmation, ), speak=lambda content: self._queue_visible_speech( content, source="workflow-message-speech", node_id=node_id, ), set_input_enabled=runtime.set_input_enabled, input_already_blocked=input_already_blocked, on_started=lambda: self._emit_trace( "message_started", nodeId=node_id, hasSpeech=bool(speech), showsMessage=show_message, requiresConfirmation=require_confirmation, ), ) if result.succeeded: await self._emit_trace( "message_completed", nodeId=node_id, action=result.action, ) return result self._state.enter(node_id, WorkflowStatus.WAITING_USER) await self._emit_trace( "message_failed", nodeId=node_id, error=result.error or "Message 节点执行失败", ) await self._require_output().emit_error( result.error or "Message 节点执行失败", node_id=node_id, code="workflow_message_error", ) return result def _set_last_action(self, outcome: ActionOutcome) -> None: legacy_status = { ActionStatus.SUCCESS: "ok", ActionStatus.FAILURE: "error", ActionStatus.CANCELLED: "cancelled", }[outcome.status] self._store.values.update( { "system__last_action_status": legacy_status, "system__last_action_invocation_id": outcome.invocation_id, "system__last_action_duration_ms": outcome.duration_ms, "system__last_action_error_code": ( outcome.error.code if outcome.error else "" ), "system__last_action_error": ( outcome.error.message if outcome.error else "" ), } ) async def _emit_action_outcome( self, node_id: str, outcome: ActionOutcome, ) -> None: event = { ActionStatus.SUCCESS: "action_completed", ActionStatus.FAILURE: "action_failed", ActionStatus.CANCELLED: "action_cancelled", }[outcome.status] await self._emit_trace(event, nodeId=node_id, outcome=outcome.trace_payload()) @staticmethod def _action_result_assignments( data: dict[str, Any], ) -> dict[str, str] | None: """Resolve node mapping semantics for ToolExecutor. ``None`` means inherit the reusable tool's mapping, while an empty dictionary explicitly disables all result assignments. """ mode = str(data.get("resultAssignmentMode") or "none") if mode == "inherit": return None if mode == "override": assignments = data.get("resultAssignments") return dict(assignments) if isinstance(assignments, dict) else {} return {} async def _enter_handoff(self, node_id: str) -> None: self._state.enter(node_id, WorkflowStatus.HANDOFF) await self._emit_node_active(node_id) data = self._engine.data(node_id) message = self._store.render(str(data.get("message") or "")) await self._require_runtime().queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "handoff-requested", "nodeId": node_id, "targetType": data.get("targetType", "human"), "target": data.get("target", ""), "message": message, } ) ) if message: await self._queue_visible_speech(message) self._store.values["system__handoff_status"] = "requested" async def _enter_end(self, node_id: str) -> None: self._ended = True self._state.enter(node_id, WorkflowStatus.ENDED) self._state.finish() await self._emit_node_active(node_id) runtime = self._require_runtime() if runtime.set_knowledge_scope: runtime.set_knowledge_scope({"mode": "disabled"}) if runtime.set_vision_scope: runtime.set_vision_scope({"enabled": False}) if runtime.set_input_enabled: runtime.set_input_enabled(False) data = self._engine.data(node_id) message = self._store.render(str(data.get("message") or "")) scope = str(data.get("scope") or "session") await self._emit_trace( "workflow_ended", nodeId=node_id, scope=scope, outcome="success", ) if scope == "flow": await runtime.queue_frame( OutputTransportMessageUrgentFrame( message={"type": "flow-ended", "nodeId": node_id} ) ) if message: await self._queue_visible_speech(message) return runtime.call_end.begin("workflow_completed") if message: await self._queue_visible_speech(message) arm_tracked = getattr(runtime.call_end, "arm_after_tracked_speech", None) if callable(arm_tracked): await arm_tracked() elif message: runtime.call_end.arm_after_speech() else: await runtime.call_end.finish() async def _emit_node_active( self, node_id: str | None, *, trace: bool = True, ) -> None: await self._require_output().emit_node_active(node_id) if trace and node_id: await self._emit_trace( "node_entered", nodeId=node_id, nodeType=self._engine.node_type(node_id), ) async def _begin_edge_transition(self, edge: dict) -> None: transition_id = self._state.begin_transition() source_id = str(edge.get("source") or self._state.current_node_id or "") target_id = str(edge.get("target") or "") if source_id: await self._emit_trace("node_exited", nodeId=source_id) await self._emit_trace( "edge_selected", edgeId=str(edge.get("id") or ""), sourceNodeId=source_id, targetNodeId=target_id, edgeMode=self._engine.edge_mode(edge), transitionId=transition_id, ) async def _emit_trace(self, event: str, **details: Any) -> None: transition_id = int(details.pop("transitionId", self._state.transition_id)) try: await self._require_output().emit_trace( event, revision=self._engine.revision, transition_id=transition_id, **details, ) except Exception as exc: # noqa: BLE001 - trace must not alter execution logger.warning(f"发送 Workflow 轨迹失败，不影响当前流程: {exc}") async def _emit_variables( self, *, reason: str, node_id: str | None, changed: list[str] | None = None, ) -> None: """Publish a safe snapshot so Workflow debug mirrors runtime state.""" await self._require_output().emit_variables( reason=reason, node_id=node_id, changed=changed, ) public_variables = self._require_output().public_variables() public_names = [ name for name in (changed or public_variables.keys()) if not name.startswith(("system__", "secret__")) ] await self._emit_trace( "variables_updated" if changed else "variables_snapshot", nodeId=node_id, reason=reason, variableNames=public_names, variables=public_variables, ) def _require_runtime(self) -> BrainRuntime: if self._runtime is None: raise RuntimeError("WorkflowBrain 尚未绑定 pipeline runtime") return self._runtime def _require_manager(self) -> FlowManager: if self._manager is None: raise RuntimeError("Workflow FlowManager 尚未初始化") return self._manager def _require_output(self) -> WorkflowOutput: if self._output is None: # A few focused unit tests bind BrainRuntime directly. Lazily create # the output adapter so those tests exercise the same production path. self._output = WorkflowOutput(self._store, self._require_runtime()) return self._output def _require_agent_stage(self) -> WorkflowAgentStage: if self._agent_stage is None: self._agent_stage = WorkflowAgentStage( cfg=self._cfg or AssistantConfig(type="workflow"), engine=self._engine, store=self._store, runtime=self._require_runtime(), ) return self._agent_stage