"""Realtime Workflow orchestration without a Pipecat FlowManager.""" from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any from loguru import logger from models import AssistantConfig, RuntimeTool from pipecat.frames.frames import OutputTransportMessageUrgentFrame from pipecat.utils.time import time_now_iso8601 from db.session import SessionLocal from services.action_runtime import ActionRunner, ActionStatus from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction from services.brains.base import RealtimeBrainRuntime, SessionVariableUpdate from services.knowledge import search as search_knowledge from services.message_policy import MESSAGE_CONFIRMATION, MESSAGE_INTERRUPTIBLE from services.message_stage import ( MessageDisplaySpec, MessageStageRunner, MessageStageSpec, ) from services.pipecat.realtime_tools import RealtimeTool, RealtimeToolResult from services.runtime_variables import DynamicVariableError, DynamicVariableStore from services.system_tools import state_update_properties, system_tool_kind from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool from services.workflow.agent import EDGE_TOOL_STAGE_INSTRUCTION from services.workflow.models import WorkflowRuntimeState, WorkflowStatus from services.workflow.output import WorkflowOutput from services.workflow_engine import WorkflowEngine MAX_AUTOMATIC_HOPS = 50 ToolHandler = Callable[[dict[str, Any]], Awaitable[RealtimeToolResult]] @dataclass(frozen=True) class RealtimeActivation: """How the provider should continue after a completed function call.""" continue_response: bool = False after_output: Callable[[], Awaitable[None]] | None = None class RealtimeWorkflowOutput(WorkflowOutput): """Emit exact configured text while the realtime model produces its audio.""" def __init__( self, store: DynamicVariableStore, runtime: RealtimeBrainRuntime, ) -> None: super().__init__(store, runtime) # type: ignore[arg-type] self._realtime = runtime.realtime async def speak( self, text: str, *, source: str, node_id: str | None = None, record_history: bool = True, ) -> Awaitable[None] | None: content = text.strip() if not content: return None if record_history: self._store.record("agent", content) await self.emit( { "type": "transcript", "role": "assistant", "content": content, "timestamp": time_now_iso8601(), "source": source, **({"nodeId": node_id} if node_id else {}), } ) return await self._realtime.speak_fixed( content, suppress_transcript=True, ) class WorkflowRealtimeController: """Run Workflow v3 over one StepFun or Qwen realtime session.""" def __init__( self, *, cfg: AssistantConfig, engine: WorkflowEngine, store: DynamicVariableStore, runtime: RealtimeBrainRuntime, ) -> None: self._cfg = cfg self._engine = engine self._store = store self._runtime = runtime self._state = WorkflowRuntimeState(current_node_id=engine.start_id or "") self._output = RealtimeWorkflowOutput(store, runtime) self._tools = ToolExecutor(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._handlers: dict[str, ToolHandler] = {} self._turn_lock = asyncio.Lock() self._started = False self._ended = False self._waiting_generated_end = False self._message_advanced = asyncio.Event() self._last_recorded_user_text = "" def system_prompt(self) -> str: current = self._state.current_node_id if self._engine.node_type(current) == "agent": return self._agent_prompt(current) return self._store.render(self._engine.global_prompt()) async def start(self) -> None: if self._started: return self._started = True await self._output.mark_client_ready() self._state.enter(self._engine.start_id or "", WorkflowStatus.STARTING) await self._emit_active(self._state.current_node_id) await self._emit_variables("initialized", self._state.current_node_id) activation = await self._resolve_path( self._state.current_node_id, allow_visible_actions=True, ) if activation.continue_response: await self._runtime.realtime.request_response() async def on_client_ready(self) -> None: await self._output.mark_client_ready() await self._emit_active(self._state.current_node_id) await self._emit_variables("client_ready", self._state.current_node_id) async def on_session_update( self, dynamic_variables: dict[str, Any], ) -> SessionVariableUpdate: if self._ended: raise ValueError("工作流会话已经结束") async with self._turn_lock: changed = self._store.assign_declared_many(dynamic_variables) activation = await self._after_variables_changed( changed, reason="session_update", ) if activation and activation.after_output: await activation.after_output() elif activation and activation.continue_response: await self._runtime.realtime.request_response() return SessionVariableUpdate( changed=changed, dynamic_variables=self._store.public_values(), ) async def handle_text_input(self, content: str) -> bool: """Own realtime text input so the active node is updated first.""" if not content or self._ended: return True if self._state.status in { WorkflowStatus.RUNNING_ACTION, WorkflowStatus.ENDED, }: return True self._store.record("user", content) self._last_recorded_user_text = content self._state.begin_user_turn(content) async with self._turn_lock: current = self._state.current_node_id edge = self._engine.deterministic_edge( current, self._store, include_default=False, ) if edge: await self._runtime.realtime.send_text( content, run_immediately=False, ) activation = await self._follow_edge( edge, allow_visible_actions=True, ) if activation.continue_response: await self._runtime.realtime.request_response() return True await self._runtime.realtime.send_text(content, run_immediately=True) return True async def on_user_speech_started(self) -> None: """Let an interruptible Message finish its deterministic continuation.""" current = self._state.current_node_id if ( self._engine.node_type(current) != "message" or self._message_completion_policy(current) != MESSAGE_INTERRUPTIBLE ): return try: await asyncio.wait_for(self._message_advanced.wait(), timeout=1.0) except TimeoutError: logger.warning("Realtime Message 被打断后未能及时进入下一节点") def record_user_message(self, content: str) -> None: if content and not self._ended: if content == self._last_recorded_user_text: self._last_recorded_user_text = "" return self._store.record("user", content) async def on_assistant_text_end( self, content: str, interrupted: bool, ) -> None: if content and not interrupted and not self._ended: 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 if self._waiting_generated_end and self._runtime.call_end.ending: self._waiting_generated_end = False await self._runtime.call_end.finish_after_current_speech( has_text=bool(content.strip()) and not interrupted ) async def dispatch_tool( self, name: str, arguments: dict[str, Any], _call_id: str, ) -> RealtimeToolResult: handler = self._handlers.get(name) if handler is None: return RealtimeToolResult( {"status": "error", "message": f"当前节点没有工具:{name}"} ) return await handler(arguments) async def _resolve_path( self, node_id: str, *, allow_visible_actions: bool, ) -> RealtimeActivation: for hop in range(MAX_AUTOMATIC_HOPS): self._state.automatic_hops = hop node_type = self._engine.node_type(node_id) if node_type == "agent": return await self._activate_agent(node_id) if not allow_visible_actions and node_type in { "message", "action", "update_state", "handoff", "end", }: return RealtimeActivation( after_output=lambda node_id=node_id: self._run_deferred_path(node_id) ) if node_type == "end": await self._enter_end(node_id) return RealtimeActivation() if node_type == "message": succeeded = await self._enter_message(node_id) if not succeeded: return RealtimeActivation() elif node_type == "action": should_route = await self._enter_action(node_id) if not should_route: return RealtimeActivation() elif node_type == "update_state": await self._enter_update_state(node_id) elif node_type == "handoff": await self._enter_handoff(node_id) elif node_type == "start": self._state.enter(node_id, WorkflowStatus.STARTING) await self._emit_active(node_id) else: await self._output.emit_error( f"工作流指向未知节点:{node_id}", node_id=node_id, ) return RealtimeActivation() edge = self._engine.deterministic_edge( node_id, self._store, include_default=True, ) if not edge: self._state.enter(node_id, WorkflowStatus.WAITING_USER) if ( node_type != "agent" and self._runtime.set_input_enabled is not None ): self._runtime.set_input_enabled(False) return RealtimeActivation() await self._begin_edge(edge) node_id = str(edge.get("target") or "") await self._output.emit_error( "工作流连续自动跳转超过安全上限", node_id=node_id, ) return RealtimeActivation() async def _run_deferred_path(self, node_id: str) -> None: activation = await self._resolve_path(node_id, allow_visible_actions=True) if activation.after_output: await activation.after_output() elif activation.continue_response: await self._runtime.realtime.request_response() async def _follow_edge( self, edge: dict, *, allow_visible_actions: bool, ) -> RealtimeActivation: await self._begin_edge(edge) return await self._resolve_path( str(edge.get("target") or ""), allow_visible_actions=allow_visible_actions, ) async def _begin_edge(self, edge: dict) -> None: transition_id = self._state.begin_transition() await self._output.emit_trace( "edge_selected", revision=self._engine.revision, transition_id=transition_id, edgeId=str(edge.get("id") or ""), sourceNodeId=str(edge.get("source") or ""), targetNodeId=str(edge.get("target") or ""), mode=self._engine.edge_mode(edge), ) async def _activate_agent(self, node_id: str) -> RealtimeActivation: self._state.enter(node_id, WorkflowStatus.WAITING_USER) self._message_advanced.set() if self._runtime.set_input_enabled: self._runtime.set_input_enabled(True) await self._emit_active(node_id) tools = self._build_agent_tools(node_id) prompt = self._agent_prompt(node_id) await self._runtime.realtime.update_session(prompt, tools) generate = str( self._engine.data(node_id).get("entryMode") or "wait_user" ) == "generate" if generate: self._state.status = WorkflowStatus.RUNNING_AGENT else: self._state.consume_user_turn() return RealtimeActivation(continue_response=generate) def _agent_prompt(self, node_id: str) -> str: return ( f"{self._engine.prompt_for(node_id, self._store)}\n\n" f"[工作流执行规则]\n{EDGE_TOOL_STAGE_INSTRUCTION}" ) def _build_agent_tools(self, node_id: str) -> list[RealtimeTool]: stage = self._engine.agent_stage_config(node_id) transition_id = self._state.transition_id tools: list[RealtimeTool] = [] handlers: dict[str, ToolHandler] = {} def add(tool: RealtimeTool, handler: ToolHandler) -> None: if tool.name in handlers: logger.warning(f"跳过 Realtime Agent 的重复工具名:{tool.name}") return tools.append(tool) handlers[tool.name] = handler for tool_id in stage.tool_ids: runtime_tool = self._tool_by_id.get(str(tool_id)) if not runtime_tool: continue if runtime_tool.type == "system": built = self._system_tool( runtime_tool, node_id=node_id, transition_id=transition_id, state_variable_names=stage.state_variable_names, ) elif runtime_tool.type in {"http", "mcp", "client"}: built = self._business_tool( runtime_tool, node_id=node_id, transition_id=transition_id, ) else: built = None if built: add(*built) knowledge = self._knowledge_tool(node_id, transition_id) if knowledge: add(*knowledge) for edge in self._engine.edge_tool_edges(node_id): add(*self._transition_tool(edge, node_id, transition_id)) self._handlers = handlers return tools def _transition_tool( self, edge: dict, node_id: str, transition_id: int, ) -> tuple[RealtimeTool, ToolHandler]: async def handler(_arguments: dict[str, Any]) -> RealtimeToolResult: async with self._turn_lock: if not self._is_current(node_id, transition_id): return RealtimeToolResult( {"status": "stale", "message": "当前 Agent 已经切换。"}, continue_response=False, ) activation = await self._follow_edge( edge, allow_visible_actions=False, ) return RealtimeToolResult( { "status": "success", "targetNodeId": str(edge.get("target") or ""), }, continue_response=activation.continue_response, after_output=activation.after_output, ) return ( RealtimeTool( name=self._engine.edge_fn_name(edge), description=self._engine.edge_description(edge), ), handler, ) def _business_tool( self, tool: RuntimeTool, *, node_id: str, transition_id: int, ) -> tuple[RealtimeTool, ToolHandler]: properties, required = self._tools.schema_parts(tool) self._tools.register_secrets(tool) policy = policy_for_tool(tool) async def handler(arguments: dict[str, Any]) -> RealtimeToolResult: try: result = await self._tools.execute(tool, arguments) except ToolExecutionError as exc: return RealtimeToolResult( {"status": "error", "message": str(exc)} ) async with self._turn_lock: if not self._is_current(node_id, transition_id): return RealtimeToolResult( { **result, "status": "stale", "message": "工具完成时当前 Agent 已经切换。", }, continue_response=False, ) activation = await self._after_variables_changed( list(result.get("updated_variables") or []), reason="tool", ) return RealtimeToolResult( result, continue_response=( activation.continue_response if activation else policy.runs_llm_after_result ), after_output=activation.after_output if activation else None, ) return ( RealtimeTool( name=tool.function_name, description=tool.description or f"调用 {tool.name}", properties=properties, required=tuple(required), ), handler, ) def _system_tool( self, tool: RuntimeTool, *, node_id: str, transition_id: int, state_variable_names: tuple[str, ...], ) -> tuple[RealtimeTool, ToolHandler] | None: kind = system_tool_kind(tool.definition or {}) if kind == "update_state": allowed = frozenset(state_variable_names) async def update_state(arguments: dict[str, Any]) -> RealtimeToolResult: unauthorized = sorted(set(arguments) - allowed) if unauthorized: return RealtimeToolResult( { "status": "error", "message": "状态变量未获当前节点授权:" + ",".join(unauthorized), } ) try: changed = self._store.assign_declared_many(arguments) except DynamicVariableError as exc: return RealtimeToolResult( {"status": "error", "message": f"状态更新失败:{exc}"} ) async with self._turn_lock: if not self._is_current(node_id, transition_id): return RealtimeToolResult( {"status": "stale", "message": "当前 Agent 已经切换。"}, continue_response=False, ) activation = await self._after_variables_changed( changed, reason="update_state", ) return RealtimeToolResult( { "status": "success", "changed": changed, "variables": self._store.public_values(), }, continue_response=( activation.continue_response if activation else True ), after_output=activation.after_output if activation else None, ) return ( RealtimeTool( name=tool.function_name, description=tool.description or "更新已授权的动态变量。", properties=state_update_properties( self._cfg.dynamic_variable_definitions, allowed_names=state_variable_names, ), ), update_state, ) if kind == "skip_turn": async def skip_turn(arguments: dict[str, Any]) -> RealtimeToolResult: return RealtimeToolResult( { "status": "success", "action": "skip_turn", "reason": str(arguments.get("reason") or ""), }, continue_response=False, ) return ( RealtimeTool( name=tool.function_name, description=tool.description or "跳过当前轮次，不生成语音回复。", properties={ "reason": {"type": "string", "description": "跳过原因"} }, ), skip_turn, ) if kind == "request_human_handoff": async def request_handoff(arguments: dict[str, Any]) -> RealtimeToolResult: reason = str(arguments.get("reason") or "human_handoff") await self._runtime.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "handoff-requested", "source": "workflow-system-tool", "nodeId": node_id, "reason": reason, "message": "用户请求转接人工服务。", } ) ) self._store.values["system__handoff_status"] = "requested" return RealtimeToolResult( { "status": "requested", "action": "human_handoff_requested", } ) return ( RealtimeTool( name=tool.function_name, description=tool.description or "提交人工接管请求。", properties={ "reason": {"type": "string", "description": "转接原因"} }, ), request_handoff, ) if kind == "end_conversation": async def end_conversation(arguments: dict[str, Any]) -> RealtimeToolResult: reason = str(arguments.get("reason") or "end_conversation") self._waiting_generated_end = True self._runtime.call_end.begin(reason) return RealtimeToolResult( {"status": "success", "action": "ending_call"} ) return ( RealtimeTool( name=tool.function_name, description=tool.description or "礼貌结束本次对话。", properties={ "reason": {"type": "string", "description": "结束原因"} }, ), end_conversation, ) return None def _knowledge_tool( self, node_id: str, transition_id: int, ) -> tuple[RealtimeTool, ToolHandler] | 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 async def handler(arguments: dict[str, Any]) -> RealtimeToolResult: if not self._is_current(node_id, transition_id): return RealtimeToolResult( {"status": "stale", "message": "当前 Agent 已经切换。"}, continue_response=False, ) query = str(arguments.get("query") or "").strip() if not query: return RealtimeToolResult( {"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 RealtimeToolResult({"status": "ok", "results": results}) except Exception as exc: # noqa: BLE001 logger.warning(f"Realtime Workflow 知识库检索失败:{exc}") return RealtimeToolResult( {"status": "error", "message": "知识库检索暂时不可用"} ) return ( RealtimeTool( name="search_knowledge_base", description="在当前 Agent 绑定的知识库中检索资料。", properties={ "query": {"type": "string", "description": "完整问题或关键词"} }, required=("query",), ), handler, ) async def _after_variables_changed( self, changed: list[str], *, reason: str, ) -> RealtimeActivation | None: if not changed: return None current = self._state.current_node_id await self._emit_variables(reason, current, changed) edge = self._engine.deterministic_edge( current, self._store, include_default=False, ) if edge: return await self._follow_edge(edge, allow_visible_actions=False) if self._engine.node_type(current) == "agent": await self._runtime.realtime.update_session( self._agent_prompt(current), self._build_agent_tools(current), ) return None async def _enter_message(self, node_id: str) -> bool: self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) self._message_advanced.clear() await self._emit_active(node_id) data = self._engine.data(node_id) speech = self._store.render(str(data.get("speech") or "")).strip() policy = self._message_completion_policy(node_id) confirmation = policy == MESSAGE_CONFIRMATION if self._runtime.set_input_enabled: self._runtime.set_input_enabled(policy == MESSAGE_INTERRUPTIBLE) result = await self._message_stages.run( MessageStageSpec( speech=speech, display=( MessageDisplaySpec( title=self._store.render( str(data.get("title") or "重要提示") ), message=self._store.render(str(data.get("message") or "")), confirm_label=self._store.render( str(data.get("confirmLabel") or "确认") ), ) if confirmation else None ), completion_policy=policy, ), speak=lambda text: self._output.speak( text, source="workflow-message-speech", node_id=node_id, ), set_input_enabled=self._runtime.set_input_enabled, input_already_blocked=policy != MESSAGE_INTERRUPTIBLE, ) if not result.succeeded: await self._output.emit_error( result.error or "Message 节点执行失败", node_id=node_id, code="workflow_message_error", ) return result.succeeded async def _enter_action(self, node_id: str) -> bool: self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_active(node_id) if self._runtime.set_input_enabled: self._runtime.set_input_enabled(False) data = self._engine.data(node_id) tool = self._tool_by_id.get(str(data.get("toolId") or "")) 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), ), ), input_policy="block", ), set_input_enabled=self._runtime.set_input_enabled, input_already_blocked=True, ) outcome = result.outcomes[0] self._set_last_action(outcome) if outcome.updated_variables: await self._emit_variables( "action", node_id, list(outcome.updated_variables), ) return outcome.status != ActionStatus.CANCELLED async def _enter_update_state(self, node_id: str) -> None: self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_active(node_id) rendered = self._store.render_data( self._engine.data(node_id).get("assignments") or {} ) if not isinstance(rendered, dict): raise DynamicVariableError("Update State 节点赋值必须是对象") changed = self._store.assign_declared_many(rendered) if changed: await self._emit_variables("update_state", node_id, changed) async def _enter_handoff(self, node_id: str) -> None: self._state.enter(node_id, WorkflowStatus.HANDOFF) if self._runtime.set_input_enabled: self._runtime.set_input_enabled(False) await self._emit_active(node_id) data = self._engine.data(node_id) message = self._store.render(str(data.get("message") or "")) await self._runtime.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "handoff-requested", "nodeId": node_id, "targetType": data.get("targetType", "human"), "target": data.get("target", ""), "message": message, } ) ) if message: completion = await self._output.speak( message, source="workflow-handoff-speech", node_id=node_id, ) if completion: await completion 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() self._handlers = {} await self._runtime.realtime.update_session("会话已经结束。", []) if self._runtime.set_input_enabled: self._runtime.set_input_enabled(False) await self._emit_active(node_id) 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 self._runtime.queue_frame( OutputTransportMessageUrgentFrame( message={"type": "flow-ended", "nodeId": node_id} ) ) if message: completion = await self._output.speak( message, source="workflow-end-speech", node_id=node_id, ) if completion: await completion return self._runtime.call_end.begin("workflow_completed") if message: self._runtime.call_end.arm_after_speech() completion = await self._output.speak( message, source="workflow-end-speech", node_id=node_id, ) if completion: await completion else: await self._runtime.call_end.finish() def _is_current(self, node_id: str, transition_id: int) -> bool: return ( not self._ended and self._state.current_node_id == node_id and self._state.transition_id == transition_id ) def _message_completion_policy(self, node_id: str) -> str: return str( self._engine.data(node_id).get("completionPolicy") or "playback" ) @staticmethod def _action_result_assignments( data: dict[str, Any], ) -> dict[str, str] | None: mode = str(data.get("resultAssignmentMode") or "none") if mode == "inherit": return None if mode == "override": value = data.get("resultAssignments") return dict(value) if isinstance(value, dict) else {} return {} def _set_last_action(self, outcome) -> None: self._store.values.update( { "system__last_action_status": ( "ok" if outcome.status == ActionStatus.SUCCESS else "error" ), "system__last_action_invocation_id": outcome.invocation_id, "system__last_action_duration_ms": outcome.duration_ms, "system__last_action_error": ( outcome.error.message if outcome.error else "" ), } ) async def _emit_active(self, node_id: str | None) -> None: await self._output.emit_node_active(node_id) async def _emit_variables( self, reason: str, node_id: str | None, changed: list[str] | None = None, ) -> None: await self._output.emit_variables( reason=reason, node_id=node_id, changed=changed, )