From 162a3d8beceb1e72cfd6f23e7b8ab757bf04412d Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 17 Jul 2026 22:01:42 +0800 Subject: [PATCH] Refactor workflow routing and greeting management in Brain classes - Update WorkflowBrain to handle greeting playback more effectively, ensuring that the initial greeting completes before transitioning to the first node. - Introduce new methods for managing greeting states and conditions, enhancing the interaction flow for user turns. - Refactor WorkflowLLMRouter to improve routing logic and ensure proper handling of conditional paths. - Enhance tests to verify the correct behavior of greeting management and routing under various scenarios, including waiting for audio playback to finish. - Update frontend components to reflect changes in edge handling and improve user experience in workflow configurations. --- backend/services/brains/base.py | 16 +- backend/services/brains/workflow_brain.py | 213 +++++++---- backend/services/node_specs.py | 27 +- backend/services/pipecat/pipeline_events.py | 38 +- backend/services/workflow_engine.py | 14 + backend/services/workflow_router.py | 26 +- backend/tests/test_brains.py | 336 ++++++++++++++++++ backend/tests/test_pipeline_events.py | 65 +++- backend/tests/test_workflow_v3.py | 105 ++++++ .../src/components/workflow/ConditionEdge.tsx | 6 +- .../components/workflow/WorkflowCanvas.tsx | 49 ++- .../src/components/workflow/edge-rules.ts | 42 +++ .../workflow/panels/EdgeSettingsPanel.tsx | 27 +- .../components/workflow/panels/controls.tsx | 7 +- frontend/src/components/workflow/types.ts | 2 +- 15 files changed, 826 insertions(+), 147 deletions(-) create mode 100644 frontend/src/components/workflow/edge-rules.ts diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index deba421..e0b1397 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -101,8 +101,16 @@ class BaseBrain: async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: """Register tools and initialize per-call orchestration.""" - async def on_connected(self) -> None: - """Handle a connected client after the common greeting is queued.""" + async def on_connected(self, *, greeting_pending: bool = False) -> None: + """Handle a connected client before an optional greeting is played. + + ``greeting_pending`` lets an orchestration brain prepare its initial + state without starting node-entry speech while the shared greeting is + still playing. + """ + + async def on_greeting_finished(self) -> None: + """Continue startup after the shared greeting has actually played.""" def prepare_greeting_context( self, @@ -166,7 +174,9 @@ class Brain(Protocol): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: ... - async def on_connected(self) -> None: ... + async def on_connected(self, *, greeting_pending: bool = False) -> None: ... + + async def on_greeting_finished(self) -> None: ... def prepare_greeting_context( self, diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index cabde71..fa81b84 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -31,7 +31,7 @@ from services.knowledge import search as search_knowledge from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.workflow_engine import WorkflowEngine -from services.workflow_router import STAY_ON_CURRENT_AGENT, WorkflowLLMRouter +from services.workflow_router import STAY_ON_CURRENT_NODE, WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 @@ -71,6 +71,7 @@ class WorkflowBrain(BaseBrain): self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) self._ended = False self._greeting_context_message: dict[str, str] | None = None + self._startup_waiting_for_greeting = False self._client_ready = False self._pending_visible_speech_events: list[dict[str, Any]] = [] @@ -95,6 +96,7 @@ class WorkflowBrain(BaseBrain): self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) self._greeting_context_message = None + self._startup_waiting_for_greeting = False self._client_ready = False self._pending_visible_speech_events = [] self._manager = FlowManager( @@ -115,28 +117,58 @@ class WorkflowBrain(BaseBrain): self._greeting_context_message = deepcopy(message) if message else None return message - async def on_connected(self) -> None: + async def on_connected(self, *, greeting_pending: bool = False) -> None: await self._emit_node_active(self._engine.start_id) await self._emit_variables( reason="initialized", node_id=self._engine.start_id, ) - edge = self._engine.deterministic_edge( + if self._manager is None: + raise RuntimeError("Workflow FlowManager 尚未初始化") + + self._startup_waiting_for_greeting = greeting_pending + if greeting_pending: + # Keep the Workflow on Start until the transport confirms that the + # shared greeting has finished. This prevents an initial Agent's + # fixed speech (or generated reply) from racing the greeting. + await self._manager.initialize( + self._passive_node_config(self._engine.start_id) + ) + logger.info("工作流等待 Start 开场白播放完毕") + return + + node_config = await self._initial_node_config() + await self._manager.initialize(node_config) + logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") + + async def on_greeting_finished(self) -> None: + """Enter the first node only after Start's greeting reaches playback end.""" + if not self._startup_waiting_for_greeting or self._ended: + return + self._startup_waiting_for_greeting = False + manager = self._require_manager() + if manager.current_node != self._engine.start_id: + return + + node_config = await self._initial_node_config() + if node_config.get("name") == self._engine.start_id: + return + await manager.set_node_from_config(node_config) + logger.info(f"Start 开场白结束,进入节点: {manager.current_node}") + + async def _initial_node_config(self) -> NodeConfig: + """Resolve the immediate path from Start without evaluating LLM edges.""" + # Start LLM conditions need an actual user turn. Expression-only and + # default-only starts may still advance immediately at connection time. + edge = await self._select_edge( self._engine.start_id, - self._store, - include_default=True, + evaluate_llm=False, ) - if not edge and self._engine.has_outgoing(self._engine.start_id): - raise RuntimeError("Start 初始化后没有命中的表达式边或默认边") - node_config = ( + return ( await self._follow_edge(edge) if edge else self._passive_node_config(self._engine.start_id) ) - if self._manager is None: - raise RuntimeError("Workflow FlowManager 尚未初始化") - await self._manager.initialize(node_config) - logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") async def on_client_ready(self) -> None: """Replay state that may have been emitted before WebRTC data was ready.""" @@ -163,26 +195,63 @@ class WorkflowBrain(BaseBrain): self._store.record("user", content) async def on_user_turn_end(self, content: str) -> bool: - """Route a complete user turn before any Agent is allowed to reply.""" + """Route a complete user turn before the active stage may reply.""" if not content or self._ended: return True self.record_user_message(content) manager = self._require_manager() current = manager.current_node - if not current or self._engine.node_type(current) != "agent": + if not current: return True - edge = self._engine.deterministic_edge( - current, + edge = await self._select_edge(current) + + if edge and manager.current_node == current: + next_config = await self._follow_edge( + edge, + triggering_user_text=content, + ) + await manager.set_node_from_config(next_config) + next_node = str(next_config.get("name") or "") + if ( + self._engine.node_type(next_node) == "agent" + and self._engine.data(next_node).get("entryMode", "wait_user") + == "wait_user" + ): + await self._require_runtime().queue_frame(LLMRunFrame()) + return True + + if self._engine.node_type(current) != "agent": + # Start/Action/Handoff have no conversational LLM of their own. + # Keep waiting so a later user turn may satisfy another condition. + return True + + # The incoming LLMContextFrame is intentionally suppressed by the + # pipeline router. Queue prompt refresh + inference in this order so + # this user turn is answered with the current Agent's latest variables. + await self._refresh_agent_prompt(current) + await self._require_runtime().queue_frame(LLMRunFrame()) + return True + + async def _select_edge( + self, + node_id: str, + *, + evaluate_llm: bool = True, + ) -> dict | None: + """Resolve conditional paths by priority, then use the default path.""" + expression_edge = self._engine.deterministic_edge( + node_id, self._store, include_default=False, ) - outgoing = self._engine.outgoing(current) - llm_edges = [ + outgoing = self._engine.outgoing(node_id) + all_llm_edges = [ candidate for candidate in outgoing if self._engine.edge_mode(candidate) == "llm" ] + llm_edges = all_llm_edges default_edge = next( ( candidate @@ -192,45 +261,51 @@ class WorkflowBrain(BaseBrain): None, ) - if edge is None and llm_edges: - selected = await self._router_for_node(current).select_edge( - node_name=self._engine.name(current), - node_prompt=self._engine.prompt_for(current, self._store), - edges=llm_edges, - history=self._store.history, - variables={ - key: value - for key, value in self._store.values.items() - if not key.startswith("system__") - }, - edge_name=self._engine.edge_fn_name, - edge_description=self._engine.edge_description, - ) - if selected and selected != STAY_ON_CURRENT_AGENT: - edge = next( - ( - candidate - for candidate in llm_edges - if self._engine.edge_fn_name(candidate) == selected - ), - None, - ) - elif selected == STAY_ON_CURRENT_AGENT: - edge = default_edge - elif edge is None and not llm_edges: - edge = default_edge + # A matching expression is a deterministic priority boundary. Only LLM + # conditions before it may win; later conditions must not bypass it. + if expression_edge: + expression_index = outgoing.index(expression_edge) + llm_edges = [ + candidate + for candidate in llm_edges + if outgoing.index(candidate) < expression_index + ] - if edge and manager.current_node == current: - next_config = await self._follow_edge(edge) - await manager.set_node_from_config(next_config) - return True + if not evaluate_llm: + if expression_edge and not llm_edges: + return expression_edge + if all_llm_edges: + return None + return default_edge - # The incoming LLMContextFrame is intentionally suppressed by the - # pipeline router. Queue prompt refresh + inference in this order so - # this user turn is answered with the current Agent's latest variables. - await self._refresh_agent_prompt(current) - await self._require_runtime().queue_frame(LLMRunFrame()) - return True + if not llm_edges: + return expression_edge or default_edge + + selected = await self._router_for_node(node_id).select_edge( + node_name=self._engine.name(node_id), + node_prompt=self._engine.routing_prompt(node_id, self._store), + edges=llm_edges, + history=self._store.history, + variables={ + key: value + for key, value in self._store.values.items() + if not key.startswith("system__") + }, + edge_name=self._engine.edge_fn_name, + edge_description=self._engine.edge_description, + ) + if selected == STAY_ON_CURRENT_NODE: + return expression_edge or default_edge + if not selected: + return expression_edge + return next( + ( + candidate + for candidate in llm_edges + if self._engine.edge_fn_name(candidate) == selected + ), + None, + ) async def on_assistant_text_end( self, @@ -481,7 +556,12 @@ class WorkflowBrain(BaseBrain): cancel_on_interruption=True, ) - async def _follow_edge(self, edge: dict) -> NodeConfig: + async def _follow_edge( + self, + edge: dict, + *, + triggering_user_text: str = "", + ) -> NodeConfig: leading_messages: list[dict[str, str]] = [] speech = self._engine.edge_transition_speech(edge) if speech: @@ -498,6 +578,7 @@ class WorkflowBrain(BaseBrain): return await self._resolve_path( str(edge.get("target") or ""), leading_messages=leading_messages, + triggering_user_text=triggering_user_text, ) async def _resolve_path( @@ -505,13 +586,23 @@ class WorkflowBrain(BaseBrain): node_id: str, *, leading_messages: list[dict[str, str]] | None = None, + triggering_user_text: str = "", ) -> NodeConfig: context_messages = list(leading_messages or []) for _ in range(MAX_AUTOMATIC_HOPS): node_type = self._engine.node_type(node_id) if node_type == "agent": await self._apply_agent_stage(node_id) - return self._agent_config(node_id, context_messages) + agent_messages = context_messages + if ( + triggering_user_text + and self._engine.data(node_id).get("contextPolicy") == "fresh" + ): + agent_messages = [ + {"role": "user", "content": triggering_user_text}, + *context_messages, + ] + return self._agent_config(node_id, agent_messages) if node_type == "end": await self._enter_end(node_id) return self._passive_node_config(node_id, context_messages) @@ -525,13 +616,9 @@ class WorkflowBrain(BaseBrain): raise RuntimeError(f"工作流指向未知节点:{node_id}") if not self._engine.has_outgoing(node_id): return self._passive_node_config(node_id, context_messages) - edge = self._engine.deterministic_edge( - node_id, - self._store, - include_default=True, - ) + edge = await self._select_edge(node_id) if not edge: - raise RuntimeError(f"自动节点 {node_id} 没有命中的表达式边或默认边") + return self._passive_node_config(node_id, context_messages) speech = self._engine.edge_transition_speech(edge) if speech: content = self._store.render(speech).strip() diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 0291afb..1e9ef24 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -116,7 +116,7 @@ def node_types_response() -> dict[str, Any]: return {"specVersion": SPEC_VERSION, "nodeTypes": NODE_SPECS} -def _edge_data_v3(edge: dict, source_type: str) -> dict: +def _edge_data_v3(edge: dict) -> dict: data = deepcopy(edge.get("data") or {}) if data.get("mode") in EDGE_MODES: data.setdefault("priority", 10) @@ -125,7 +125,7 @@ def _edge_data_v3(edge: dict, source_type: str) -> dict: transition = data.pop("transition_speech", data.get("transitionSpeech", "")) data.update( { - "mode": "llm" if condition and source_type == "agent" else "always", + "mode": "llm" if condition else "always", "priority": 10, "condition": condition, "transitionSpeech": transition, @@ -185,7 +185,6 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: edges = source.get("edges") or [] global_prompt = "" mapped_nodes: list[dict] = [] - type_by_id: dict[str, str] = {} start_prompt_nodes: dict[str, str] = {} type_map = { @@ -220,15 +219,11 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: data.pop(key, None) migrated["data"] = data mapped_nodes.append(migrated) - if migrated.get("id"): - type_by_id[str(migrated["id"])] = new_type mapped_edges: list[dict] = [] for edge in edges: migrated = deepcopy(edge) - migrated["data"] = _edge_data_v3( - migrated, type_by_id.get(str(migrated.get("source")), "") - ) + migrated["data"] = _edge_data_v3(migrated) mapped_edges.append(migrated) # A v2 Start was conversational. Insert a synthetic Agent so its prompt remains active. @@ -254,7 +249,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: for edge in mapped_edges: if edge.get("source") == start_id: edge["source"] = synthetic_id - edge["data"] = _edge_data_v3(edge, "agent") + edge["data"] = _edge_data_v3(edge) if str(edge["data"].get("condition") or "").strip(): edge["data"]["mode"] = "llm" mapped_edges.append( @@ -352,10 +347,8 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: mode = data.get("mode") if mode not in EDGE_MODES: errors.append(f"边 {edge_id} 的判断模式无效:{mode}") - if mode == "llm" and node_by_id[source_id].get("type") != "agent": - errors.append(f"LLM 判断边只能从 Agent 发出:{edge_id}") if mode == "llm" and not str(data.get("condition") or "").strip(): - errors.append(f"LLM 判断边缺少自然语言条件:{edge_id}") + errors.append(f"大模型判断边缺少自然语言条件:{edge_id}") if mode == "expression": expression_errors = _validate_expression(data.get("expression")) errors.extend(f"边 {edge_id}:{item}" for item in expression_errors) @@ -370,7 +363,7 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if mode == "always": always_counts[source_id] += 1 if always_counts[source_id] > 1: - errors.append(f"节点 {source_id} 最多只能有一条默认边") + errors.append(f"节点 {source_id} 最多只能有一条默认路径") incoming[target_id] += 1 outgoing[source_id] += 1 adj[source_id].append(target_id) @@ -394,14 +387,6 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: errors.append(f"节点 {node_id} 的{label}不能少于 {lo}") if hi is not None and actual > hi: errors.append(f"节点 {node_id} 的{label}不能多于 {hi}") - node_type = node.get("type") - if ( - node_type in AUTOMATIC_NODE_TYPES - and outgoing[node_id] > 0 - and always_counts[node_id] != 1 - ): - errors.append(f"自动节点 {node_id} 存在出边时必须有且仅有一条默认边") - start_id = next( (node_id for node_id, node in node_by_id.items() if node.get("type") == "start"), None, diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py index 074c170..ff8e979 100644 --- a/backend/services/pipecat/pipeline_events.py +++ b/backend/services/pipecat/pipeline_events.py @@ -3,6 +3,8 @@ from loguru import logger from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, EndFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, @@ -33,6 +35,28 @@ def bind_cascade_pipeline_events( pending_text_inputs: list[str] = [] greeting_transcript_sent = False greeting_timestamp = "" + greeting_playback_pending = False + greeting_playback_started = False + + # FlowManager already observes downstream frames for its own actions. Add + # to that filter instead of replacing it, then use the real transport + # playback boundary to release Workflow startup. + worker.add_reached_downstream_filter( + (BotStartedSpeakingFrame, BotStoppedSpeakingFrame) + ) + + @worker.event_handler("on_frame_reached_downstream") + async def on_frame_reached_downstream(_worker, frame): + nonlocal greeting_playback_pending, greeting_playback_started + if not greeting_playback_pending: + return + if isinstance(frame, BotStartedSpeakingFrame): + greeting_playback_started = True + return + if isinstance(frame, BotStoppedSpeakingFrame) and greeting_playback_started: + greeting_playback_pending = False + greeting_playback_started = False + await brain.on_greeting_finished() async def queue_transcript(role: str, content: str, timestamp: str) -> None: if not content: @@ -132,7 +156,7 @@ def bind_cascade_pipeline_events( @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): - nonlocal greeting_timestamp + nonlocal greeting_timestamp, greeting_playback_pending if vision_enabled: try: vision_state["client_id"] = get_transport_client_id( @@ -145,16 +169,24 @@ def bind_cascade_pipeline_events( ) except Exception as exc: # noqa: BLE001 - media availability is optional logger.warning(f"视觉理解摄像头捕获初始化失败: {exc}") - if greeting: + has_greeting = bool(greeting.strip()) + if has_greeting: # Preserve the actual playback order. The transcript is delivered # later on client-ready, but the preview sorts by this timestamp. greeting_timestamp = greeting_timestamp or time_now_iso8601() if brain.spec.owns_context: brain.prepare_greeting_context(greeting, context) + greeting_playback_pending = True + + # Initialize the Workflow before the greeting is queued so a very + # short TTS response cannot finish before the brain arms its startup + # gate. Other brain types simply ignore greeting_pending. + await brain.on_connected(greeting_pending=has_greeting) + + if has_greeting: await worker.queue_frame( TTSSpeakFrame(greeting, append_to_context=False) ) - await brain.on_connected() @transport.event_handler("on_client_disconnected") async def on_client_disconnected(_transport, _client): diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 1114231..d2b7c85 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -152,6 +152,20 @@ class WorkflowEngine: def greeting(self, store: DynamicVariableStore) -> str: return store.render(str(self.data(self.start_id).get("greeting") or "")) + def routing_prompt(self, node_id: str, store: DynamicVariableStore) -> str: + """Describe the current node to the small LLM edge router.""" + if self.node_type(node_id) == "agent": + return self.prompt_for(node_id, store) + data = self.data(node_id) + details = ( + data.get("greeting") + or data.get("message") + or data.get("target") + or "" + ) + rendered = store.render(str(details)).strip() + return f"{self.node_type(node_id) or 'workflow'} 节点：{rendered or self.name(node_id)}" + def expression_matches(self, expression: dict, values: dict[str, Any]) -> bool: results = [] for rule in expression.get("rules") or []: diff --git a/backend/services/workflow_router.py b/backend/services/workflow_router.py index 8170a3f..c76c9f5 100644 --- a/backend/services/workflow_router.py +++ b/backend/services/workflow_router.py @@ -1,4 +1,4 @@ -"""Pre-response LLM routing for Workflow Agent edges. +"""Small LLM router for Workflow conditional edges. The router deliberately uses a separate, short completion. Its only output is a required function choice, so the current Agent cannot speak before the graph @@ -16,12 +16,14 @@ from models import AssistantConfig from openai import AsyncOpenAI -STAY_ON_CURRENT_AGENT = "workflow_stay_on_current_agent" +STAY_ON_CURRENT_NODE = "workflow_stay_on_current_node" +# Compatibility alias for callers saved before all source nodes supported LLM edges. +STAY_ON_CURRENT_AGENT = STAY_ON_CURRENT_NODE MAX_ROUTING_HISTORY_ENTRIES = 20 class WorkflowLLMRouter: - """Select one LLM edge before the conversational LLM is allowed to reply.""" + """Select one LLM edge without allowing the router to speak.""" def __init__(self, cfg: AssistantConfig): self._cfg = cfg @@ -39,10 +41,10 @@ class WorkflowLLMRouter: ) -> str | None: """Return an edge function name, STAY, or None when routing failed.""" if not edges: - return STAY_ON_CURRENT_AGENT + return STAY_ON_CURRENT_NODE names = {edge_name(edge) for edge in edges} - stay_name = STAY_ON_CURRENT_AGENT + stay_name = STAY_ON_CURRENT_NODE while stay_name in names: stay_name = f"_{stay_name}" @@ -62,7 +64,7 @@ class WorkflowLLMRouter: "type": "function", "function": { "name": stay_name, - "description": "所有转移条件都不满足，继续由当前 Agent 处理用户消息。", + "description": "所有转移条件都不满足，留在当前节点。", "parameters": {"type": "object", "properties": {}}, }, } @@ -76,7 +78,7 @@ class WorkflowLLMRouter: "你是工作流路由器，不是对话助手。收到一轮完整用户输入后，" "必须且只能调用一个提供的函数，禁止输出任何口头回复。\n" "按给出的顺序判断转移条件；选择第一个明确满足的转移函数。" - "如果没有条件满足，调用留在当前 Agent 的函数。\n\n" + "如果没有条件满足，调用留在当前节点的函数。\n\n" f"当前节点：{node_name}\n" f"当前节点任务：{node_prompt or '未配置'}\n" f"转移条件：\n{ordered_conditions}" @@ -113,17 +115,17 @@ class WorkflowLLMRouter: ) tool_calls = response.choices[0].message.tool_calls or [] if not tool_calls: - logger.warning("Workflow 路由 LLM 未返回函数调用，留在当前 Agent") - return STAY_ON_CURRENT_AGENT + logger.warning("Workflow 路由 LLM 未返回函数调用，留在当前节点") + return STAY_ON_CURRENT_NODE selected = str(tool_calls[0].function.name or "") if selected == stay_name: - return STAY_ON_CURRENT_AGENT + return STAY_ON_CURRENT_NODE if selected not in names: logger.warning(f"Workflow 路由 LLM 返回未知函数:{selected}") - return STAY_ON_CURRENT_AGENT + return STAY_ON_CURRENT_NODE return selected except Exception as exc: # noqa: BLE001 - routing failure must not end the call - logger.warning(f"Workflow LLM 边判断失败，留在当前 Agent:{exc}") + logger.warning(f"Workflow LLM 边判断失败，留在当前节点:{exc}") return None finally: await client.close() diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index d4af2f6..4a1c5ac 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -28,6 +28,7 @@ from services.brains.dify_llm import ( ) from services.brains.workflow_brain import WorkflowBrain from services.runtime_variables import prepare_dynamic_config +from services.workflow_router import STAY_ON_CURRENT_NODE class FakeLLM: @@ -460,6 +461,86 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_initial_fixed_speech_waits_for_start_greeting_to_finish(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, + "nodes": [ + { + "id": "start", + "type": "start", + "data": {"greeting": "欢迎使用"}, + }, + { + "id": "agent", + "type": "agent", + "data": { + "prompt": "收集用户信息", + "entryMode": "fixed_speech", + "entrySpeech": "请问您怎么称呼？", + }, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + } + ], + } + ) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + class FakeManager: + def __init__(self): + self.current_node = None + + async def initialize(self, config): + self.current_node = config["name"] + + async def set_node_from_config(self, config): + self.current_node = config["name"] + for action in config.get("pre_actions", []): + await action["handler"](action, self) + + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + brain._manager = FakeManager() + + await brain.on_connected(greeting_pending=True) + + self.assertEqual(brain._manager.current_node, "start") + self.assertFalse(any(isinstance(frame, TTSSpeakFrame) for frame in queued)) + + await brain.on_greeting_finished() + + self.assertEqual(brain._manager.current_node, "agent") + fixed_speech_frames = [ + frame for frame in queued if isinstance(frame, TTSSpeakFrame) + ] + self.assertEqual(len(fixed_speech_frames), 1) + self.assertEqual(fixed_speech_frames[0].text, "请问您怎么称呼？") + + # Playback notifications may be duplicated by a transport reconnect; + # the initial entry behavior must still run only once. + await brain.on_greeting_finished() + self.assertEqual( + len([frame for frame in queued if isinstance(frame, TTSSpeakFrame)]), + 1, + ) + async def test_action_publishes_updated_session_variables(self): tool = RuntimeTool( id="lookup", @@ -652,6 +733,261 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) ) + async def test_start_llm_conditions_wait_for_and_route_first_user_turn(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, + "nodes": [ + { + "id": "start", + "type": "start", + "data": {"name": "Start", "greeting": "你想做什么？"}, + }, + { + "id": "eat", + "type": "agent", + "data": { + "name": "点饭", + "prompt": "帮助用户点饭", + "contextPolicy": "fresh", + "entryMode": "wait_user", + }, + }, + {"id": "drink", "type": "agent", "data": {}}, + {"id": "run", "type": "agent", "data": {}}, + ], + "edges": [ + { + "id": "eat", + "source": "start", + "target": "eat", + "data": { + "mode": "llm", + "priority": 10, + "condition": "用户想吃饭", + }, + }, + { + "id": "drink", + "source": "start", + "target": "drink", + "data": { + "mode": "llm", + "priority": 20, + "condition": "用户想喝水", + }, + }, + { + "id": "run", + "source": "start", + "target": "run", + "data": { + "mode": "llm", + "priority": 30, + "condition": "用户想跑步", + }, + }, + ], + } + ) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + + class FakeManager: + def __init__(self): + self.current_node = None + self.config = None + + async def initialize(self, config): + self.current_node = config["name"] + self.config = config + + async def set_node_from_config(self, config): + self.current_node = config["name"] + self.config = config + + class FakeRouter: + def __init__(self): + self.calls = 0 + + async def select_edge(self, **_kwargs): + self.calls += 1 + return "goto_eat" + + manager = FakeManager() + router = FakeRouter() + brain._manager = manager + brain._router = router + + await brain.on_connected() + self.assertEqual(manager.current_node, "start") + self.assertEqual(router.calls, 0) + + handled = await brain.on_user_turn_end("我想吃饭") + + self.assertTrue(handled) + self.assertEqual(router.calls, 1) + self.assertEqual(manager.current_node, "eat") + self.assertIn( + {"role": "user", "content": "我想吃饭"}, + manager.config["task_messages"], + ) + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) + self.assertIn("我想吃饭", brain._store.values["system__conversation_history"]) + + async def test_automatic_node_can_follow_llm_condition(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "handoff", + "type": "handoff", + "data": {"name": "人工转接", "targetType": "human"}, + }, + { + "id": "agent", + "type": "agent", + "data": {"name": "继续服务", "prompt": "继续处理"}, + }, + ], + "edges": [ + { + "id": "to-agent", + "source": "handoff", + "target": "agent", + "data": { + "mode": "llm", + "priority": 10, + "condition": "转接后仍需 AI 继续服务", + }, + } + ], + } + ) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + + class FakeRouter: + async def select_edge(self, **kwargs): + self.node_name = kwargs["node_name"] + return "goto_to_agent" + + router = FakeRouter() + brain._router = router + + config = await brain._resolve_path("handoff") + + self.assertEqual(config["name"], "agent") + self.assertEqual(router.node_name, "人工转接") + + async def test_mixed_edge_conditions_follow_priority(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "llm-target", "type": "end", "data": {}}, + {"id": "expression-target", "type": "end", "data": {}}, + {"id": "default-target", "type": "end", "data": {}}, + ], + "edges": [ + { + "id": "llm", + "source": "agent", + "target": "llm-target", + "data": { + "mode": "llm", + "priority": 10, + "condition": "大模型条件成立", + }, + }, + { + "id": "expression", + "source": "agent", + "target": "expression-target", + "data": { + "mode": "expression", + "priority": 20, + "expression": { + "combinator": "and", + "rules": [ + { + "variable": "route", + "operator": "eq", + "value": "expression", + } + ], + }, + }, + }, + { + "id": "default", + "source": "agent", + "target": "default-target", + "data": {"mode": "always", "priority": 30}, + }, + ], + } + ) + brain._store.values["route"] = "expression" + + class FakeRouter: + def __init__(self): + self.result = STAY_ON_CURRENT_NODE + self.edge_ids = [] + + async def select_edge(self, **kwargs): + self.edge_ids = [edge["id"] for edge in kwargs["edges"]] + return self.result + + router = FakeRouter() + brain._router = router + + selected = await brain._select_edge("agent") + self.assertEqual(router.edge_ids, ["llm"]) + self.assertEqual(selected["id"], "expression") + + router.result = "goto_llm" + selected = await brain._select_edge("agent") + self.assertEqual(selected["id"], "llm") + + expression_edge = next( + edge for edge in brain._engine.edges if edge["id"] == "expression" + ) + expression_edge["data"]["priority"] = 5 + router.edge_ids = [] + selected = await brain._select_edge("agent") + self.assertEqual(selected["id"], "expression") + self.assertEqual(router.edge_ids, []) + async def test_transition_and_end_are_owned_by_workflow_brain(self): graph = { "specVersion": 3, diff --git a/backend/tests/test_pipeline_events.py b/backend/tests/test_pipeline_events.py index 2a82d08..8af5eb9 100644 --- a/backend/tests/test_pipeline_events.py +++ b/backend/tests/test_pipeline_events.py @@ -4,7 +4,11 @@ import unittest from types import SimpleNamespace from unittest.mock import patch -from pipecat.frames.frames import OutputTransportMessageUrgentFrame +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + OutputTransportMessageUrgentFrame, +) from services.pipecat.pipeline_events import bind_cascade_pipeline_events @@ -23,6 +27,18 @@ class _EventSource: class _Worker: def __init__(self): self.frames = [] + self.handlers = {} + self.downstream_types = set() + + def add_reached_downstream_filter(self, types): + self.downstream_types.update(types) + + def event_handler(self, name): + def decorator(handler): + self.handlers[name] = handler + return handler + + return decorator async def queue_frame(self, frame): self.frames.append(frame) @@ -34,12 +50,17 @@ class _Brain: def __init__(self, worker): self.worker = worker self.prepared_greeting = "" + self.greeting_pending = False + self.greeting_finished = 0 def prepare_greeting_context(self, greeting, _context): self.prepared_greeting = greeting - async def on_connected(self): - pass + async def on_connected(self, *, greeting_pending=False): + self.greeting_pending = greeting_pending + + async def on_greeting_finished(self): + self.greeting_finished += 1 async def on_client_ready(self): for content, timestamp in ( @@ -101,8 +122,46 @@ class PipelineEventTest(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(transcripts[0]["timestamp"], greeting_time) self.assertEqual(brain.prepared_greeting, "助手开场白") + self.assertTrue(brain.greeting_pending) clock.assert_called_once_with() + async def test_greeting_releases_workflow_only_after_real_playback_stop(self): + transport = _EventSource() + text_input = _EventSource() + user_aggregator = _EventSource() + assistant_aggregator = _EventSource() + worker = _Worker() + brain = _Brain(worker) + + bind_cascade_pipeline_events( + transport=transport, + worker=worker, + brain=brain, + context=SimpleNamespace(), + text_input=text_input, + user_aggregator=user_aggregator, + assistant_aggregator=assistant_aggregator, + greeting="助手开场白", + vision_enabled=False, + vision_state={"client_id": None}, + ) + + await transport.handlers["on_client_connected"](transport, object()) + playback_handler = worker.handlers["on_frame_reached_downstream"] + + # An unrelated stop cannot release startup until this greeting has + # actually produced audio. + await playback_handler(worker, BotStoppedSpeakingFrame()) + self.assertEqual(brain.greeting_finished, 0) + + await playback_handler(worker, BotStartedSpeakingFrame()) + await playback_handler(worker, BotStoppedSpeakingFrame()) + self.assertEqual(brain.greeting_finished, 1) + + # Duplicate transport notifications are harmless. + await playback_handler(worker, BotStoppedSpeakingFrame()) + self.assertEqual(brain.greeting_finished, 1) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 9fde4bc..69f68dd 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -284,6 +284,111 @@ class WorkflowGraphTests(unittest.TestCase): ) ) + def test_all_source_nodes_allow_multiple_conditional_paths(self): + expression = { + "combinator": "and", + "rules": [{"variable": "route", "operator": "eq", "value": "yes"}], + } + cases = { + "start": { + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + {"id": "default", "source": "start", "target": "agent", "data": {"mode": "always", "priority": 10}}, + {"id": "llm", "source": "start", "target": "end", "data": {"mode": "llm", "priority": 20, "condition": "应当结束"}}, + {"id": "expr", "source": "start", "target": "action", "data": {"mode": "expression", "priority": 30, "expression": expression}}, + {"id": "action-end", "source": "action", "target": "end", "data": {"mode": "always", "priority": 10}}, + ], + }, + "agent": { + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + {"id": "begin", "source": "start", "target": "agent", "data": {"mode": "always", "priority": 10}}, + {"id": "default", "source": "agent", "target": "end", "data": {"mode": "always", "priority": 10}}, + {"id": "llm", "source": "agent", "target": "action", "data": {"mode": "llm", "priority": 20, "condition": "执行操作"}}, + {"id": "expr", "source": "agent", "target": "handoff", "data": {"mode": "expression", "priority": 30, "expression": expression}}, + {"id": "action-end", "source": "action", "target": "end", "data": {"mode": "always", "priority": 10}}, + ], + }, + "action": { + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + {"id": "begin", "source": "start", "target": "action", "data": {"mode": "always", "priority": 10}}, + {"id": "default", "source": "action", "target": "agent", "data": {"mode": "always", "priority": 10}}, + {"id": "llm", "source": "action", "target": "end", "data": {"mode": "llm", "priority": 20, "condition": "执行失败"}}, + {"id": "expr", "source": "action", "target": "handoff", "data": {"mode": "expression", "priority": 30, "expression": expression}}, + ], + }, + "handoff": { + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + {"id": "begin", "source": "start", "target": "handoff", "data": {"mode": "always", "priority": 10}}, + {"id": "default", "source": "handoff", "target": "agent", "data": {"mode": "always", "priority": 10}}, + {"id": "llm", "source": "handoff", "target": "end", "data": {"mode": "llm", "priority": 20, "condition": "转接完成"}}, + {"id": "expr", "source": "handoff", "target": "action", "data": {"mode": "expression", "priority": 30, "expression": expression}}, + {"id": "action-end", "source": "action", "target": "end", "data": {"mode": "always", "priority": 10}}, + ], + }, + } + + for source_type, parts in cases.items(): + graph = {"specVersion": 3, "settings": {}, **parts} + with self.subTest(source_type=source_type): + self.assertEqual(validate_graph(graph), []) + + graph["edges"].append( + { + "id": "duplicate-default", + "source": source_type, + "target": "end", + "data": {"mode": "always", "priority": 40}, + } + ) + self.assertTrue( + any("最多只能有一条默认路径" in error for error in validate_graph(graph)) + ) + + def test_v2_condition_on_non_agent_becomes_llm_path(self): + graph = normalize_graph( + { + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + ], + "edges": [ + { + "id": "route", + "source": "start", + "target": "handoff", + "data": {"condition": "用户需要人工服务"}, + } + ], + } + ) + self.assertEqual(graph["edges"][0]["data"]["mode"], "llm") + def test_v2_start_prompt_is_preserved_in_synthetic_agent(self): graph = normalize_graph( { diff --git a/frontend/src/components/workflow/ConditionEdge.tsx b/frontend/src/components/workflow/ConditionEdge.tsx index b508f83..fb7e4d0 100644 --- a/frontend/src/components/workflow/ConditionEdge.tsx +++ b/frontend/src/components/workflow/ConditionEdge.tsx @@ -49,7 +49,11 @@ export function ConditionEdge({ const label = ( (data?.label as string) || (mode === "llm" ? (data?.condition as string) : "") || - (mode === "expression" ? "变量表达式" : "默认路径") + (mode === "llm" + ? "大模型判断" + : mode === "expression" + ? "表达式" + : "默认路径") ).toString().trim(); const expanded = hovered || selected; diff --git a/frontend/src/components/workflow/WorkflowCanvas.tsx b/frontend/src/components/workflow/WorkflowCanvas.tsx index a8ee414..18555c2 100644 --- a/frontend/src/components/workflow/WorkflowCanvas.tsx +++ b/frontend/src/components/workflow/WorkflowCanvas.tsx @@ -41,6 +41,7 @@ import { } from "@/components/ui/tooltip"; import { edgeTypes } from "./ConditionEdge"; +import { hasDefaultPath, newEdgeData } from "./edge-rules"; import { ActiveNodeContext, EdgeActionContext, @@ -201,9 +202,7 @@ export function WorkflowCanvas({ const onConnect = useCallback( (connection: Connection) => { - const sourceType = nodes.find((node) => node.id === connection.source)?.type; - const priority = - edges.filter((edge) => edge.source === connection.source).length * 10 + 10; + if (!connection.source || !connection.target) return; setEdges((eds) => addEdge( { @@ -211,20 +210,13 @@ export function WorkflowCanvas({ id: `e-${connection.source}-${connection.target}-${Date.now()}`, type: "condition", animated: true, - data: - sourceType === "agent" - ? { - mode: "llm", - priority, - condition: "当前阶段任务已经完成", - } - : { mode: "always", priority }, + data: newEdgeData(eds, connection.source), }, eds, ), ); }, - [nodes, edges, setEdges], + [setEdges], ); // 连线约束:不能连入开始节点(无入边句柄),不能自连。 @@ -278,8 +270,6 @@ export function WorkflowCanvas({ setNodes((ns) => [...ns, { id, type: spec.type, position, data }]); if (source) { setEdges((currentEdges) => { - const priority = - currentEdges.filter((edge) => edge.source === source.id).length * 10 + 10; return addEdge( { id: `e-${source.id}-${id}-${Date.now()}`, @@ -287,14 +277,7 @@ export function WorkflowCanvas({ target: id, type: "condition", animated: true, - data: - source.type === "agent" - ? { - mode: "llm", - priority, - condition: "当前阶段任务已经完成", - } - : { mode: "always", priority }, + data: newEdgeData(currentEdges, source.id), }, currentEdges, ); @@ -363,9 +346,16 @@ export function WorkflowCanvas({ patch: WorkflowGraph["edges"][number]["data"], ) => { setEdges((es) => - es.map((e) => - e.id === id ? { ...e, data: { ...(e.data ?? {}), ...patch } } : e, - ), + es.map((e) => { + if (e.id !== id) return e; + if ( + patch.mode === "always" && + hasDefaultPath(es, e.source, e.id) + ) { + return e; + } + return { ...e, data: { ...(e.data ?? {}), ...patch } }; + }), ); }, [setEdges], @@ -392,7 +382,7 @@ export function WorkflowCanvas({ ) { return false; } - return source.type === "agent" || outgoingCount === 0; + return true; }, [edges, nodes, specsByType], ); @@ -751,7 +741,11 @@ export function WorkflowCanvas({ node.id === editingEdge.source)?.type} + hasOtherDefaultPath={hasDefaultPath( + edges, + editingEdge.source, + editingEdge.id, + )} onChange={(patch) => updateEdgeData(editingEdge.id, patch) } @@ -769,4 +763,3 @@ export function WorkflowCanvas({ ); } - diff --git a/frontend/src/components/workflow/edge-rules.ts b/frontend/src/components/workflow/edge-rules.ts new file mode 100644 index 0000000..a74d203 --- /dev/null +++ b/frontend/src/components/workflow/edge-rules.ts @@ -0,0 +1,42 @@ +/** Shared rules for creating and editing Workflow outgoing paths. */ + +import type { Edge } from "@xyflow/react"; + +import type { WorkflowEdgeData } from "./specs"; + +export function isDefaultPath(edge: Pick): boolean { + const mode = (edge.data as Partial | undefined)?.mode; + return !mode || mode === "always"; +} + +export function hasDefaultPath( + edges: Edge[], + sourceId: string, + excludingEdgeId?: string, +): boolean { + return edges.some( + (edge) => + edge.source === sourceId && + edge.id !== excludingEdgeId && + isDefaultPath(edge), + ); +} + +export function nextEdgePriority(edges: Edge[], sourceId: string): number { + const priorities = edges + .filter((edge) => edge.source === sourceId) + .map((edge) => Number((edge.data as Partial)?.priority)) + .filter(Number.isFinite); + return priorities.length === 0 ? 10 : Math.max(...priorities) + 10; +} + +/** + * The first path from a node is deterministic. Further paths start as an + * incomplete LLM condition so the graph never silently gains two defaults. + */ +export function newEdgeData(edges: Edge[], sourceId: string): WorkflowEdgeData { + const priority = nextEdgePriority(edges, sourceId); + return hasDefaultPath(edges, sourceId) + ? { mode: "llm", priority, condition: "" } + : { mode: "always", priority }; +} diff --git a/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx b/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx index 53c23f0..af089f0 100644 --- a/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx +++ b/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx @@ -27,11 +27,11 @@ import type { ExpressionRule, WorkflowEdgeData } from "../specs"; export function EdgeSettingsPanel({ edge, - sourceType, + hasOtherDefaultPath, onChange, }: { edge: Edge; - sourceType?: string; + hasOtherDefaultPath: boolean; onChange: (patch: WorkflowEdgeData) => void; }) { const data = (edge.data ?? { mode: "always", priority: 10 }) as WorkflowEdgeData; @@ -100,15 +100,19 @@ export function EdgeSettingsPanel({ } title="路由方式" - description="选择由 Agent 判断、动态变量表达式判断，或作为确定性默认路径" + description="每个节点最多一条默认路径，也可以配置多条条件路径" > { const nextMode = @@ -118,6 +122,11 @@ export function EdgeSettingsPanel({ }} allowNone={false} /> + {mode !== "always" && hasOtherDefaultPath && ( + + 当前节点已经有一条默认路径；其它出边需要使用大模型判断或表达式。 + + )}