"""Pipecat Flows-backed Workflow v3 brain.""" from __future__ import annotations import asyncio from copy import deepcopy from dataclasses import 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 from services.knowledge import search as search_knowledge 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 class ConfiguredFlowManager(FlowManager): """Preserve Flow transitions while suppressing late async-tool replies.""" 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._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._greeting_context_message: dict[str, str] | None = None self._startup_waiting_for_greeting = False async def greeting(self, cfg: AssistantConfig) -> str: return self._engine.greeting(self._store) or cfg.greeting 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._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._greeting_context_message = None self._startup_waiting_for_greeting = False 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 def prepare_greeting_context( self, greeting: str, context: LLMContext, ) -> dict[str, str] | None: message = super().prepare_greeting_context(greeting, context) self._greeting_context_message = deepcopy(message) if message else None return message 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 尚未初始化") self._startup_waiting_for_greeting = greeting_pending if greeting_pending: # Keep the Workflow on Start until the transport confirms that the # shared greeting has finished. This prevents an initial Agent's # fixed speech (or generated reply) from racing the greeting. await self._manager.initialize( self._passive_node_config(self._engine.start_id) ) logger.info("工作流等待 Start 开场白播放完毕") return node_config = await self._initial_node_config() await self._manager.initialize(node_config) await self._after_node_activated(node_config) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") async def on_greeting_finished(self) -> None: """Enter the first node only after Start's greeting reaches playback end.""" if not self._startup_waiting_for_greeting or self._ended: return self._startup_waiting_for_greeting = False manager = self._require_manager() if manager.current_node != self._engine.start_id: return node_config = await self._initial_node_config() if node_config.get("name") == self._engine.start_id: self._state.enter(self._engine.start_id, WorkflowStatus.WAITING_USER) return await manager.set_node_from_config(node_config) await self._after_node_activated(node_config) logger.info(f"Start 开场白结束,进入节点: {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) await self._emit_variables( reason="client_ready", node_id=current_node, ) 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 manager.set_node_from_config(next_config) await self._after_node_activated( 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, greeting_context_message=self._greeting_context_message, leading_messages=leading_messages, ) async def _after_node_activated( self, node_config: NodeConfig, *, triggering_user_text: str = "", ) -> None: """Publish activation and perform exactly one Agent entry behavior.""" node_id = str(node_config.get("name") or "") node_type = self._engine.node_type(node_id) 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") if entry_mode == "fixed_speech": entry_speech = self._store.render(str(data.get("entrySpeech") or "")) await self._queue_visible_speech( entry_speech, source="workflow-fixed-reply", node_id=node_id, ) self._state.enter(node_id, WorkflowStatus.WAITING_USER) self._state.consume_user_turn() return 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 _queue_visible_speech( self, text: str, *, source: str = "workflow-speech", node_id: str | None = None, ) -> None: 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" 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 "") if self._engine.node_type(node_id) != "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 and entry_mode != "fixed_speech" ) configured["pre_actions"] = [ { "type": "workflow_function_transition_entry", "node_id": node_id, "entry_mode": entry_mode, "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) entry_mode = str(action.get("entry_mode") or "wait_user") if entry_mode == "fixed_speech": entry_speech = self._store.render( str(self._engine.data(node_id).get("entrySpeech") or "") ) await self._queue_visible_speech( entry_speech, source="workflow-fixed-reply", node_id=node_id, ) self._state.enter(node_id, WorkflowStatus.WAITING_USER) self._state.consume_user_turn() elif 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, *, triggering_user_text: str = "", ) -> NodeConfig: self._state.begin_transition() leading_messages: list[dict[str, str]] = [] 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, ) leading_messages.append( {"role": "assistant", "content": content} ) return await self._resolve_path( str(edge.get("target") or ""), leading_messages=leading_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": await self._enter_action(node_id) 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) self._state.begin_transition() 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("工作流连续自动跳转超过安全上限") async def _enter_action(self, node_id: str) -> None: self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_node_active(node_id) data = self._engine.data(node_id) tool_id = str(data.get("toolId") or "") tool = self._tool_by_id.get(tool_id) if not tool: self._store.values["system__last_action_status"] = "error" self._store.values["system__last_action_error"] = f"工具不存在:{tool_id}" return try: arguments = self._store.render_data(data.get("arguments") or {}) result = await self._tools.execute( tool, arguments, result_assignments=data.get("resultAssignments") or {}, ) updated_variables = list(result.get("updated_variables") or []) if updated_variables: await self._emit_variables( reason="action", node_id=node_id, changed=updated_variables, ) self._store.values["system__last_action_status"] = "ok" self._store.values["system__last_action_error"] = "" except (ToolExecutionError, ValueError) as exc: self._store.values["system__last_action_status"] = "error" self._store.values["system__last_action_error"] = str(exc)[:2048] 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") 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) -> None: await self._require_output().emit_node_active(node_id) 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, ) 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