diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index cdd7d4c..331fb0d 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -37,6 +37,14 @@ def _validate_workflow(body: AssistantUpsert) -> None: return body.graph = normalize_graph(body.graph or {}) errors = validate_graph(body.graph) + settings = body.graph.get("settings") or {} + graph_runtime_mode = str(settings.get("runtimeMode") or "pipeline") + if graph_runtime_mode != body.runtime_mode: + errors.append("工作流全局运行模式与 Assistant runtimeMode 不一致") + graph_realtime_id = str(settings.get("defaultRealtimeResourceId") or "") + bound_realtime_id = str(body.model_resource_ids.get("Realtime") or "") + if graph_realtime_id != bound_realtime_id: + errors.append("工作流 Realtime 模型与 Assistant Realtime 绑定不一致") declared_variables = set(body.dynamic_variable_definitions) for node in body.graph.get("nodes") or []: node_id = str(node.get("id") or "") @@ -80,6 +88,7 @@ async def _validate_workflow_references( resource_expectations: dict[str, str] = {} vision_resource_ids: set[str] = set() for key, capability in ( + ("defaultRealtimeResourceId", "Realtime"), ("defaultLlmResourceId", "LLM"), ("defaultAsrResourceId", "ASR"), ("defaultTtsResourceId", "TTS"), diff --git a/backend/schemas.py b/backend/schemas.py index bf02295..d3e5aae 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -40,10 +40,10 @@ SystemToolKind = Literal[ # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵 EXTERNAL_TYPES = {"dify", "fastgpt", "opencode"} -# MVP 仅 PromptBrain 支持 realtime;Workflow 和外部托管大脑只走 pipeline。 +# Prompt 和 Workflow 支持 realtime;外部托管大脑只走 pipeline。 # 与 services.brains 各 BrainSpec.supported_runtime_modes 对齐(此处独立声明, # 避免 HTTP schema 层为做校验而引入 pipecat 重依赖)。 -REALTIME_CAPABLE_TYPES = {"prompt"} +REALTIME_CAPABLE_TYPES = {"prompt", "workflow"} class CamelModel(BaseModel): diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 6587e01..08e2e03 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -101,6 +101,18 @@ class BrainRuntime: flow_global_functions: list[Any] = field(default_factory=list) +@dataclass(frozen=True) +class RealtimeBrainRuntime: + """Pipeline-owned capabilities for a speech-to-speech brain session.""" + + realtime: Any + queue_frame: Callable[[Frame], Awaitable[None]] + call_end: CallEndPort + session_id: str = "" + client_tools: ClientToolPort | None = None + set_input_enabled: Callable[[bool], None] | None = None + + class BaseBrain: """No-op lifecycle defaults for brains without local orchestration.""" @@ -118,6 +130,16 @@ class BaseBrain: async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: """Register tools and initialize per-call orchestration.""" + async def setup_realtime( + self, + cfg: AssistantConfig, + runtime: RealtimeBrainRuntime, + ) -> None: + """Initialize optional speech-to-speech orchestration.""" + + async def on_realtime_user_speech_started(self) -> None: + """Allow a workflow to move past an interruptible fixed message.""" + async def run_preflight(self) -> None: """Run deterministic server-side startup work before media starts.""" @@ -211,6 +233,14 @@ class Brain(Protocol): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: ... + async def setup_realtime( + self, + cfg: AssistantConfig, + runtime: RealtimeBrainRuntime, + ) -> None: ... + + async def on_realtime_user_speech_started(self) -> None: ... + async def run_preflight(self) -> None: ... async def on_connected(self, *, greeting_pending: bool = False) -> None: ... diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index a969c4f..59920e5 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -33,6 +33,7 @@ from services.brains.base import ( BaseBrain, BrainRuntime, BrainSpec, + RealtimeBrainRuntime, SessionVariableUpdate, ) from services.action_runtime import ( @@ -64,6 +65,7 @@ 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.realtime import WorkflowRealtimeController from services.workflow_engine import WorkflowEngine from services.workflow_router import WorkflowLLMRouter @@ -147,7 +149,7 @@ class ConfiguredFlowManager(FlowManager): class WorkflowBrain(BaseBrain): spec = BrainSpec( type="workflow", - supported_runtime_modes=frozenset({"pipeline"}), + supported_runtime_modes=frozenset({"pipeline", "realtime"}), owns_context=True, ) @@ -188,12 +190,15 @@ class WorkflowBrain(BaseBrain): self._waiting_for_generated_end_speech = False self._next_message_token = 1 self._pending_message: _MessageContinuation | None = None + self._realtime_controller: WorkflowRealtimeController | 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: + if self._realtime_controller is not None: + return self._realtime_controller.system_prompt() return self._store.render(self._engine.global_prompt()) def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: @@ -234,6 +239,7 @@ class WorkflowBrain(BaseBrain): self._waiting_for_generated_end_speech = False self._next_message_token = 1 self._pending_message = None + self._realtime_controller = None self._manager = ConfiguredFlowManager( worker=runtime.worker, llm=runtime.llm, @@ -243,7 +249,30 @@ class WorkflowBrain(BaseBrain): ) self._manager.state["variables"] = self._store.values + async def setup_realtime( + self, + cfg: AssistantConfig, + runtime: RealtimeBrainRuntime, + ) -> None: + self._cfg = cfg + self._store = DynamicVariableStore.from_config(cfg) + self._realtime_controller = WorkflowRealtimeController( + cfg=cfg, + engine=self._engine, + store=self._store, + runtime=runtime, + ) + runtime.realtime.set_tool_dispatcher( + self._realtime_controller.dispatch_tool + ) + runtime.realtime.set_speech_started_handler( + self.on_realtime_user_speech_started + ) + async def on_connected(self, *, greeting_pending: bool = False) -> None: + if self._realtime_controller is not None: + await self._realtime_controller.start() + return self._state.enter(self._engine.start_id, WorkflowStatus.STARTING) await self._emit_node_active(self._engine.start_id) await self._emit_variables( @@ -282,6 +311,9 @@ class WorkflowBrain(BaseBrain): async def on_client_ready(self) -> None: """Replay state that may have been emitted before WebRTC data was ready.""" + if self._realtime_controller is not None: + await self._realtime_controller.on_client_ready() + return await self._require_output().mark_client_ready() current_node = ( str(self._manager.current_node) @@ -300,24 +332,41 @@ class WorkflowBrain(BaseBrain): self, dynamic_variables: dict[str, Any], ) -> SessionVariableUpdate: + if self._realtime_controller is not None: + return await self._realtime_controller.on_session_update( + dynamic_variables + ) 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, + async with self._turn_lock: + changed = self._store.assign_declared_many(dynamic_variables) + current = self._state.current_node_id + next_config = ( + await self._after_variables_changed( + current, + changed, + reason="session_update", + ) + if current + else None ) + if changed and not current: + await self._emit_variables( + reason="session_update", + node_id=None, + changed=changed, + ) + if next_config: + await self._activate_node_config(next_config) return SessionVariableUpdate( changed=changed, dynamic_variables=self._store.public_values(), ) def record_user_message(self, content: str) -> None: + if self._realtime_controller is not None: + self._realtime_controller.record_user_message(content) + return if content and not self._ended: self._store.record("user", content) @@ -327,6 +376,8 @@ class WorkflowBrain(BaseBrain): user_message: dict[str, Any] | None = None, ) -> bool: """Route a complete user turn before the active stage may reply.""" + if self._realtime_controller is not None: + return await self._realtime_controller.handle_text_input(content) if not content or self._ended: return True async with self._turn_lock: @@ -369,6 +420,26 @@ class WorkflowBrain(BaseBrain): self.record_user_message(content) self._state.begin_user_turn(content) + if self._engine.llm_routing_mode() == "edge_tool": + self._state.status = WorkflowStatus.ROUTING + edge = self._engine.deterministic_edge( + current, + self._store, + include_default=False, + ) + if edge and manager.current_node == current: + next_config = await self._follow_edge( + edge, + triggering_user_text=content, + triggering_user_message=user_message, + ) + await self._activate_node_config( + next_config, + triggering_user_text=content, + ) + return True + return await self._continue_current_node_after_no_transition(current) + self._state.status = WorkflowStatus.ROUTING decision = await self._edge_evaluator.evaluate( current, @@ -487,6 +558,12 @@ class WorkflowBrain(BaseBrain): node_id: str, ) -> dict | None: """Compatibility helper used by automatic-node traversal and tests.""" + if self._engine.llm_routing_mode() == "edge_tool": + return self._engine.deterministic_edge( + node_id, + self._store, + include_default=True, + ) decision = await self._edge_evaluator.evaluate(node_id) if decision.status == RouteStatus.ERROR: await self._require_output().emit_error( @@ -507,6 +584,12 @@ class WorkflowBrain(BaseBrain): content: str, interrupted: bool, ) -> None: + if self._realtime_controller is not None: + await self._realtime_controller.on_assistant_text_end( + content, + interrupted, + ) + return if content and not interrupted and not self._ended: self._store.record("agent", content, completed_agent_turn=True) self._state.consume_user_turn() @@ -522,6 +605,10 @@ class WorkflowBrain(BaseBrain): has_text=bool(content.strip()) and not interrupted ) + async def on_realtime_user_speech_started(self) -> None: + if self._realtime_controller is not None: + await self._realtime_controller.on_user_speech_started() + async def _refresh_agent_prompt(self, node_id: str) -> None: await self._require_agent_stage().refresh_prompt(node_id) @@ -579,6 +666,9 @@ class WorkflowBrain(BaseBrain): append_function(self._knowledge_function(node_id)) if stage.vision_enabled and self._require_runtime().vision_function: append_function(self._require_runtime().vision_function) + if self._engine.llm_routing_mode() == "edge_tool": + for edge in self._engine.edge_tool_edges(node_id): + append_function(self._edge_tool(edge, node_id)) return self._require_agent_stage().node_config( node_id, functions=functions, @@ -688,26 +778,13 @@ class WorkflowBrain(BaseBrain): } 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 "" - ), + async with self._turn_lock: + next_config = await self._after_variables_changed( + node_id, + updated_variables, + reason="tool", ) + if next_config: return result, self._flow_managed_transition_config( next_config, triggering_user_text=( @@ -735,6 +812,80 @@ class WorkflowBrain(BaseBrain): ), ) + def _edge_tool(self, edge: dict, node_id: str) -> FlowsFunctionSchema: + """Expose one natural-language edge as an Agent-owned transition tool.""" + registered_transition_id = self._state.transition_id + + async def handler(_args, _flow_manager): + async with self._turn_lock: + if ( + self._state.current_node_id != node_id + or self._state.transition_id != registered_transition_id + ): + return { + "status": "stale", + "message": "当前 Agent 已经切换，本次跳转不再执行。", + } + triggering_user_text = ( + self._state.pending_user_turn.text + if self._state.pending_user_turn + else "" + ) + next_config = await self._follow_edge( + edge, + triggering_user_text=triggering_user_text, + ) + return ( + { + "status": "success", + "targetNodeId": str(edge.get("target") or ""), + }, + self._flow_managed_transition_config( + next_config, + triggering_user_text=triggering_user_text, + ), + ) + + return FlowsFunctionSchema( + name=self._engine.edge_fn_name(edge), + description=self._engine.edge_description(edge), + properties={}, + required=[], + handler=handler, + cancel_on_interruption=True, + ) + + async def _after_variables_changed( + self, + node_id: str, + changed: list[str], + *, + reason: str, + ) -> NodeConfig | None: + """Refresh the Agent and take a matching expression without an LLM.""" + if not changed: + return None + await self._emit_variables(reason=reason, node_id=node_id, changed=changed) + if self._state.current_node_id != node_id: + return None + if self._engine.node_type(node_id) == "agent": + await self._refresh_agent_prompt(node_id) + edge = self._engine.deterministic_edge( + node_id, + self._store, + include_default=False, + ) + if not edge: + return None + return await self._follow_edge( + edge, + triggering_user_text=( + self._state.pending_user_turn.text + if self._state.pending_user_turn + else "" + ), + ) + def _workflow_system_tool( self, tool: RuntimeTool, @@ -779,18 +930,27 @@ class WorkflowBrain(BaseBrain): changed = self._store.assign_declared_many(values) except DynamicVariableError as exc: return {"status": "error", "message": f"状态更新失败: {exc}"} - if changed: - await self._emit_variables( - reason="update_state", - node_id=node_id, - changed=changed, - ) - await self._refresh_agent_prompt(node_id) - return { + result = { "status": "success", "changed": changed, "variables": self._store.public_values(), } + async with self._turn_lock: + next_config = await self._after_variables_changed( + node_id, + changed, + reason="update_state", + ) + if next_config: + 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 return FlowsFunctionSchema( name=tool.function_name, diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 7c11ad7..93d09c5 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -24,6 +24,8 @@ NODE_TYPES = { "end", } EDGE_MODES = {"llm", "expression", "always"} +WORKFLOW_RUNTIME_MODES = {"pipeline", "realtime"} +WORKFLOW_LLM_ROUTING_MODES = {"llm_router", "edge_tool"} AGENT_ENTRY_MODES = {"wait_user", "generate"} ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"} ACTION_USER_INPUT_POLICIES = {"queue", "block"} @@ -254,7 +256,12 @@ def _normalize_message_data(data: dict[str, Any]) -> None: def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: + # Missing values belong to graphs saved before workflow runtime/routing + # became explicit. Preserve their dedicated-router behavior. + settings.setdefault("runtimeMode", "pipeline") + settings.setdefault("llmRoutingMode", "llm_router") settings.setdefault("globalPrompt", global_prompt) + settings.setdefault("defaultRealtimeResourceId", "") settings.setdefault("defaultLlmResourceId", "") settings.setdefault("defaultAsrResourceId", "") settings.setdefault("defaultTtsResourceId", "") @@ -418,6 +425,27 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: return [] errors: list[str] = [] + settings = graph.get("settings") or {} + runtime_mode = str(settings.get("runtimeMode") or "pipeline") + llm_routing_mode = str(settings.get("llmRoutingMode") or "llm_router") + if runtime_mode not in WORKFLOW_RUNTIME_MODES: + errors.append(f"工作流运行模式无效:{runtime_mode}") + if llm_routing_mode not in WORKFLOW_LLM_ROUTING_MODES: + errors.append(f"大模型判断路由模式无效:{llm_routing_mode}") + if runtime_mode == "realtime" and llm_routing_mode != "edge_tool": + errors.append("Realtime 工作流的大模型判断路由只能使用边工具模式") + if runtime_mode == "realtime" and not settings.get( + "defaultRealtimeResourceId" + ): + errors.append("Realtime 工作流必须选择 Realtime 模型") + if runtime_mode == "realtime" and settings.get("visionEnabled"): + errors.append("Realtime 工作流暂不支持视觉能力") + if ( + runtime_mode == "realtime" + and settings.get("knowledgeBaseId") + and settings.get("knowledgeMode") != "on_demand" + ): + errors.append("Realtime 工作流的知识库只能使用按需模式") node_by_id: dict[str, dict] = {} counts: dict[str, int] = defaultdict(int) for node in nodes: @@ -443,6 +471,34 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: not isinstance(name, str) for name in state_names ): errors.append(f"Agent 节点 {node_id} 的状态变量授权必须是列表") + if runtime_mode == "realtime": + if data.get("contextPolicy", "inherit") != "inherit": + errors.append( + f"Realtime Agent 节点 {node_id} 只能继承会话上下文" + ) + if any( + data.get(key) + for key in ( + "llmResourceId", + "asrResourceId", + "ttsResourceId", + ) + ) or any(key in data for key in ("enableInterrupt", "turnConfig")): + errors.append( + f"Realtime Agent 节点 {node_id} 不支持节点级模型、语音或交互策略覆盖" + ) + source = settings if data.get("inheritGlobalConfig", True) else data + if source.get("visionEnabled"): + errors.append( + f"Realtime Agent 节点 {node_id} 暂不支持视觉能力" + ) + if ( + source.get("knowledgeBaseId") + and source.get("knowledgeMode") != "on_demand" + ): + errors.append( + f"Realtime Agent 节点 {node_id} 的知识库只能使用按需模式" + ) elif node_type == "message": data = node.get("data") or {} speech = data.get("speech") @@ -500,6 +556,10 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: errors.append( f"Action 节点 {node_id} 的用户输入策略无效:{input_policy}" ) + elif runtime_mode == "realtime" and input_policy != "block": + errors.append( + f"Realtime Action 节点 {node_id} 必须阻止执行期间的用户输入" + ) elif node_type == "update_state": data = node.get("data") or {} assignments = data.get("assignments") @@ -550,6 +610,13 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if priority in priorities[source_id]: errors.append(f"节点 {source_id} 的条件边优先级不能重复:{priority}") priorities[source_id].add(priority) + source_type = node_by_id[source_id].get("type") + if llm_routing_mode == "edge_tool" and mode == "llm" and source_type != "agent": + errors.append(f"边工具模式的大模型判断边只能从 Agent 发出:{edge_id}") + if llm_routing_mode == "edge_tool" and mode == "always" and source_type == "agent": + errors.append( + f"边工具模式的 Agent 不能使用默认路径:{source_id}" + ) incoming[target_id] += 1 outgoing[source_id] += 1 adj[source_id].append(target_id) @@ -633,6 +700,7 @@ def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]: str(value) for value in ( settings.get("defaultLlmResourceId"), + settings.get("defaultRealtimeResourceId"), settings.get("defaultAsrResourceId"), settings.get("defaultTtsResourceId"), ( diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 0086a4f..bbfa09d 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -16,6 +16,7 @@ from models import AssistantConfig from openai import AsyncOpenAI from PIL import Image from services.brains import Brain, BrainRuntime, build_brain +from services.brains.base import RealtimeBrainRuntime from services.conversation_history import ConversationRecorder from services.pipecat.call_lifecycle import ( CallEndCoordinator, @@ -73,6 +74,7 @@ from services.pipecat.processors import ( KnowledgeRetrievalProcessor, PassthroughLLMAssistantAggregator, RealtimeDynamicVariableProcessor, + RealtimeInputAudioGateProcessor, RealtimeUserInputProcessor, SessionUpdateProcessor, UserInput, @@ -890,7 +892,32 @@ async def run_realtime_pipeline( instructions=brain.system_prompt(cfg), ) input_sample_rate, output_sample_rate = realtime_audio_sample_rates(cfg) - user_input = RealtimeUserInputProcessor() + worker_holder: dict[str, PipelineWorker] = {} + input_state = {"enabled": True} + + async def queue_call_end(reason: str) -> None: + worker = worker_holder.get("worker") + if worker is None: + return + logger.info(f"结束 Realtime 通话: reason={reason}") + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "call-ended", "reason": reason} + ) + ) + await worker.queue_frame(EndFrame()) + + call_end = CallEndCoordinator(queue_call_end) + client_tools = ClientToolBroker() + client_tools.set_interrupt_handler(realtime.interrupt) + user_input = RealtimeUserInputProcessor( + should_ignore_input=lambda: ( + call_end.ending or not input_state["enabled"] + ) + ) + input_gate = RealtimeInputAudioGateProcessor( + lambda: not call_end.ending and input_state["enabled"] + ) dynamic_variables = RealtimeDynamicVariableProcessor(brain, cfg, realtime) async def refresh_realtime_instructions() -> None: @@ -910,14 +937,22 @@ async def run_realtime_pipeline( channel=channel, runtime_mode=cfg.runtimeMode, session_id=cfg.conversation_id or None, + extra=( + WorkflowEngine(cfg.graph).session_metadata() + if cfg.type == "workflow" + else None + ), ) pipeline = Pipeline( [ transport.input(), + client_tools, session_update, user_input, + input_gate, realtime, dynamic_variables, + EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), transport.output(), ] @@ -931,11 +966,28 @@ async def run_realtime_pipeline( ), enable_rtvi=False, ) + worker_holder["worker"] = worker + + def set_input_enabled(enabled: bool) -> None: + input_state["enabled"] = enabled + + await brain.setup_realtime( + cfg, + RealtimeBrainRuntime( + realtime=realtime, + queue_frame=worker.queue_frame, + call_end=call_end, + session_id=cfg.conversation_id or "", + client_tools=client_tools, + set_input_enabled=set_input_enabled, + ), + ) bind_realtime_pipeline_events( transport=transport, worker=worker, realtime=realtime, + brain=brain, text_input=user_input, greeting=greeting, ) diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py index 9fd1b89..277c1b9 100644 --- a/backend/services/pipecat/pipeline_events.py +++ b/backend/services/pipecat/pipeline_events.py @@ -227,6 +227,7 @@ def bind_realtime_pipeline_events( transport, worker, realtime, + brain, text_input, greeting: str, ) -> None: @@ -251,10 +252,16 @@ def bind_realtime_pipeline_events( await queue_transcript("user", user_input.text) if user_input.run_immediately and user_input.interrupt: await realtime.interrupt() - await realtime.send_text( - user_input.text, - run_immediately=user_input.run_immediately, + handled = ( + await brain.on_user_turn_end(user_input.text) + if brain.spec.type == "workflow" + else False ) + if not handled: + await realtime.send_text( + user_input.text, + run_immediately=user_input.run_immediately, + ) await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ @@ -267,6 +274,8 @@ def bind_realtime_pipeline_events( @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): + await brain.on_connected(greeting_pending=bool(greeting)) + await brain.on_client_ready() if greeting: await realtime.speak(greeting) diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index 8039382..b9a97a8 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -19,6 +19,7 @@ from pipecat.frames.frames import ( FunctionCallCancelFrame, FunctionCallResultFrame, FunctionCallsStartedFrame, + InputAudioRawFrame, InputTransportMessageFrame, InterruptionFrame, LLMContextFrame, @@ -520,11 +521,26 @@ class RealtimeDynamicVariableProcessor(FrameProcessor): await self.push_frame(frame, direction) +class RealtimeInputAudioGateProcessor(FrameProcessor): + """Drop live microphone frames while a deterministic node owns the turn.""" + + def __init__(self, is_enabled: Callable[[], bool]): + super().__init__() + self._is_enabled = is_enabled + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, InputAudioRawFrame) and not self._is_enabled(): + return + await self.push_frame(frame, direction) + + class RealtimeUserInputProcessor(FrameProcessor): """Route text-only user-input messages to a realtime service.""" - def __init__(self): + def __init__(self, should_ignore_input: Callable[[], bool] | None = None): super().__init__() + self._should_ignore_input = should_ignore_input or (lambda: False) self._register_event_handler("on_user_input") async def process_frame(self, frame, direction: FrameDirection): @@ -542,6 +558,12 @@ class RealtimeUserInputProcessor(FrameProcessor): if user_input is None: await self.push_frame(frame, direction) return + if self._should_ignore_input(): + await self._emit_error( + user_input.input_id, + "当前工作流节点暂不接收用户输入", + ) + return if user_input.has_camera_frame: await self._emit_error( user_input.input_id, diff --git a/backend/services/pipecat/qwen_audio_realtime.py b/backend/services/pipecat/qwen_audio_realtime.py index 6107ea0..2dfa759 100644 --- a/backend/services/pipecat/qwen_audio_realtime.py +++ b/backend/services/pipecat/qwen_audio_realtime.py @@ -35,6 +35,12 @@ from pipecat.utils.time import time_now_iso8601 from websockets.asyncio.client import connect as websocket_connect from websockets.protocol import State +from services.pipecat.realtime_tools import ( + RealtimeTool, + RealtimeToolDispatcher, + RealtimeToolSession, +) + DEFAULT_QWEN_AUDIO_REALTIME_MODEL = "qwen-audio-3.0-realtime-flash" DEFAULT_QWEN_AUDIO_REALTIME_VOICE = "longanqian" @@ -43,6 +49,7 @@ QWEN_OUTPUT_SAMPLE_RATE = 24_000 SUPPORTED_TURN_DETECTION_MODES = frozenset({"server_vad", "smart_turn"}) ExtraEventHandler = Callable[[dict[str, Any]], Awaitable[None] | None] +SpeechStartedHandler = Callable[[], Awaitable[None]] class QwenAudioRealtimeService(AIService): @@ -115,6 +122,12 @@ class QwenAudioRealtimeService(AIService): self._user_transcript_item_id = "" self._user_transcript_timestamp = "" self._deferred_assistant_messages: list[dict[str, Any]] = [] + self._tools: list[RealtimeTool] = [] + self._tool_session = RealtimeToolSession(self._send_tool_event) + self._fixed_speech_completion: asyncio.Future[None] | None = None + self._suppress_response_transcript = False + self._speech_started_handler: SpeechStartedHandler | None = None + self._function_names: dict[str, str] = {} async def start(self, frame: StartFrame) -> None: await super().start(frame) @@ -197,6 +210,15 @@ class QwenAudioRealtimeService(AIService): await self._cancel_active_response() await self.broadcast_interruption() + async def request_response(self) -> None: + await self._send_event({"type": "response.create"}) + + def set_speech_started_handler( + self, + handler: SpeechStartedHandler | None, + ) -> None: + self._speech_started_handler = handler + async def speak(self, text: str) -> None: """Ask Qwen to speak a fixed greeting, then remove the hidden request. @@ -204,8 +226,21 @@ class QwenAudioRealtimeService(AIService): instruction field. A temporary user item keeps this behavior within the supported protocol; it is deleted after the response completes. """ + await self.speak_fixed(text, suppress_transcript=False) + + async def speak_fixed( + self, + text: str, + *, + suppress_transcript: bool = True, + ) -> Awaitable[None] | None: + """Speak configured text and expose the provider response boundary.""" if not text: - return + return None + completion = asyncio.get_running_loop().create_future() + self._resolve_fixed_speech() + self._fixed_speech_completion = completion + self._suppress_response_transcript = suppress_transcript item_id = f"item_{uuid4().hex}" self._greeting_request_item_id = item_id await self._send_event( @@ -228,6 +263,7 @@ class QwenAudioRealtimeService(AIService): } ) await self._send_event({"type": "response.create"}) + return completion async def update_instructions(self, instructions: str) -> None: """Update only instructions after startup. @@ -246,6 +282,33 @@ class QwenAudioRealtimeService(AIService): wait_until_ready=False, ) + async def update_session( + self, + instructions: str, + tools: list[RealtimeTool], + ) -> None: + """Atomically replace the active Workflow prompt and tool catalog.""" + self._instructions = instructions + self._tools = list(tools) + if self._session_ready.is_set(): + await self._send_event( + { + "type": "session.update", + "session": { + "instructions": instructions, + "tools": [tool.provider_schema() for tool in tools], + "tool_choice": "auto", + }, + }, + wait_until_ready=False, + ) + + def set_tool_dispatcher( + self, + dispatcher: RealtimeToolDispatcher | None, + ) -> None: + self._tool_session.set_dispatcher(dispatcher) + def _connection_url(self) -> str: parts = urlsplit(self._base_url) query = dict(parse_qsl(parts.query)) @@ -263,6 +326,8 @@ class QwenAudioRealtimeService(AIService): "output_audio_format": "pcm", "turn_detection": self._turn_detection_config(), "max_history_turns": self._max_history_turns, + "tools": [tool.provider_schema() for tool in self._tools], + "tool_choice": "auto", } def _turn_detection_config(self) -> dict[str, Any]: @@ -314,6 +379,9 @@ class QwenAudioRealtimeService(AIService): self._user_transcript_item_id = "" self._user_transcript_timestamp = "" self._deferred_assistant_messages.clear() + self._tool_session.clear() + self._function_names.clear() + self._resolve_fixed_speech() if websocket and websocket.state is State.OPEN: try: await websocket.close() @@ -368,11 +436,15 @@ class QwenAudioRealtimeService(AIService): ) ) elif event_type in {"response.audio_transcript.delta", "response.text.delta"}: - if not self._audio_suppressed: + if not self._audio_suppressed and not self._suppress_response_transcript: await self._append_assistant_text(str(event.get("delta") or "")) elif event_type in {"response.audio_transcript.done", "response.text.done"}: transcript = str(event.get("transcript") or event.get("text") or "") - if transcript and not self._audio_suppressed: + if ( + transcript + and not self._audio_suppressed + and not self._suppress_response_transcript + ): if self._assistant_turn_id: self._assistant_text = transcript else: @@ -385,6 +457,8 @@ class QwenAudioRealtimeService(AIService): user_turn_timestamp = time_now_iso8601() await self._cancel_active_response() await self.broadcast_interruption() + if self._speech_started_handler is not None: + await self._speech_started_handler() await self._start_user_transcript_turn(event, user_turn_timestamp) elif ( event_type == "input_audio_buffer.speech_stopped" @@ -398,11 +472,20 @@ class QwenAudioRealtimeService(AIService): self._response_active = False await self._finish_assistant_text(interrupted=interrupted) await self._delete_greeting_request() + self._resolve_fixed_speech() + elif event_type == "response.output_item.added": + self._remember_function_call(event) + elif event_type in { + "response.function_call_arguments.done", + "response.output_item.done", + }: + await self._handle_function_call_event(event) elif event_type == "error": error = event.get("error") message = error.get("message") if isinstance(error, dict) else str(error) if "cancel" not in str(message).lower(): await self.push_error(f"Qwen-Audio Realtime error: {message}") + self._resolve_fixed_speech() handler = self._extra_event_handlers.get(event_type) if handler: @@ -420,6 +503,52 @@ class QwenAudioRealtimeService(AIService): ) self._response_active = False await self._finish_assistant_text(interrupted=True) + self._resolve_fixed_speech() + + async def _send_tool_event(self, payload: dict[str, Any]) -> None: + await self._send_event(payload, wait_until_ready=False) + + async def _handle_function_call_event(self, event: dict[str, Any]) -> None: + item = event.get("item") + source = item if isinstance(item, dict) else event + if isinstance(item, dict) and item.get("type") != "function_call": + return + call_id = str( + source.get("call_id") + or event.get("call_id") + or source.get("id") + or "" + ) + name = str( + source.get("name") + or event.get("name") + or self._function_names.get(call_id) + or "" + ) + if not name: + return + await self._tool_session.handle_call( + name=name, + call_id=call_id, + arguments=source.get("arguments", event.get("arguments")), + ) + self._function_names.pop(call_id, None) + + def _remember_function_call(self, event: dict[str, Any]) -> None: + item = event.get("item") + if not isinstance(item, dict) or item.get("type") != "function_call": + return + call_id = str(item.get("call_id") or item.get("id") or "") + name = str(item.get("name") or "") + if call_id and name: + self._function_names[call_id] = name + + def _resolve_fixed_speech(self) -> None: + completion = self._fixed_speech_completion + self._fixed_speech_completion = None + self._suppress_response_transcript = False + if completion is not None and not completion.done(): + completion.set_result(None) async def _delete_greeting_request(self) -> None: item_id = self._greeting_request_item_id diff --git a/backend/services/pipecat/realtime_tools.py b/backend/services/pipecat/realtime_tools.py new file mode 100644 index 0000000..1c1437c --- /dev/null +++ b/backend/services/pipecat/realtime_tools.py @@ -0,0 +1,128 @@ +"""Provider-neutral function calling for speech-to-speech sessions.""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any + +from loguru import logger + + +@dataclass(frozen=True) +class RealtimeTool: + """Small JSON-schema tool definition understood by both providers.""" + + name: str + description: str + properties: dict[str, Any] = field(default_factory=dict) + required: tuple[str, ...] = () + + def provider_schema(self) -> dict[str, Any]: + return { + "type": "function", + "name": self.name, + "description": self.description, + "parameters": { + "type": "object", + "properties": self.properties, + "required": list(self.required), + }, + } + + +@dataclass(frozen=True) +class RealtimeToolResult: + """Tool output plus whether the model should continue immediately.""" + + output: dict[str, Any] + continue_response: bool = True + after_output: Callable[[], Awaitable[None]] | None = None + + +RealtimeToolDispatcher = Callable[ + [str, dict[str, Any], str], Awaitable[RealtimeToolResult] +] +SendProviderEvent = Callable[[dict[str, Any]], Awaitable[None]] + + +class RealtimeToolSession: + """Serialize provider calls and answer every call id at most once.""" + + def __init__(self, send_event: SendProviderEvent) -> None: + self._send_event = send_event + self._dispatcher: RealtimeToolDispatcher | None = None + self._handled_call_ids: set[str] = set() + self._lock = asyncio.Lock() + + def set_dispatcher(self, dispatcher: RealtimeToolDispatcher | None) -> None: + self._dispatcher = dispatcher + + async def handle_call( + self, + *, + name: str, + call_id: str, + arguments: str | dict[str, Any] | None, + ) -> None: + if not call_id: + logger.warning("Realtime function call 缺少 call_id，已忽略") + return + async with self._lock: + if call_id in self._handled_call_ids: + return + self._handled_call_ids.add(call_id) + parsed = self._parse_arguments(arguments) + try: + if self._dispatcher is None: + result = RealtimeToolResult( + {"status": "error", "message": "当前会话未注册工具处理器"} + ) + else: + result = await self._dispatcher(name, parsed, call_id) + except Exception as exc: # noqa: BLE001 - return tool errors to provider + logger.exception(f"Realtime 工具 {name} 执行失败:{exc}") + result = RealtimeToolResult( + { + "status": "error", + "message": f"工具执行失败:{type(exc).__name__}", + } + ) + + await self._send_event( + { + "type": "conversation.item.create", + "item": { + "type": "function_call_output", + "call_id": call_id, + "output": json.dumps( + result.output, + ensure_ascii=False, + separators=(",", ":"), + ), + }, + } + ) + if result.after_output is not None: + await result.after_output() + if result.continue_response: + await self._send_event({"type": "response.create"}) + + def clear(self) -> None: + self._handled_call_ids.clear() + + @staticmethod + def _parse_arguments( + arguments: str | dict[str, Any] | None, + ) -> dict[str, Any]: + if isinstance(arguments, dict): + return dict(arguments) + if not arguments: + return {} + try: + parsed = json.loads(arguments) + except (TypeError, json.JSONDecodeError): + return {} + return dict(parsed) if isinstance(parsed, dict) else {} diff --git a/backend/services/pipecat/stepfun_realtime.py b/backend/services/pipecat/stepfun_realtime.py index adf49f2..6b8819c 100644 --- a/backend/services/pipecat/stepfun_realtime.py +++ b/backend/services/pipecat/stepfun_realtime.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import base64 import json +from collections.abc import Awaitable, Callable from typing import Any from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from uuid import uuid4 @@ -28,7 +29,14 @@ from pipecat.utils.time import time_now_iso8601 from websockets.asyncio.client import connect as websocket_connect from websockets.protocol import State +from services.pipecat.realtime_tools import ( + RealtimeTool, + RealtimeToolDispatcher, + RealtimeToolSession, +) + DEFAULT_STEPFUN_REALTIME_URL = "wss://api.stepfun.com/v1/realtime" +SpeechStartedHandler = Callable[[], Awaitable[None]] class StepFunRealtimeService(AIService): @@ -68,6 +76,12 @@ class StepFunRealtimeService(AIService): self._assistant_turn_id: str | None = None self._assistant_text = "" self._assistant_timestamp = "" + self._tools: list[RealtimeTool] = [] + self._tool_session = RealtimeToolSession(self._send_tool_event) + self._fixed_speech_completion: asyncio.Future[None] | None = None + self._suppress_response_transcript = False + self._speech_started_handler: SpeechStartedHandler | None = None + self._function_names: dict[str, str] = {} async def start(self, frame: StartFrame) -> None: await super().start(frame) @@ -120,6 +134,7 @@ class StepFunRealtimeService(AIService): if isinstance(frame, InterruptionFrame): await self._send_event({"type": "response.cancel"}, wait_until_ready=False) await self._finish_assistant_text(interrupted=True) + self._resolve_fixed_speech() await self.push_frame(frame, direction) @@ -140,12 +155,35 @@ class StepFunRealtimeService(AIService): async def interrupt(self) -> None: await self._send_event({"type": "response.cancel"}, wait_until_ready=False) await self._finish_assistant_text(interrupted=True) + self._resolve_fixed_speech() await self.broadcast_interruption() + async def request_response(self) -> None: + await self._send_event({"type": "response.create"}) + + def set_speech_started_handler( + self, + handler: SpeechStartedHandler | None, + ) -> None: + self._speech_started_handler = handler + async def speak(self, text: str) -> None: """Ask the realtime model to voice a fixed greeting.""" + await self.speak_fixed(text, suppress_transcript=False) + + async def speak_fixed( + self, + text: str, + *, + suppress_transcript: bool = True, + ) -> Awaitable[None] | None: + """Speak configured text and expose the provider response boundary.""" if not text: - return + return None + completion = asyncio.get_running_loop().create_future() + self._resolve_fixed_speech() + self._fixed_speech_completion = completion + self._suppress_response_transcript = suppress_transcript await self._send_event( { "type": "response.create", @@ -154,6 +192,7 @@ class StepFunRealtimeService(AIService): }, } ) + return completion async def _connect(self) -> None: if self._websocket and self._websocket.state is State.OPEN: @@ -186,6 +225,9 @@ class StepFunRealtimeService(AIService): websocket = self._websocket self._websocket = None self._session_ready.clear() + self._tool_session.clear() + self._function_names.clear() + self._resolve_fixed_speech() if websocket and websocket.state is State.OPEN: try: await websocket.close() @@ -240,10 +282,11 @@ class StepFunRealtimeService(AIService): ) ) elif event_type in {"response.audio_transcript.delta", "response.text.delta"}: - await self._append_assistant_text(str(event.get("delta") or "")) + if not self._suppress_response_transcript: + await self._append_assistant_text(str(event.get("delta") or "")) elif event_type in {"response.audio_transcript.done", "response.text.done"}: transcript = str(event.get("transcript") or event.get("text") or "") - if transcript: + if transcript and not self._suppress_response_transcript: if not self._assistant_turn_id: await self._append_assistant_text(transcript) else: @@ -254,6 +297,9 @@ class StepFunRealtimeService(AIService): elif event_type == "input_audio_buffer.speech_started": await self._send_event({"type": "response.cancel"}, wait_until_ready=False) await self.broadcast_interruption() + self._resolve_fixed_speech() + if self._speech_started_handler is not None: + await self._speech_started_handler() elif event_type == "response.done": response = event.get("response") interrupted = isinstance(response, dict) and response.get("status") in { @@ -262,11 +308,20 @@ class StepFunRealtimeService(AIService): "interrupted", } await self._finish_assistant_text(interrupted=interrupted) + self._resolve_fixed_speech() + elif event_type == "response.output_item.added": + self._remember_function_call(event) + elif event_type in { + "response.function_call_arguments.done", + "response.output_item.done", + }: + await self._handle_function_call_event(event) elif event_type == "error": error = event.get("error") message = error.get("message") if isinstance(error, dict) else str(error) if "cancel" not in str(message).lower(): await self.push_error(f"StepFun Realtime error: {message}") + self._resolve_fixed_speech() async def _send_session_update(self) -> None: await self._send_event( @@ -284,6 +339,8 @@ class StepFunRealtimeService(AIService): "silence_duration_ms": self._silence_duration_ms, "energy_awakeness_threshold": self._energy_awakeness_threshold, }, + "tools": [tool.provider_schema() for tool in self._tools], + "tool_choice": "auto", }, }, wait_until_ready=False, @@ -293,7 +350,85 @@ class StepFunRealtimeService(AIService): """Refresh model instructions without rebuilding the realtime session.""" self._instructions = instructions if self._session_ready.is_set(): - await self._send_session_update() + await self._send_event( + { + "type": "session.update", + "session": {"instructions": instructions}, + }, + wait_until_ready=False, + ) + + async def update_session( + self, + instructions: str, + tools: list[RealtimeTool], + ) -> None: + """Atomically replace the active Workflow prompt and tool catalog.""" + self._instructions = instructions + self._tools = list(tools) + if self._session_ready.is_set(): + await self._send_event( + { + "type": "session.update", + "session": { + "instructions": instructions, + "tools": [tool.provider_schema() for tool in tools], + "tool_choice": "auto", + }, + }, + wait_until_ready=False, + ) + + def set_tool_dispatcher( + self, + dispatcher: RealtimeToolDispatcher | None, + ) -> None: + self._tool_session.set_dispatcher(dispatcher) + + async def _send_tool_event(self, payload: dict[str, Any]) -> None: + await self._send_event(payload, wait_until_ready=False) + + async def _handle_function_call_event(self, event: dict[str, Any]) -> None: + item = event.get("item") + source = item if isinstance(item, dict) else event + if isinstance(item, dict) and item.get("type") != "function_call": + return + call_id = str( + source.get("call_id") + or event.get("call_id") + or source.get("id") + or "" + ) + name = str( + source.get("name") + or event.get("name") + or self._function_names.get(call_id) + or "" + ) + if not name: + return + await self._tool_session.handle_call( + name=name, + call_id=call_id, + arguments=source.get("arguments", event.get("arguments")), + ) + self._function_names.pop(call_id, None) + + def _remember_function_call(self, event: dict[str, Any]) -> None: + item = event.get("item") + if not isinstance(item, dict) or item.get("type") != "function_call": + return + call_id = str(item.get("call_id") or item.get("id") or "") + name = str(item.get("name") or "") + if call_id and name: + self._function_names[call_id] = name + + def _resolve_fixed_speech(self) -> None: + completion = self._fixed_speech_completion + self._fixed_speech_completion = None + self._suppress_response_transcript = False + if completion is not None and not completion.done(): + completion.set_result(None) async def _send_event( self, payload: dict[str, Any], *, wait_until_ready: bool = True diff --git a/backend/services/workflow/agent.py b/backend/services/workflow/agent.py index 09df9b1..ff1dd17 100644 --- a/backend/services/workflow/agent.py +++ b/backend/services/workflow/agent.py @@ -20,6 +20,11 @@ AGENT_STAGE_INSTRUCTION = ( "工作流路由已在用户一轮输入结束时完成。只执行当前阶段任务，" "不要自行解释、模拟或宣布节点切换。" ) +EDGE_TOOL_STAGE_INSTRUCTION = ( + "只执行当前阶段任务。若某个工作流跳转工具的条件已经明确满足，" + "必须只调用一个对应的 goto 工具，不要口头宣布、解释或模拟节点跳转。" + "若没有条件满足，则正常回答用户并停留在当前阶段。" +) class WorkflowAgentStage: @@ -43,9 +48,14 @@ class WorkflowAgentStage: stage = self._engine.agent_stage_config(node_id) if stage.vision_enabled: stage_prompt = f"{stage_prompt}\n\n[视觉能力]\n{VISION_SYSTEM_HINT}" + instruction = ( + EDGE_TOOL_STAGE_INSTRUCTION + if self._engine.llm_routing_mode() == "edge_tool" + else AGENT_STAGE_INSTRUCTION + ) return ( f"{stage_prompt}\n\n[工作流执行规则]\n" - f"{AGENT_STAGE_INSTRUCTION}" + f"{instruction}" ) async def refresh_prompt(self, node_id: str) -> None: diff --git a/backend/services/workflow/realtime.py b/backend/services/workflow/realtime.py new file mode 100644 index 0000000..5d15a3b --- /dev/null +++ b/backend/services/workflow/realtime.py @@ -0,0 +1,918 @@ +"""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, + ) diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 1a19066..3672117 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -117,6 +117,12 @@ class WorkflowEngine: def global_prompt(self) -> str: return str(self.settings.get("globalPrompt") or "").strip() + def runtime_mode(self) -> str: + return str(self.settings.get("runtimeMode") or "pipeline") + + def llm_routing_mode(self) -> str: + return str(self.settings.get("llmRoutingMode") or "llm_router") + def inherits_global_config(self, node_id: str) -> bool: """Return the Agent's explicit configuration scope, defaulting to global.""" return bool(self.data(node_id).get("inheritGlobalConfig", True)) @@ -266,8 +272,17 @@ class WorkflowEngine: return default if include_default else None def llm_edges(self, node_id: str) -> list[dict]: + """Edges considered by the compatibility dedicated router.""" return [ edge for edge in self.outgoing(node_id) if self.edge_mode(edge) in {"llm", "always"} ] + + def edge_tool_edges(self, node_id: str) -> list[dict]: + """Natural-language transitions exposed to the active Agent.""" + return [ + edge + for edge in self.outgoing(node_id) + if self.edge_mode(edge) == "llm" + ] diff --git a/frontend/src/components/assistant-editor/workflow-page.tsx b/frontend/src/components/assistant-editor/workflow-page.tsx index 8d7f362..12df9fb 100644 --- a/frontend/src/components/assistant-editor/workflow-page.tsx +++ b/frontend/src/components/assistant-editor/workflow-page.tsx @@ -36,6 +36,7 @@ type WorkflowPageProps = { dirty: boolean; saveError: string | null; modelOptions: { + realtime: ResourceOption[]; llm: ResourceOption[]; asr: ResourceOption[]; tts: ResourceOption[]; diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 4eb7dbe..94d609e 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -711,6 +711,13 @@ export function AssistantPage(props: AssistantPageProps) { ? (assistant.graph as WorkflowGraph) : defaultGraph(); const wfSettings: WorkflowSettings = { + runtimeMode: + graph.settings?.runtimeMode ?? assistant.runtimeMode ?? "pipeline", + llmRoutingMode: + graph.settings?.llmRoutingMode ?? "llm_router", + realtime: + graph.settings?.defaultRealtimeResourceId || + assistant.modelResourceIds.Realtime, llm: graph.settings?.defaultLlmResourceId || assistant.modelResourceIds.LLM, @@ -811,11 +818,15 @@ export function AssistantPage(props: AssistantPageProps) { baseUpsert({ name: workflowName.trim(), type: "workflow", + runtimeMode: workflowSettings.runtimeMode, enableInterrupt: workflowSettings.allowInterrupt, turnConfig: workflowSettings.turnConfig, visionEnabled: workflowUsesVision(workflowGraph), visionModelResourceId: null, modelResourceIds: { + ...(workflowSettings.realtime + ? { Realtime: workflowSettings.realtime } + : {}), ...(workflowSettings.llm ? { LLM: workflowSettings.llm } : {}), ...(workflowSettings.asr ? { ASR: workflowSettings.asr } : {}), ...(workflowSettings.tts ? { TTS: workflowSettings.tts } : {}), @@ -1294,6 +1305,7 @@ export function AssistantPage(props: AssistantPageProps) { dirty={dirty} saveError={saveError} modelOptions={{ + realtime: credOptions("Realtime"), llm: credOptions("LLM"), asr: credOptions("ASR"), tts: credOptions("TTS"), diff --git a/frontend/src/components/workflow/WorkflowCanvas.tsx b/frontend/src/components/workflow/WorkflowCanvas.tsx index 06d54e2..ebe6f5f 100644 --- a/frontend/src/components/workflow/WorkflowCanvas.tsx +++ b/frontend/src/components/workflow/WorkflowCanvas.tsx @@ -119,7 +119,10 @@ function fromFlow(nodes: Node[], edges: Edge[]): WorkflowGraph { return { specVersion: 3, settings: { + runtimeMode: "pipeline", + llmRoutingMode: "edge_tool", globalPrompt: "", + defaultRealtimeResourceId: "", defaultLlmResourceId: "", defaultAsrResourceId: "", defaultTtsResourceId: "", @@ -809,6 +812,7 @@ export function WorkflowCanvas({ editingEdge.source, editingEdge.id, )} + llmRoutingMode={settings.llmRoutingMode} onChange={(patch) => updateEdgeData(editingEdge.id, patch) } diff --git a/frontend/src/components/workflow/graph-settings.ts b/frontend/src/components/workflow/graph-settings.ts index 8d09142..8cda4b7 100644 --- a/frontend/src/components/workflow/graph-settings.ts +++ b/frontend/src/components/workflow/graph-settings.ts @@ -6,6 +6,9 @@ export function settingsFromWorkflowGraph( graph: WorkflowGraph, ): WorkflowSettings { return { + runtimeMode: graph.settings.runtimeMode ?? "pipeline", + llmRoutingMode: graph.settings.llmRoutingMode ?? "llm_router", + realtime: graph.settings.defaultRealtimeResourceId, globalPrompt: graph.settings.globalPrompt, llm: graph.settings.defaultLlmResourceId, asr: graph.settings.defaultAsrResourceId, @@ -32,7 +35,10 @@ export function workflowGraphWithSettings( return { ...graph, settings: { + runtimeMode: settings.runtimeMode, + llmRoutingMode: settings.llmRoutingMode, globalPrompt: settings.globalPrompt, + defaultRealtimeResourceId: settings.realtime ?? "", defaultLlmResourceId: settings.llm ?? "", defaultAsrResourceId: settings.asr ?? "", defaultTtsResourceId: settings.tts ?? "", diff --git a/frontend/src/components/workflow/panels/ActionNodePanel.tsx b/frontend/src/components/workflow/panels/ActionNodePanel.tsx index 3fdb2fd..30b7d25 100644 --- a/frontend/src/components/workflow/panels/ActionNodePanel.tsx +++ b/frontend/src/components/workflow/panels/ActionNodePanel.tsx @@ -23,6 +23,7 @@ type ActionNodePanelProps = { setArgumentsJson: (value: string) => void; setAssignmentsJson: (value: string) => void; commitActionJson: (argumentsValue: string, assignmentsValue: string) => void; + runtimeMode: "pipeline" | "realtime"; }; export function ActionNodePanel({ @@ -35,6 +36,7 @@ export function ActionNodePanel({ setArgumentsJson, setAssignmentsJson, commitActionJson, + runtimeMode, }: ActionNodePanelProps) { const resultAssignmentMode = actionResultAssignmentMode(draft); const userInputPolicy = actionUserInputPolicy(draft); @@ -113,14 +115,23 @@ export function ActionNodePanel({ label="执行期间用户输入" value={userInputPolicy} options={[ - { value: "queue", label: "允许输入并排队（默认）" }, + ...(runtimeMode === "pipeline" + ? [{ value: "queue", label: "允许输入并排队（默认）" }] + : []), { value: "block", label: "暂时禁止输入" }, ]} - onChange={(value) => set("userInputPolicy", value || "queue")} + onChange={(value) => + set( + "userInputPolicy", + value || (runtimeMode === "realtime" ? "block" : "queue"), + ) + } allowNone={false} />

- {userInputPolicy === "queue" + {runtimeMode === "realtime" + ? "Realtime 模式必须阻止 Action 执行期间的输入，避免端到端模型提前生成回复。" + : userInputPolicy === "queue" ? "用户输入会保留，Action 完成并进入后续节点后再处理。" : "Action 执行期间忽略新的语音、文本和图片输入。"}

diff --git a/frontend/src/components/workflow/panels/AgentNodePanel.tsx b/frontend/src/components/workflow/panels/AgentNodePanel.tsx index 2ab1bff..aafa34a 100644 --- a/frontend/src/components/workflow/panels/AgentNodePanel.tsx +++ b/frontend/src/components/workflow/panels/AgentNodePanel.tsx @@ -53,6 +53,7 @@ export function AgentNodePanel({ dynamicVariableOptions: ModelOption[]; }) { const inheritsGlobal = draft.inheritGlobalConfig !== false; + const isRealtime = workflowSettings.runtimeMode === "realtime"; const knowledgeConfig: KnowledgeRetrievalConfig = { mode: draft.knowledgeMode === "on_demand" ? "on_demand" : "automatic", @@ -70,17 +71,24 @@ export function AgentNodePanel({ } setPatch({ inheritGlobalConfig: false, - llmResourceId: - (draft.llmResourceId as string) || workflowSettings.llm || "", - asrResourceId: - (draft.asrResourceId as string) || workflowSettings.asr || "", - ttsResourceId: - (draft.ttsResourceId as string) || workflowSettings.tts || "", - visionEnabled: - draft.visionEnabled ?? workflowSettings.visionEnabled, - visionModelResourceId: - (draft.visionModelResourceId as string) || - workflowSettings.visionModelResourceId, + ...(!isRealtime + ? { + llmResourceId: + (draft.llmResourceId as string) || workflowSettings.llm || "", + asrResourceId: + (draft.asrResourceId as string) || workflowSettings.asr || "", + ttsResourceId: + (draft.ttsResourceId as string) || workflowSettings.tts || "", + visionEnabled: + draft.visionEnabled ?? workflowSettings.visionEnabled, + visionModelResourceId: + (draft.visionModelResourceId as string) || + workflowSettings.visionModelResourceId, + enableInterrupt: + draft.enableInterrupt ?? workflowSettings.allowInterrupt, + turnConfig: agentTurnConfig, + } + : {}), toolIds: draft.toolIds?.length ? draft.toolIds : workflowSettings.toolIds, @@ -98,9 +106,6 @@ export function AgentNodePanel({ knowledgeScoreThreshold: draft.knowledgeScoreThreshold ?? workflowSettings.knowledgeRetrievalConfig.scoreThreshold, - enableInterrupt: - draft.enableInterrupt ?? workflowSettings.allowInterrupt, - turnConfig: agentTurnConfig, }); }; const selectedToolIds = inheritsGlobal @@ -133,9 +138,13 @@ export function AgentNodePanel({ : []), ...(!inheritsGlobal ? [ - { id: "models", label: "模型与语音" }, + ...(!isRealtime + ? [{ id: "models", label: "模型与语音" }] + : []), { id: "capabilities", label: "知识与工具" }, - { id: "interaction", label: "交互策略" }, + ...(!isRealtime + ? [{ id: "interaction", label: "交互策略" }] + : []), ] : []), ]} @@ -248,7 +257,7 @@ export function AgentNodePanel({ {!inheritsGlobal && ( <> - + {!isRealtime ? } title="模型与语音" @@ -283,10 +292,10 @@ export function AgentNodePanel({ noneLabel="请选择语音合成" /> - + : null} - set("visionModelResourceId", visionModelResourceId) } - /> + /> : null} } @@ -332,6 +341,13 @@ export function AgentNodePanel({ onChange={(value) => set("knowledgeBaseId", value || "")} noneLabel="无" /> + {isRealtime && + draft.knowledgeBaseId && + knowledgeConfig.mode !== "on_demand" ? ( +

+ Realtime 仅支持按需知识库，请将检索模式切换为按需调用。 +

+ ) : null} - + {!isRealtime ? } title="交互策略" @@ -365,7 +381,7 @@ export function AgentNodePanel({ onConfigChange={(turnConfig) => set("turnConfig", turnConfig)} /> - + : null} )} diff --git a/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx b/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx index 9d16495..a2503f8 100644 --- a/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx +++ b/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx @@ -33,12 +33,14 @@ export function EdgeSettingsPanel({ sourceType, isOnlyOutgoing, hasOtherDefaultPath, + llmRoutingMode, onChange, }: { edge: Edge; sourceType?: string; isOnlyOutgoing: boolean; hasOtherDefaultPath: boolean; + llmRoutingMode: "llm_router" | "edge_tool"; onChange: (patch: Partial) => void; }) { const data = (edge.data ?? { mode: "always", priority: 10 }) as WorkflowEdgeData; @@ -110,11 +112,17 @@ export function EdgeSettingsPanel({ value: "always", label: "默认路径", disabled: + (llmRoutingMode === "edge_tool" && sourceType === "agent") || mode !== "always" && (hasOtherDefaultPath || (sourceType === "agent" && isOnlyOutgoing)), }, - { value: "llm", label: "大模型判断" }, + { + value: "llm", + label: "大模型判断", + disabled: + llmRoutingMode === "edge_tool" && sourceType !== "agent", + }, { value: "expression", label: "表达式" }, ]} onChange={(value) => { @@ -135,6 +143,13 @@ export function EdgeSettingsPanel({ Agent 不能只有默认路径；请改为条件路径，或删除连接以持续对话。 )} + {llmRoutingMode === "edge_tool" && + ((sourceType === "agent" && mode === "always") || + (sourceType !== "agent" && mode === "llm")) ? ( + + 边工具模式仅允许 Agent 发出大模型判断边；自动节点请使用表达式或默认路径。 + + ) : null} diff --git a/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx b/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx index 74fc3f7..8dd47ab 100644 --- a/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx +++ b/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx @@ -1,6 +1,7 @@ "use client"; import { + Activity, Brain, Database, MessageSquareText, @@ -42,12 +43,65 @@ export function GlobalSettingsPanel({ + + } + title="运行与路由" + description="选择语音运行管线，以及大模型判断边的执行方式" + > + + onSettingsChange({ + ...settings, + runtimeMode: + runtimeMode === "realtime" ? "realtime" : "pipeline", + ...(runtimeMode === "realtime" + ? { llmRoutingMode: "edge_tool" } + : {}), + }) + } + allowNone={false} + /> + + onSettingsChange({ + ...settings, + llmRoutingMode: + llmRoutingMode === "llm_router" ? "llm_router" : "edge_tool", + }) + } + allowNone={false} + /> +

+ 此设置仅影响大模型判断边。边工具会将其注册为 function + tool；表达式始终由服务端直接判断。 +

+ + + } @@ -72,39 +126,56 @@ export function GlobalSettingsPanel({ } title="模型与语音" - description="继承全局配置的 Agent 共用的推理、语音识别和语音合成资源" + description={ + settings.runtimeMode === "realtime" + ? "整个工作流共用一个端到端语音模型，节点切换只更新提示词和工具" + : "继承全局配置的 Agent 共用的推理、语音识别和语音合成资源" + } > - - onSettingsChange({ - ...settings, - llm, - ...(settings.visionModelResourceId === llm - ? { visionModelResourceId: "" } - : {}), - }) - } - /> - onSettingsChange({ ...settings, asr })} - /> - onSettingsChange({ ...settings, tts })} - /> + {settings.runtimeMode === "realtime" ? ( + + onSettingsChange({ ...settings, realtime }) + } + /> + ) : ( + <> + + onSettingsChange({ + ...settings, + llm, + ...(settings.visionModelResourceId === llm + ? { visionModelResourceId: "" } + : {}), + }) + } + /> + onSettingsChange({ ...settings, asr })} + /> + onSettingsChange({ ...settings, tts })} + /> + + )} - onSettingsChange({ ...settings, visionModelResourceId }) } - /> + /> : null} } @@ -150,6 +221,13 @@ export function GlobalSettingsPanel({ } noneLabel="无" /> + {settings.runtimeMode === "realtime" && + settings.knowledgeBaseId && + settings.knowledgeRetrievalConfig.mode !== "on_demand" ? ( +

+ Realtime 仅支持按需知识库，请在上方检索设置中切换为按需调用。 +

+ ) : null} ); @@ -262,6 +263,7 @@ export function NodeSettingsPanel({ setArgumentsJson={setArgumentsJson} setAssignmentsJson={setAssignmentsJson} commitActionJson={commitActionJson} + runtimeMode={workflowSettings.runtimeMode} /> )} @@ -322,6 +324,7 @@ function WorkflowNodePanelForm({ setAssignmentsJson, commitActionJson, dynamicVariableOptions, + runtimeMode, }: { spec: RuntimeNodeSpec; draft: WorkflowNodeData; @@ -335,6 +338,7 @@ function WorkflowNodePanelForm({ setAssignmentsJson: (value: string) => void; commitActionJson: (argumentsValue: string, assignmentsValue: string) => void; dynamicVariableOptions: ModelOption[]; + runtimeMode: "pipeline" | "realtime"; }) { return (
@@ -350,6 +354,7 @@ function WorkflowNodePanelForm({ setArgumentsJson={setArgumentsJson} setAssignmentsJson={setAssignmentsJson} commitActionJson={commitActionJson} + runtimeMode={runtimeMode} /> )} diff --git a/frontend/src/components/workflow/specs.ts b/frontend/src/components/workflow/specs.ts index 6b72def..ceeffe8 100644 --- a/frontend/src/components/workflow/specs.ts +++ b/frontend/src/components/workflow/specs.ts @@ -24,6 +24,8 @@ export type MessageCompletionPolicy = export type ActionResultAssignmentMode = "inherit" | "override" | "none"; export type ActionUserInputPolicy = "queue" | "block"; export type EdgeMode = "llm" | "expression" | "always"; +export type WorkflowRuntimeMode = "pipeline" | "realtime"; +export type WorkflowLlmRoutingMode = "llm_router" | "edge_tool"; export type ExpressionOperator = | "eq" | "neq" @@ -202,7 +204,10 @@ export type NodeSpecMap = Record; export type WorkflowGraph = { specVersion: 3; settings: { + runtimeMode: WorkflowRuntimeMode; + llmRoutingMode: WorkflowLlmRoutingMode; globalPrompt: string; + defaultRealtimeResourceId: string; defaultLlmResourceId: string; defaultAsrResourceId: string; defaultTtsResourceId: string; @@ -235,8 +240,11 @@ export function defaultGraph(): WorkflowGraph { return { specVersion: 3, settings: { + runtimeMode: "pipeline", + llmRoutingMode: "edge_tool", globalPrompt: "你是一个友好、专业的语音助手。请使用简短、自然、适合口语表达的句子。", + defaultRealtimeResourceId: "", defaultLlmResourceId: "", defaultAsrResourceId: "", defaultTtsResourceId: "", diff --git a/frontend/src/components/workflow/types.ts b/frontend/src/components/workflow/types.ts index 8f12575..ae62768 100644 --- a/frontend/src/components/workflow/types.ts +++ b/frontend/src/components/workflow/types.ts @@ -10,6 +10,9 @@ import type { import type { WorkflowGraph } from "./specs"; export type WorkflowSettings = { + runtimeMode: "pipeline" | "realtime"; + llmRoutingMode: "llm_router" | "edge_tool"; + realtime?: string; llm?: string; asr?: string; tts?: string; @@ -37,6 +40,7 @@ export type WorkflowEditorProps = { settings: WorkflowSettings; onSettingsChange: (settings: WorkflowSettings) => void; modelOptions: { + realtime: ModelOption[]; llm: ModelOption[]; asr: ModelOption[]; tts: ModelOption[];