From bdf3d3dd9cf239d850ad795a423fb0209431737c Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 17 Jul 2026 22:37:15 +0800 Subject: [PATCH] Refactor workflow agent and routing components for improved functionality - Introduce WorkflowAgentStage to manage agent stage configurations and enhance interaction with the workflow engine. - Implement WorkflowEdgeEvaluator for priority-aware edge evaluation, improving routing decisions based on conditions and user turns. - Update WorkflowBrain to handle user turns and routing more effectively, ensuring agents cannot have only one default path. - Enhance CallEndCoordinator to track speech events and manage call termination based on queued speech. - Add new models and output handling for workflow interactions, improving clarity and maintainability. - Update tests to validate the new routing logic and agent behavior under various scenarios. --- AGENTS.md | 6 +- backend/services/brains/base.py | 4 + backend/services/brains/workflow_brain.py | 536 +++++++++--------- backend/services/node_specs.py | 28 +- backend/services/pipecat/call_lifecycle.py | 22 +- backend/services/workflow/__init__.py | 19 + backend/services/workflow/agent.py | 134 +++++ backend/services/workflow/models.py | 92 +++ backend/services/workflow/output.py | 119 ++++ backend/services/workflow/routing.py | 97 ++++ backend/services/workflow_engine.py | 5 + backend/services/workflow_router.py | 30 +- backend/tests/test_brains.py | 86 ++- backend/tests/test_call_lifecycle.py | 14 + backend/tests/test_workflow_router.py | 4 +- backend/tests/test_workflow_routing.py | 80 +++ backend/tests/test_workflow_v3.py | 75 ++- .../use-workflow-editor-state.ts | 86 ++- .../src/components/workflow/AddNodeDialog.tsx | 116 ++++ .../components/workflow/WorkflowCanvas.tsx | 194 ++----- .../components/workflow/WorkflowSidePanel.tsx | 38 ++ .../src/components/workflow/edge-rules.ts | 14 +- .../workflow/panels/EdgeSettingsPanel.tsx | 50 +- 23 files changed, 1374 insertions(+), 475 deletions(-) create mode 100644 backend/services/workflow/__init__.py create mode 100644 backend/services/workflow/agent.py create mode 100644 backend/services/workflow/models.py create mode 100644 backend/services/workflow/output.py create mode 100644 backend/services/workflow/routing.py create mode 100644 backend/tests/test_workflow_routing.py create mode 100644 frontend/src/components/workflow/AddNodeDialog.tsx create mode 100644 frontend/src/components/workflow/WorkflowSidePanel.tsx diff --git a/AGENTS.md b/AGENTS.md index a072005..124428a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,6 +1,6 @@ -你编写具有可维护性、高可读性、可扩展性、模块化的的代码，尽量不去对pipecat框架本身修改 -以MVP构建为目标 +你编写具有可维护性、高可读性、可扩展性、模块化的的代码 适合CS的本科学生阅读修改 +尽量不去对pipecat框架本身修改 +以MVP构建为目标 编写代码之前先用易于理解的语言说清楚思路 - 界面设计要参考 frontend/DESIGN.md \ No newline at end of file diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index e0b1397..b0d5681 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -55,6 +55,10 @@ class CallEndPort(Protocol): def arm_after_speech(self) -> None: ... + def track_speech(self) -> None: ... + + async def arm_after_tracked_speech(self) -> None: ... + async def finish_after_current_speech(self, *, has_text: bool) -> None: ... async def finish(self) -> None: ... diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index fa81b84..0b9b880 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio from copy import deepcopy from typing import Any @@ -19,26 +20,26 @@ from pipecat.frames.frames import ( LLMRunFrame, LLMUpdateSettingsFrame, OutputTransportMessageUrgentFrame, - TTSSpeakFrame, ) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor -from pipecat.services.settings import LLMSettings -from pipecat.utils.time import time_now_iso8601 - from services.brains.base import BaseBrain, BrainRuntime, BrainSpec from services.knowledge import search as search_knowledge from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor +from services.workflow.agent import WorkflowAgentStage +from services.workflow.models import ( + RouteStatus, + WorkflowRuntimeState, + WorkflowStatus, +) +from services.workflow.output import WorkflowOutput +from services.workflow.routing import WorkflowEdgeEvaluator from services.workflow_engine import WorkflowEngine -from services.workflow_router import STAY_ON_CURRENT_NODE, WorkflowLLMRouter +from services.workflow_router import WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 -AGENT_STAGE_INSTRUCTION = ( - "工作流路由已在用户一轮输入结束时完成。只执行当前阶段任务，" - "不要自行解释、模拟或宣布节点切换。" -) class WorkflowBrain(BaseBrain): @@ -69,11 +70,18 @@ class WorkflowBrain(BaseBrain): self._runtime: BrainRuntime | None = None self._manager: FlowManager | None = None self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) + self._edge_evaluator = WorkflowEdgeEvaluator( + self._engine, + self._store, + self._router_for_node, + ) + self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id) + self._turn_lock = asyncio.Lock() + self._output: WorkflowOutput | None = None + self._agent_stage: WorkflowAgentStage | None = None self._ended = False self._greeting_context_message: dict[str, str] | None = None self._startup_waiting_for_greeting = False - self._client_ready = False - self._pending_visible_speech_events: list[dict[str, Any]] = [] async def greeting(self, cfg: AssistantConfig) -> str: return self._engine.greeting(self._store) or cfg.greeting @@ -95,10 +103,23 @@ class WorkflowBrain(BaseBrain): self._tools = ToolExecutor(self._store) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) + self._edge_evaluator = WorkflowEdgeEvaluator( + self._engine, + self._store, + self._router_for_node, + ) + self._state = WorkflowRuntimeState(current_node_id=self._engine.start_id) + self._turn_lock = asyncio.Lock() + self._output = WorkflowOutput(self._store, runtime) + self._agent_stage = WorkflowAgentStage( + cfg=cfg, + engine=self._engine, + store=self._store, + runtime=runtime, + ) + self._ended = False self._greeting_context_message = None self._startup_waiting_for_greeting = False - self._client_ready = False - self._pending_visible_speech_events = [] self._manager = FlowManager( worker=runtime.worker, llm=runtime.llm, @@ -118,6 +139,7 @@ class WorkflowBrain(BaseBrain): return message async def on_connected(self, *, greeting_pending: bool = False) -> None: + self._state.enter(self._engine.start_id, WorkflowStatus.STARTING) await self._emit_node_active(self._engine.start_id) await self._emit_variables( reason="initialized", @@ -139,6 +161,7 @@ class WorkflowBrain(BaseBrain): node_config = await self._initial_node_config() await self._manager.initialize(node_config) + await self._after_node_activated(node_config) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") async def on_greeting_finished(self) -> None: @@ -152,17 +175,28 @@ class WorkflowBrain(BaseBrain): node_config = await self._initial_node_config() if node_config.get("name") == self._engine.start_id: + self._state.enter(self._engine.start_id, WorkflowStatus.WAITING_USER) return await manager.set_node_from_config(node_config) + await self._after_node_activated(node_config) logger.info(f"Start 开场白结束,进入节点: {manager.current_node}") async def _initial_node_config(self) -> NodeConfig: - """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, - evaluate_llm=False, + """Only a default-only Start advances before the first user turn.""" + outgoing = self._engine.outgoing(self._engine.start_id) + has_condition = any( + self._engine.edge_mode(edge) != "always" for edge in outgoing + ) + if has_condition: + self._state.enter(self._engine.start_id, WorkflowStatus.WAITING_USER) + return self._passive_node_config(self._engine.start_id) + edge = next( + ( + candidate + for candidate in outgoing + if self._engine.edge_mode(candidate) == "always" + ), + None, ) return ( await self._follow_edge(edge) @@ -172,18 +206,14 @@ class WorkflowBrain(BaseBrain): async def on_client_ready(self) -> None: """Replay state that may have been emitted before WebRTC data was ready.""" - self._client_ready = True - pending_speech_events = self._pending_visible_speech_events - self._pending_visible_speech_events = [] - for message in pending_speech_events: - await self._require_runtime().queue_frame( - OutputTransportMessageUrgentFrame(message=message) - ) + await self._require_output().mark_client_ready() current_node = ( str(self._manager.current_node) if self._manager and self._manager.current_node - else self._engine.start_id + else self._state.current_node_id ) + if current_node != self._state.current_node_id: + self._state.current_node_id = current_node await self._emit_node_active(current_node) await self._emit_variables( reason="client_ready", @@ -198,114 +228,69 @@ class WorkflowBrain(BaseBrain): """Route a complete user turn before the active stage may reply.""" if not content or self._ended: return True + async with self._turn_lock: + return await self._handle_user_turn_end(content) + + async def _handle_user_turn_end(self, content: str) -> bool: + """Serialized implementation so one user turn cannot transition twice.""" self.record_user_message(content) + self._state.begin_user_turn(content) manager = self._require_manager() - current = manager.current_node + current = self._state.current_node_id if not current: return True - edge = await self._select_edge(current) + self._state.status = WorkflowStatus.ROUTING + decision = await self._edge_evaluator.evaluate(current) + if decision.status == RouteStatus.ERROR: + await self._require_output().emit_error( + decision.error or "工作流路由失败", + node_id=current, + code="workflow_routing_error", + ) + return await self._continue_current_node_after_no_transition(current) - if edge and manager.current_node == current: + if decision.edge and manager.current_node == current: next_config = await self._follow_edge( - edge, + decision.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()) + await self._after_node_activated( + next_config, + triggering_user_text=content, + ) 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 await self._continue_current_node_after_no_transition(current) + + async def _continue_current_node_after_no_transition( + self, + node_id: str, + ) -> bool: + if self._engine.node_type(node_id) != "agent": + self._state.enter(node_id, WorkflowStatus.WAITING_USER) return True - # 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._refresh_agent_prompt(node_id) + self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) await self._require_runtime().queue_frame(LLMRunFrame()) return True async def _select_edge( self, node_id: str, - *, - 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(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 - for candidate in outgoing - if self._engine.edge_mode(candidate) == "always" - ), - None, - ) - - # 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 not evaluate_llm: - if expression_edge and not llm_edges: - return expression_edge - if all_llm_edges: - return None - return default_edge - - 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, - ) + """Compatibility helper used by automatic-node traversal and tests.""" + decision = await self._edge_evaluator.evaluate(node_id) + if decision.status == RouteStatus.ERROR: + await self._require_output().emit_error( + decision.error or "工作流路由失败", + node_id=node_id, + code="workflow_routing_error", + ) + return None + return decision.edge async def on_assistant_text_end( self, @@ -316,88 +301,29 @@ class WorkflowBrain(BaseBrain): if not content or interrupted or self._ended: return self._store.record("agent", content, completed_agent_turn=True) + self._state.consume_user_turn() + if self._engine.node_type(self._state.current_node_id) == "agent": + self._state.status = WorkflowStatus.WAITING_USER async def _refresh_agent_prompt(self, node_id: str) -> None: - runtime = self._require_runtime() - await runtime.queue_frame( - LLMUpdateSettingsFrame( - delta=LLMSettings( - system_instruction=self._agent_role_message(node_id) - ) - ) - ) + await self._require_agent_stage().refresh_prompt(node_id) def _agent_role_message(self, node_id: str) -> str: - """Build one provider-compatible system instruction for an Agent stage.""" - stage_prompt = self._engine.prompt_for(node_id, self._store) - return f"{stage_prompt}\n\n[工作流执行规则]\n{AGENT_STAGE_INSTRUCTION}" + return self._require_agent_stage().role_message(node_id) def _router_for_node(self, node_id: str) -> WorkflowLLMRouter: - stage = self._engine.agent_stage_config(node_id) - resource_id = stage.llm_resource_id - cfg = self._cfg - resource = cfg.workflow_model_resources.get(resource_id) if cfg else None - if not cfg or not resource: + if self._agent_stage is None: return self._router - from services.pipecat.service_factory import config_with_resource - - return WorkflowLLMRouter(config_with_resource(cfg, resource)) + return self._agent_stage.router_for_node(node_id, self._router) async def _apply_agent_stage(self, node_id: str) -> None: - stage = self._engine.agent_stage_config(node_id) - await self._emit_node_active(node_id) - if self._runtime and self._runtime.set_input_enabled: - self._runtime.set_input_enabled(True) - runtime = self._require_runtime() - if runtime.apply_turn_config: - await runtime.apply_turn_config( - stage.enable_interrupt, - stage.turn_config, - ) - if runtime.switch_services: - await runtime.switch_services( - stage.llm_resource_id or None, - stage.asr_resource_id or None, - stage.tts_resource_id or None, - ) - if runtime.set_knowledge_scope: - runtime.set_knowledge_scope( - { - "knowledge_base_id": stage.knowledge_base_id, - "mode": stage.knowledge_mode, - "top_n": stage.knowledge_top_n, - "score_threshold": stage.knowledge_score_threshold, - } - ) + await self._require_agent_stage().apply(node_id) def _agent_config( self, node_id: str, leading_messages: list[dict[str, str]] | None = None, ) -> NodeConfig: - data = self._engine.data(node_id) - entry_mode = str(data.get("entryMode") or "wait_user") - entry_speech = self._store.render(str(data.get("entrySpeech") or "")) - fixed_reply_messages: list[dict[str, str]] = ( - [{"role": "assistant", "content": entry_speech}] - if entry_mode == "fixed_speech" and entry_speech - else [] - ) - strategy = ( - ContextStrategy.RESET - if data.get("contextPolicy") == "fresh" - else ContextStrategy.APPEND - ) - greeting_messages = ( - [deepcopy(self._greeting_context_message)] - if strategy == ContextStrategy.RESET and self._greeting_context_message - else [] - ) - task_messages = [ - *greeting_messages, - *(leading_messages or []), - *fixed_reply_messages, - ] stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] for tool_id in stage.tool_ids: @@ -407,36 +333,49 @@ class WorkflowBrain(BaseBrain): knowledge_function = self._knowledge_function(node_id) if knowledge_function: functions.append(knowledge_function) - config: NodeConfig = { - "name": node_id, - "role_message": self._agent_role_message(node_id), - # Flows writes task_messages into the Pipecat LLM context. The - # pre-action below is responsible only for display, persistence, - # dynamic conversation history, and TTS playback. - "task_messages": task_messages, - "functions": functions, - "context_strategy": ContextStrategyConfig(strategy=strategy), - "respond_immediately": entry_mode == "generate", - } - if entry_mode == "fixed_speech": - config["pre_actions"] = [ - { - "type": "workflow_fixed_speech", - "text": entry_speech, - "node_id": node_id, - "handler": self._play_fixed_speech, - } - ] - return config - - async def _play_fixed_speech(self, action: dict, _flow_manager: FlowManager) -> None: - """Play and persist Agent entry speech without creating an LLM turn.""" - await self._queue_visible_speech( - str(action.get("text") or ""), - source="workflow-fixed-reply", - node_id=str(action.get("node_id") or "") or None, + return self._require_agent_stage().node_config( + node_id, + functions=functions, + greeting_context_message=self._greeting_context_message, + leading_messages=leading_messages, ) + async def _after_node_activated( + self, + node_config: NodeConfig, + *, + triggering_user_text: str = "", + ) -> None: + """Publish activation and perform exactly one Agent entry behavior.""" + node_id = str(node_config.get("name") or "") + node_type = self._engine.node_type(node_id) + if node_type != "agent": + if node_type == "start": + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + return + + await self._emit_node_active(node_id) + data = self._engine.data(node_id) + entry_mode = str(data.get("entryMode") or "wait_user") + if entry_mode == "fixed_speech": + entry_speech = self._store.render(str(data.get("entrySpeech") or "")) + await self._queue_visible_speech( + entry_speech, + source="workflow-fixed-reply", + node_id=node_id, + ) + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + self._state.consume_user_turn() + return + + should_run = entry_mode == "generate" or bool(triggering_user_text) + if should_run: + self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) + await self._require_runtime().queue_frame(LLMRunFrame()) + return + + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + async def _queue_visible_speech( self, text: str, @@ -444,27 +383,11 @@ class WorkflowBrain(BaseBrain): source: str = "workflow-speech", node_id: str | None = None, ) -> None: - """Show and persist fixed workflow speech before sending it to TTS.""" - content = text.strip() - if not content: - return - self._store.record("agent", content) - runtime = self._require_runtime() - transcript_message = { - "type": "transcript", - "role": "assistant", - "content": content, - "timestamp": time_now_iso8601(), - "source": source, - **({"nodeId": node_id} if node_id else {}), - } - if self._client_ready: - await runtime.queue_frame( - OutputTransportMessageUrgentFrame(message=transcript_message) - ) - else: - self._pending_visible_speech_events.append(transcript_message) - await runtime.queue_frame(TTSSpeakFrame(content, append_to_context=False)) + await self._require_output().speak( + text, + source=source, + node_id=node_id, + ) def _passive_node_config( self, @@ -486,10 +409,19 @@ class WorkflowBrain(BaseBrain): self._tools.register_secrets(tool) async def handler(args, _flow_manager): + transition_id = self._state.transition_id try: result = await self._tools.execute(tool, dict(args or {})) except ToolExecutionError as exc: return {"status": "error", "message": str(exc)} + if ( + self._state.current_node_id != node_id + or self._state.transition_id != transition_id + ): + return { + "status": "stale", + "message": "工具完成时当前 Agent 已经切换，结果不再触发路由。", + } updated_variables = list(result.get("updated_variables") or []) if updated_variables: await self._emit_variables( @@ -504,7 +436,22 @@ class WorkflowBrain(BaseBrain): include_default=False, ) if edge: - return result, await self._follow_edge(edge) + next_config = await self._follow_edge( + edge, + triggering_user_text=( + self._state.pending_user_turn.text + if self._state.pending_user_turn + else "" + ), + ) + return result, self._flow_managed_transition_config( + next_config, + triggering_user_text=( + self._state.pending_user_turn.text + if self._state.pending_user_turn + else "" + ), + ) return result return FlowsFunctionSchema( @@ -516,6 +463,65 @@ class WorkflowBrain(BaseBrain): cancel_on_interruption=True, ) + def _flow_managed_transition_config( + self, + node_config: NodeConfig, + *, + triggering_user_text: str, + ) -> NodeConfig: + """Let FlowManager finish a tool result before activating its target. + + Function-returned transitions are the one place where FlowManager must + schedule the LLM run. Its pre-action only updates our explicit runtime + state; it never performs a second LLM run. + """ + node_id = str(node_config.get("name") or "") + if self._engine.node_type(node_id) != "agent": + return node_config + entry_mode = str( + self._engine.data(node_id).get("entryMode") or "wait_user" + ) + should_run = entry_mode == "generate" or bool(triggering_user_text) + configured = dict(node_config) + configured["respond_immediately"] = ( + should_run and entry_mode != "fixed_speech" + ) + configured["pre_actions"] = [ + { + "type": "workflow_function_transition_entry", + "node_id": node_id, + "entry_mode": entry_mode, + "should_run": should_run, + "handler": self._activate_from_flow_transition, + } + ] + return configured + + async def _activate_from_flow_transition( + self, + action: dict, + _flow_manager: FlowManager, + ) -> None: + """Apply visible entry state without manually queueing an LLM run.""" + node_id = str(action.get("node_id") or "") + await self._emit_node_active(node_id) + entry_mode = str(action.get("entry_mode") or "wait_user") + if entry_mode == "fixed_speech": + entry_speech = self._store.render( + str(self._engine.data(node_id).get("entrySpeech") or "") + ) + await self._queue_visible_speech( + entry_speech, + source="workflow-fixed-reply", + node_id=node_id, + ) + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + self._state.consume_user_turn() + elif action.get("should_run"): + self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) + else: + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None: stage = self._engine.agent_stage_config(node_id) knowledge_id = str(stage.knowledge_base_id or "") @@ -562,6 +568,7 @@ class WorkflowBrain(BaseBrain): *, triggering_user_text: str = "", ) -> NodeConfig: + self._state.begin_transition() leading_messages: list[dict[str, str]] = [] speech = self._engine.edge_transition_speech(edge) if speech: @@ -589,7 +596,8 @@ class WorkflowBrain(BaseBrain): triggering_user_text: str = "", ) -> NodeConfig: context_messages = list(leading_messages or []) - for _ in range(MAX_AUTOMATIC_HOPS): + for hop in range(MAX_AUTOMATIC_HOPS): + self._state.automatic_hops = hop node_type = self._engine.node_type(node_id) if node_type == "agent": await self._apply_agent_stage(node_id) @@ -611,6 +619,7 @@ class WorkflowBrain(BaseBrain): elif node_type == "handoff": await self._enter_handoff(node_id) elif node_type == "start": + self._state.enter(node_id, WorkflowStatus.WAITING_USER) await self._emit_node_active(node_id) else: raise RuntimeError(f"工作流指向未知节点:{node_id}") @@ -619,6 +628,7 @@ class WorkflowBrain(BaseBrain): edge = await self._select_edge(node_id) if not edge: return self._passive_node_config(node_id, context_messages) + self._state.begin_transition() speech = self._engine.edge_transition_speech(edge) if speech: content = self._store.render(speech).strip() @@ -636,6 +646,7 @@ class WorkflowBrain(BaseBrain): raise RuntimeError("工作流连续自动跳转超过安全上限") async def _enter_action(self, node_id: str) -> None: + self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_node_active(node_id) data = self._engine.data(node_id) tool_id = str(data.get("toolId") or "") @@ -665,6 +676,7 @@ class WorkflowBrain(BaseBrain): self._store.values["system__last_action_error"] = str(exc)[:2048] async def _enter_handoff(self, node_id: str) -> None: + self._state.enter(node_id, WorkflowStatus.HANDOFF) await self._emit_node_active(node_id) data = self._engine.data(node_id) message = self._store.render(str(data.get("message") or "")) @@ -685,6 +697,8 @@ class WorkflowBrain(BaseBrain): async def _enter_end(self, node_id: str) -> None: self._ended = True + self._state.enter(node_id, WorkflowStatus.ENDED) + self._state.finish() await self._emit_node_active(node_id) runtime = self._require_runtime() if runtime.set_knowledge_scope: @@ -705,27 +719,17 @@ class WorkflowBrain(BaseBrain): return runtime.call_end.begin("workflow_completed") if message: - runtime.call_end.arm_after_speech() await self._queue_visible_speech(message) + arm_tracked = getattr(runtime.call_end, "arm_after_tracked_speech", None) + if callable(arm_tracked): + await arm_tracked() + elif message: + runtime.call_end.arm_after_speech() else: await runtime.call_end.finish() async def _emit_node_active(self, node_id: str | None) -> None: - if node_id: - await self._require_runtime().queue_frame( - OutputTransportMessageUrgentFrame( - message={"type": "node-active", "nodeId": node_id} - ) - ) - - def _public_variables(self) -> dict[str, str | int | float | bool]: - """Return the browser-safe part of this session's variable state.""" - return { - name: value - for name, value in self._store.values.items() - if not name.startswith(("system__", "secret__")) - and isinstance(value, (str, int, float, bool)) - } + await self._require_output().emit_node_active(node_id) async def _emit_variables( self, @@ -735,21 +739,10 @@ class WorkflowBrain(BaseBrain): changed: list[str] | None = None, ) -> None: """Publish a safe snapshot so Workflow debug mirrors runtime state.""" - message: dict[str, Any] = { - "type": "workflow-variables", - "reason": reason, - "variables": self._public_variables(), - } - if node_id: - message["nodeId"] = node_id - if changed: - message["changed"] = [ - name - for name in changed - if not name.startswith(("system__", "secret__")) - ] - await self._require_runtime().queue_frame( - OutputTransportMessageUrgentFrame(message=message) + await self._require_output().emit_variables( + reason=reason, + node_id=node_id, + changed=changed, ) def _require_runtime(self) -> BrainRuntime: @@ -761,3 +754,20 @@ class WorkflowBrain(BaseBrain): if self._manager is None: raise RuntimeError("Workflow FlowManager 尚未初始化") return self._manager + + def _require_output(self) -> WorkflowOutput: + if self._output is None: + # A few focused unit tests bind BrainRuntime directly. Lazily create + # the output adapter so those tests exercise the same production path. + self._output = WorkflowOutput(self._store, self._require_runtime()) + return self._output + + def _require_agent_stage(self) -> WorkflowAgentStage: + if self._agent_stage is None: + self._agent_stage = WorkflowAgentStage( + cfg=self._cfg or AssistantConfig(type="workflow"), + engine=self._engine, + store=self._store, + runtime=self._require_runtime(), + ) + return self._agent_stage diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 1e9ef24..69a7783 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -73,7 +73,7 @@ NODE_SPECS: list[dict[str, Any]] = [ "icon": "Zap", "accent": "peach", "addable": True, - "constraints": {"minIncoming": 1, "minOutgoing": 1}, + "constraints": {"minIncoming": 1, "minOutgoing": 0}, "fields": [ {"key": "name", "label": "节点名称", "type": "text", "default": "Action"}, ], @@ -352,18 +352,19 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if mode == "expression": expression_errors = _validate_expression(data.get("expression")) errors.extend(f"边 {edge_id}:{item}" for item in expression_errors) - try: - priority = int(data.get("priority", 10)) - except (TypeError, ValueError): - errors.append(f"边 {edge_id} 的优先级必须是整数") - priority = 10 - if priority in priorities[source_id]: - errors.append(f"节点 {source_id} 的出边优先级不能重复:{priority}") - priorities[source_id].add(priority) if mode == "always": always_counts[source_id] += 1 if always_counts[source_id] > 1: errors.append(f"节点 {source_id} 最多只能有一条默认路径") + else: + try: + priority = int(data.get("priority", 10)) + except (TypeError, ValueError): + errors.append(f"边 {edge_id} 的优先级必须是整数") + priority = 10 + if priority in priorities[source_id]: + errors.append(f"节点 {source_id} 的条件边优先级不能重复:{priority}") + priorities[source_id].add(priority) incoming[target_id] += 1 outgoing[source_id] += 1 adj[source_id].append(target_id) @@ -387,6 +388,15 @@ 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}") + if ( + node.get("type") == "agent" + and outgoing[node_id] == 1 + and always_counts[node_id] == 1 + ): + errors.append( + f"Agent 节点 {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/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index 6e65872..99f66d7 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -18,6 +18,8 @@ class CallEndCoordinator: self._armed = False self._speaking = False self._response_speech_started = False + self._tracked_speeches = 0 + self._finish_after_tracked_speech = False self._finished = False self._reason = "completed" @@ -37,6 +39,16 @@ class CallEndCoordinator: """Wait for the next observed bot speech to finish.""" self._armed = True + def track_speech(self) -> None: + """Register one fixed utterance before its TTSSpeakFrame is queued.""" + self._tracked_speeches += 1 + + async def arm_after_tracked_speech(self) -> None: + """Finish after every already queued fixed utterance has played.""" + self._finish_after_tracked_speech = True + if self._tracked_speeches == 0: + await self.finish() + async def finish_after_current_speech(self, *, has_text: bool) -> None: """Finish now if speech is absent/done, otherwise wait for its stop.""" if not has_text: @@ -59,7 +71,15 @@ class CallEndCoordinator: self._response_speech_started = True elif isinstance(frame, BotStoppedSpeakingFrame) and self._speaking: self._speaking = False - if self._armed: + if self._tracked_speeches > 0: + self._tracked_speeches -= 1 + if ( + self._finish_after_tracked_speech + and self._tracked_speeches == 0 + ): + logger.info("所有工作流结束语播报完毕,挂断通话") + await self.finish() + elif self._armed: logger.info("结束语播报完毕,挂断通话") await self.finish() diff --git a/backend/services/workflow/__init__.py b/backend/services/workflow/__init__.py new file mode 100644 index 0000000..ad04681 --- /dev/null +++ b/backend/services/workflow/__init__.py @@ -0,0 +1,19 @@ +"""Small, explicit building blocks for the local Workflow runtime.""" + +from services.workflow.models import ( + EdgeEvaluation, + LLMRouteResult, + RouteStatus, + UserTurn, + WorkflowRuntimeState, + WorkflowStatus, +) + +__all__ = [ + "EdgeEvaluation", + "LLMRouteResult", + "RouteStatus", + "UserTurn", + "WorkflowRuntimeState", + "WorkflowStatus", +] diff --git a/backend/services/workflow/agent.py b/backend/services/workflow/agent.py new file mode 100644 index 0000000..0ab9c22 --- /dev/null +++ b/backend/services/workflow/agent.py @@ -0,0 +1,134 @@ +"""Resolve and apply one Workflow Agent's complete stage configuration.""" + +from __future__ import annotations + +from copy import deepcopy + +from models import AssistantConfig +from pipecat.flows import ContextStrategy, ContextStrategyConfig, NodeConfig +from pipecat.frames.frames import LLMUpdateSettingsFrame +from pipecat.services.settings import LLMSettings + +from services.brains.base import BrainRuntime +from services.runtime_variables import DynamicVariableStore +from services.workflow_engine import WorkflowEngine +from services.workflow_router import WorkflowLLMRouter + + +AGENT_STAGE_INSTRUCTION = ( + "工作流路由已在用户一轮输入结束时完成。只执行当前阶段任务，" + "不要自行解释、模拟或宣布节点切换。" +) + + +class WorkflowAgentStage: + """Translate graph-level Agent data into Pipecat Flows configuration.""" + + def __init__( + self, + *, + cfg: AssistantConfig, + engine: WorkflowEngine, + store: DynamicVariableStore, + runtime: BrainRuntime, + ) -> None: + self._cfg = cfg + self._engine = engine + self._store = store + self._runtime = runtime + + def role_message(self, node_id: str) -> str: + stage_prompt = self._engine.prompt_for(node_id, self._store) + return ( + f"{stage_prompt}\n\n[工作流执行规则]\n" + f"{AGENT_STAGE_INSTRUCTION}" + ) + + async def refresh_prompt(self, node_id: str) -> None: + await self._runtime.queue_frame( + LLMUpdateSettingsFrame( + delta=LLMSettings( + system_instruction=self.role_message(node_id) + ) + ) + ) + + def router_for_node( + self, + node_id: str, + default_router: WorkflowLLMRouter, + ) -> WorkflowLLMRouter: + resource_id = self._engine.agent_stage_config(node_id).llm_resource_id + resource = self._cfg.workflow_model_resources.get(resource_id) + if not resource: + return default_router + from services.pipecat.service_factory import config_with_resource + + return WorkflowLLMRouter(config_with_resource(self._cfg, resource)) + + async def apply(self, node_id: str) -> None: + stage = self._engine.agent_stage_config(node_id) + if self._runtime.set_input_enabled: + self._runtime.set_input_enabled(True) + if self._runtime.apply_turn_config: + await self._runtime.apply_turn_config( + stage.enable_interrupt, + stage.turn_config, + ) + if self._runtime.switch_services: + await self._runtime.switch_services( + stage.llm_resource_id or None, + stage.asr_resource_id or None, + stage.tts_resource_id or None, + ) + if self._runtime.set_knowledge_scope: + self._runtime.set_knowledge_scope( + { + "knowledge_base_id": stage.knowledge_base_id, + "mode": stage.knowledge_mode, + "top_n": stage.knowledge_top_n, + "score_threshold": stage.knowledge_score_threshold, + } + ) + + def node_config( + self, + node_id: str, + *, + functions: list, + greeting_context_message: dict[str, str] | None, + leading_messages: list[dict[str, str]] | None = None, + ) -> NodeConfig: + data = self._engine.data(node_id) + entry_mode = str(data.get("entryMode") or "wait_user") + entry_speech = self._store.render(str(data.get("entrySpeech") or "")) + strategy = ( + ContextStrategy.RESET + if data.get("contextPolicy") == "fresh" + else ContextStrategy.APPEND + ) + greeting_messages = ( + [deepcopy(greeting_context_message)] + if strategy == ContextStrategy.RESET and greeting_context_message + else [] + ) + fixed_reply_messages = ( + [{"role": "assistant", "content": entry_speech}] + if entry_mode == "fixed_speech" and entry_speech + else [] + ) + return { + "name": node_id, + "role_message": self.role_message(node_id), + "task_messages": [ + *greeting_messages, + *(leading_messages or []), + *fixed_reply_messages, + ], + "functions": functions, + "context_strategy": ContextStrategyConfig(strategy=strategy), + # Direct node activations let WorkflowRuntime decide whether to run + # the LLM. A Pipecat function transition may explicitly override + # this because FlowManager must coordinate the tool result first. + "respond_immediately": False, + } diff --git a/backend/services/workflow/models.py b/backend/services/workflow/models.py new file mode 100644 index 0000000..89d0fdf --- /dev/null +++ b/backend/services/workflow/models.py @@ -0,0 +1,92 @@ +"""Readable runtime values shared by Workflow orchestration modules.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + + +class WorkflowStatus(StrEnum): + """The small set of states useful to operators and future debug tooling.""" + + STARTING = "starting" + WAITING_USER = "waiting_user" + ROUTING = "routing" + RUNNING_AGENT = "running_agent" + RUNNING_ACTION = "running_action" + HANDOFF = "handoff" + ENDED = "ended" + + +class RouteStatus(StrEnum): + """A routing error is deliberately different from a valid no-match.""" + + MATCHED = "matched" + NO_MATCH = "no_match" + ERROR = "error" + + +@dataclass(frozen=True) +class UserTurn: + """One committed user turn that may cross automatic Workflow nodes.""" + + id: int + text: str + + +@dataclass +class WorkflowRuntimeState: + """Mutable per-call state; graph definitions remain immutable.""" + + current_node_id: str + status: WorkflowStatus = WorkflowStatus.STARTING + pending_user_turn: UserTurn | None = None + transition_id: int = 0 + automatic_hops: int = 0 + ended: bool = False + _next_turn_id: int = 1 + + def begin_user_turn(self, text: str) -> UserTurn: + turn = UserTurn(id=self._next_turn_id, text=text) + self._next_turn_id += 1 + self.pending_user_turn = turn + self.automatic_hops = 0 + return turn + + def enter(self, node_id: str, status: WorkflowStatus) -> None: + self.current_node_id = node_id + self.status = status + + def begin_transition(self) -> int: + self.transition_id += 1 + return self.transition_id + + def consume_user_turn(self) -> UserTurn | None: + turn = self.pending_user_turn + self.pending_user_turn = None + return turn + + def finish(self) -> None: + self.ended = True + self.status = WorkflowStatus.ENDED + self.pending_user_turn = None + + +@dataclass(frozen=True) +class LLMRouteResult: + """Control-plane result returned by the small routing LLM.""" + + status: RouteStatus + function_name: str | None = None + error: str | None = None + + +@dataclass(frozen=True) +class EdgeEvaluation: + """Final graph-level decision after expression, LLM and default handling.""" + + status: RouteStatus + edge: dict[str, Any] | None = None + error: str | None = None + diff --git a/backend/services/workflow/output.py b/backend/services/workflow/output.py new file mode 100644 index 0000000..15e4361 --- /dev/null +++ b/backend/services/workflow/output.py @@ -0,0 +1,119 @@ +"""Client-visible Workflow output and fixed speech in one place.""" + +from __future__ import annotations + +from typing import Any + +from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame +from pipecat.utils.time import time_now_iso8601 + +from services.brains.base import BrainRuntime +from services.runtime_variables import DynamicVariableStore + + +class WorkflowOutput: + """Publish debug events and fixed speech without duplicating persistence.""" + + def __init__( + self, + store: DynamicVariableStore, + runtime: BrainRuntime, + ) -> None: + self._store = store + self._runtime = runtime + self._client_ready = False + self._pending_transcripts: list[dict[str, Any]] = [] + + async def mark_client_ready(self) -> None: + self._client_ready = True + pending = self._pending_transcripts + self._pending_transcripts = [] + for message in pending: + await self.emit(message) + + async def speak( + self, + text: str, + *, + source: str, + node_id: str | None = None, + ) -> None: + """Record, display and synthesize one Workflow-owned utterance.""" + content = text.strip() + if not content: + return + self._store.record("agent", content) + transcript = { + "type": "transcript", + "role": "assistant", + "content": content, + "timestamp": time_now_iso8601(), + "source": source, + **({"nodeId": node_id} if node_id else {}), + } + if self._client_ready: + await self.emit(transcript) + else: + self._pending_transcripts.append(transcript) + + track_speech = getattr(self._runtime.call_end, "track_speech", None) + if callable(track_speech): + track_speech() + await self._runtime.queue_frame( + TTSSpeakFrame(content, append_to_context=False) + ) + + async def emit_node_active(self, node_id: str | None) -> None: + if node_id: + await self.emit({"type": "node-active", "nodeId": node_id}) + + async def emit_variables( + self, + *, + reason: str, + node_id: str | None, + changed: list[str] | None = None, + ) -> None: + message: dict[str, Any] = { + "type": "workflow-variables", + "reason": reason, + "variables": self.public_variables(), + } + if node_id: + message["nodeId"] = node_id + if changed: + message["changed"] = [ + name + for name in changed + if not name.startswith(("system__", "secret__")) + ] + await self.emit(message) + + async def emit_error( + self, + message: str, + *, + node_id: str | None, + code: str = "workflow_runtime_error", + ) -> None: + payload: dict[str, Any] = { + "type": "workflow-error", + "code": code, + "message": message, + } + if node_id: + payload["nodeId"] = node_id + await self.emit(payload) + + def public_variables(self) -> dict[str, str | int | float | bool]: + return { + name: value + for name, value in self._store.values.items() + if not name.startswith(("system__", "secret__")) + and isinstance(value, (str, int, float, bool)) + } + + async def emit(self, message: dict[str, Any]) -> None: + await self._runtime.queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) diff --git a/backend/services/workflow/routing.py b/backend/services/workflow/routing.py new file mode 100644 index 0000000..5e9a5f7 --- /dev/null +++ b/backend/services/workflow/routing.py @@ -0,0 +1,97 @@ +"""Priority-aware Workflow edge evaluation.""" + +from __future__ import annotations + +from collections.abc import Callable + +from services.runtime_variables import DynamicVariableStore +from services.workflow.models import EdgeEvaluation, RouteStatus +from services.workflow_engine import WorkflowEngine +from services.workflow_router import WorkflowLLMRouter + + +class WorkflowEdgeEvaluator: + """Evaluate one node's paths without changing runtime state.""" + + def __init__( + self, + engine: WorkflowEngine, + store: DynamicVariableStore, + router_for_node: Callable[[str], WorkflowLLMRouter], + ) -> None: + self._engine = engine + self._store = store + self._router_for_node = router_for_node + + async def evaluate(self, node_id: str) -> EdgeEvaluation: + """Select the first matching conditional path, then the default path.""" + outgoing = self._engine.outgoing(node_id) + expression_edge = self._engine.deterministic_edge( + node_id, + self._store, + include_default=False, + ) + default_edge = next( + ( + edge + for edge in outgoing + if self._engine.edge_mode(edge) == "always" + ), + None, + ) + llm_edges = [ + edge for edge in outgoing if self._engine.edge_mode(edge) == "llm" + ] + + # A matching expression is a priority boundary. A later LLM condition + # cannot bypass it, while an earlier LLM condition still gets one chance. + if expression_edge: + expression_index = outgoing.index(expression_edge) + llm_edges = [ + edge + for edge in llm_edges + if outgoing.index(edge) < expression_index + ] + + if not llm_edges: + return self._matched_or_no_match(expression_edge or default_edge) + + result = 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__", "secret__")) + }, + edge_name=self._engine.edge_fn_name, + edge_description=self._engine.edge_description, + ) + if result.status == RouteStatus.ERROR: + return EdgeEvaluation(status=RouteStatus.ERROR, error=result.error) + if result.status == RouteStatus.NO_MATCH: + return self._matched_or_no_match(expression_edge or default_edge) + + selected = next( + ( + edge + for edge in llm_edges + if self._engine.edge_fn_name(edge) == result.function_name + ), + None, + ) + if selected is None: + return EdgeEvaluation( + status=RouteStatus.ERROR, + error="路由模型返回了不属于当前节点的连接", + ) + return EdgeEvaluation(status=RouteStatus.MATCHED, edge=selected) + + @staticmethod + def _matched_or_no_match(edge: dict | None) -> EdgeEvaluation: + if edge is None: + return EdgeEvaluation(status=RouteStatus.NO_MATCH) + return EdgeEvaluation(status=RouteStatus.MATCHED, edge=edge) + diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index d2b7c85..2d0951c 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -177,6 +177,11 @@ class WorkflowEngine: try: if operator == "exists": matched = exists if expected is not False else not exists + elif not exists: + # Missing values are never equal, unequal, greater or + # contained. Use the explicit ``exists`` operator when the + # distinction between missing and present is important. + matched = False elif operator == "eq": matched = actual == expected elif operator == "neq": diff --git a/backend/services/workflow_router.py b/backend/services/workflow_router.py index c76c9f5..11a09c1 100644 --- a/backend/services/workflow_router.py +++ b/backend/services/workflow_router.py @@ -15,6 +15,8 @@ from loguru import logger from models import AssistantConfig from openai import AsyncOpenAI +from services.workflow.models import LLMRouteResult, RouteStatus + STAY_ON_CURRENT_NODE = "workflow_stay_on_current_node" # Compatibility alias for callers saved before all source nodes supported LLM edges. @@ -38,10 +40,10 @@ class WorkflowLLMRouter: variables: dict[str, Any], edge_name: Callable[[dict[str, Any]], str], edge_description: Callable[[dict[str, Any]], str], - ) -> str | None: - """Return an edge function name, STAY, or None when routing failed.""" + ) -> LLMRouteResult: + """Return a typed match, no-match or technical error.""" if not edges: - return STAY_ON_CURRENT_NODE + return LLMRouteResult(status=RouteStatus.NO_MATCH) names = {edge_name(edge) for edge in edges} stay_name = STAY_ON_CURRENT_NODE @@ -116,16 +118,28 @@ class WorkflowLLMRouter: tool_calls = response.choices[0].message.tool_calls or [] if not tool_calls: logger.warning("Workflow 路由 LLM 未返回函数调用，留在当前节点") - return STAY_ON_CURRENT_NODE + return LLMRouteResult( + status=RouteStatus.ERROR, + error="路由模型没有返回函数调用", + ) selected = str(tool_calls[0].function.name or "") if selected == stay_name: - return STAY_ON_CURRENT_NODE + return LLMRouteResult(status=RouteStatus.NO_MATCH) if selected not in names: logger.warning(f"Workflow 路由 LLM 返回未知函数:{selected}") - return STAY_ON_CURRENT_NODE - return selected + return LLMRouteResult( + status=RouteStatus.ERROR, + error=f"路由模型返回未知函数:{selected}", + ) + return LLMRouteResult( + status=RouteStatus.MATCHED, + function_name=selected, + ) except Exception as exc: # noqa: BLE001 - routing failure must not end the call logger.warning(f"Workflow LLM 边判断失败，留在当前节点:{exc}") - return None + return LLMRouteResult( + status=RouteStatus.ERROR, + error=f"大模型路由失败:{type(exc).__name__}", + ) finally: await client.close() diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 4a1c5ac..badddd0 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -28,7 +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 +from services.workflow.models import LLMRouteResult, RouteStatus class FakeLLM: @@ -47,6 +47,7 @@ class FakeCallEnd: self.finished = False self.response_started = False self.waited_for_text: bool | None = None + self.tracked_speeches = 0 def begin(self, reason: str) -> None: self.ending = True @@ -58,6 +59,14 @@ class FakeCallEnd: def arm_after_speech(self) -> None: self.armed = True + def track_speech(self) -> None: + self.tracked_speeches += 1 + + async def arm_after_tracked_speech(self) -> None: + self.armed = True + if self.tracked_speeches == 0: + await self.finish() + async def finish_after_current_speech(self, *, has_text: bool) -> None: self.waited_for_text = has_text if has_text: @@ -824,7 +833,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def select_edge(self, **_kwargs): self.calls += 1 - return "goto_eat" + return LLMRouteResult( + status=RouteStatus.MATCHED, + function_name="goto_eat", + ) manager = FakeManager() router = FakeRouter() @@ -847,6 +859,45 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) self.assertIn("我想吃饭", brain._store.values["system__conversation_history"]) + async def test_start_expression_condition_also_waits_for_user_turn(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + ], + "edges": [ + { + "id": "route", + "source": "start", + "target": "agent", + "data": { + "mode": "expression", + "priority": 10, + "expression": { + "combinator": "and", + "rules": [ + { + "variable": "route", + "operator": "eq", + "value": "agent", + } + ], + }, + }, + } + ], + } + ) + brain._store.values["route"] = "agent" + + config = await brain._initial_node_config() + + self.assertEqual(config["name"], "start") + self.assertEqual(brain._state.status.value, "waiting_user") + async def test_automatic_node_can_follow_llm_condition(self): brain = WorkflowBrain( { @@ -896,7 +947,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): class FakeRouter: async def select_edge(self, **kwargs): self.node_name = kwargs["node_name"] - return "goto_to_agent" + return LLMRouteResult( + status=RouteStatus.MATCHED, + function_name="goto_to_agent", + ) router = FakeRouter() brain._router = router @@ -961,7 +1015,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): class FakeRouter: def __init__(self): - self.result = STAY_ON_CURRENT_NODE + self.result = LLMRouteResult(status=RouteStatus.NO_MATCH) self.edge_ids = [] async def select_edge(self, **kwargs): @@ -975,7 +1029,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(router.edge_ids, ["llm"]) self.assertEqual(selected["id"], "expression") - router.result = "goto_llm" + router.result = LLMRouteResult( + status=RouteStatus.MATCHED, + function_name="goto_llm", + ) selected = await brain._select_edge("agent") self.assertEqual(selected["id"], "llm") @@ -1183,21 +1240,19 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): brain._engine.data("agent")["entryMode"] = "generate" generate_config = brain._agent_config("agent") - self.assertTrue(generate_config["respond_immediately"]) + self.assertFalse(generate_config["respond_immediately"]) worker.frames.clear() await brain._manager.set_node_from_config(generate_config) - self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + await brain._after_node_activated(generate_config) + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) brain._engine.data("agent").update( {"entryMode": "fixed_speech", "entrySpeech": "您好，{{user_name}}"} ) fixed_config = brain._agent_config("agent") self.assertFalse(fixed_config["respond_immediately"]) - self.assertEqual( - fixed_config["pre_actions"][0]["type"], - "workflow_fixed_speech", - ) - self.assertEqual(fixed_config["pre_actions"][0]["text"], "您好，王先生") + self.assertNotIn("pre_actions", fixed_config) self.assertEqual( fixed_config["task_messages"], [ @@ -1216,10 +1271,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): {"role": "assistant", "content": "您好，王先生"}, ], ) - self.assertEqual(fixed_config["pre_actions"][0]["node_id"], "agent") worker.frames.clear() queued.clear() await brain._manager.set_node_from_config(fixed_config) + await brain._after_node_activated(fixed_config) self.assertTrue(any(isinstance(frame, TTSSpeakFrame) for frame in queued)) self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) context_updates = [ @@ -1263,7 +1318,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): class FakeRouter: async def select_edge(self, **_kwargs): - return "goto_finish" + return LLMRouteResult( + status=RouteStatus.MATCHED, + function_name="goto_finish", + ) brain._router = FakeRouter() handled = await brain.on_user_turn_end("我的需求已经说完了") diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py index 779f7fe..683fd3c 100644 --- a/backend/tests/test_call_lifecycle.py +++ b/backend/tests/test_call_lifecycle.py @@ -44,6 +44,20 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.reasons, ["tool_only"]) + async def test_workflow_end_waits_for_every_queued_fixed_speech(self): + self.coordinator.track_speech() + self.coordinator.track_speech() + self.coordinator.begin("workflow_completed") + await self.coordinator.arm_after_tracked_speech() + + await self.coordinator.observe(BotStartedSpeakingFrame()) + await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.assertEqual(self.reasons, []) + + await self.coordinator.observe(BotStartedSpeakingFrame()) + await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.assertEqual(self.reasons, ["workflow_completed"]) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_router.py b/backend/tests/test_workflow_router.py index 67d7766..ae84e6c 100644 --- a/backend/tests/test_workflow_router.py +++ b/backend/tests/test_workflow_router.py @@ -5,6 +5,7 @@ from types import SimpleNamespace from unittest.mock import patch from models import AssistantConfig +from services.workflow.models import RouteStatus from services.workflow_router import WorkflowLLMRouter @@ -62,7 +63,8 @@ class WorkflowLLMRouterTest(unittest.IsolatedAsyncioTestCase): edge_description=lambda _edge: "用户已经回答姓名", ) - self.assertEqual(selected, "goto_age") + self.assertEqual(selected.status, RouteStatus.MATCHED) + self.assertEqual(selected.function_name, "goto_age") self.assertEqual(requests[0]["tool_choice"], "required") self.assertEqual( [message["role"] for message in requests[0]["messages"]], diff --git a/backend/tests/test_workflow_routing.py b/backend/tests/test_workflow_routing.py new file mode 100644 index 0000000..618f304 --- /dev/null +++ b/backend/tests/test_workflow_routing.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import unittest + +from services.runtime_variables import DynamicVariableStore +from services.workflow.models import LLMRouteResult, RouteStatus +from services.workflow.routing import WorkflowEdgeEvaluator +from services.workflow_engine import WorkflowEngine + + +def routing_graph() -> dict: + return { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {"name": "Agent"}}, + {"id": "matched", "type": "end", "data": {}}, + {"id": "fallback", "type": "end", "data": {}}, + ], + "edges": [ + { + "id": "condition", + "source": "agent", + "target": "matched", + "data": { + "mode": "llm", + "priority": 10, + "condition": "用户要求结束", + }, + }, + { + "id": "default", + "source": "agent", + "target": "fallback", + "data": {"mode": "always", "priority": 10}, + }, + ], + } + + +class WorkflowEdgeEvaluatorTest(unittest.IsolatedAsyncioTestCase): + async def test_router_error_does_not_take_default_path(self): + class ErrorRouter: + async def select_edge(self, **_kwargs): + return LLMRouteResult( + status=RouteStatus.ERROR, + error="provider unavailable", + ) + + evaluator = WorkflowEdgeEvaluator( + WorkflowEngine(routing_graph()), + DynamicVariableStore({}), + lambda _node_id: ErrorRouter(), + ) + + decision = await evaluator.evaluate("agent") + + self.assertEqual(decision.status, RouteStatus.ERROR) + self.assertIsNone(decision.edge) + + async def test_explicit_no_match_takes_default_path(self): + class NoMatchRouter: + async def select_edge(self, **_kwargs): + return LLMRouteResult(status=RouteStatus.NO_MATCH) + + evaluator = WorkflowEdgeEvaluator( + WorkflowEngine(routing_graph()), + DynamicVariableStore({}), + lambda _node_id: NoMatchRouter(), + ) + + decision = await evaluator.evaluate("agent") + + self.assertEqual(decision.status, RouteStatus.MATCHED) + self.assertEqual(decision.edge["id"], "default") + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 69f68dd..99ac888 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -211,7 +211,7 @@ class WorkflowGraphTests(unittest.TestCase): "smart_turn", ) - def test_start_agent_and_handoff_may_have_no_outgoing_edge(self): + def test_start_agent_action_and_handoff_may_have_no_outgoing_edge(self): terminal_graphs = [ { "specVersion": 3, @@ -277,13 +277,47 @@ class WorkflowGraphTests(unittest.TestCase): } ], } + self.assertEqual(validate_graph(action_without_exit), []) + + def test_agent_cannot_have_only_one_default_path(self): + graph = { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "agent", "type": "agent", "data": {}}, + {"id": "end", "type": "end", "data": {}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, + { + "id": "leave", + "source": "agent", + "target": "end", + "data": {"mode": "always", "priority": 0}, + }, + ], + } + self.assertTrue( any( - "action 的出边不能少于 1" in error - for error in validate_graph(action_without_exit) + "Agent 节点 agent 不能只有默认路径" in error + for error in validate_graph(graph) ) ) + graph["edges"][1]["data"] = { + "mode": "llm", + "priority": 10, + "condition": "当前阶段已经完成", + } + self.assertEqual(validate_graph(graph), []) + def test_all_source_nodes_allow_multiple_conditional_paths(self): expression = { "combinator": "and", @@ -448,6 +482,41 @@ class WorkflowGraphTests(unittest.TestCase): self.assertNotIn("服务 王先生", custom_prompt) self.assertIn("处理订单", custom_prompt) + def test_missing_variable_only_matches_explicit_exists_rule(self): + engine = WorkflowEngine(valid_graph()) + values = {} + + self.assertFalse( + engine.expression_matches( + { + "combinator": "and", + "rules": [ + { + "variable": "order_status", + "operator": "neq", + "value": "paid", + } + ], + }, + values, + ) + ) + self.assertTrue( + engine.expression_matches( + { + "combinator": "and", + "rules": [ + { + "variable": "order_status", + "operator": "exists", + "value": False, + } + ], + }, + values, + ) + ) + if __name__ == "__main__": unittest.main() diff --git a/frontend/src/components/assistant-editor/use-workflow-editor-state.ts b/frontend/src/components/assistant-editor/use-workflow-editor-state.ts index 65a7678..84fcc8e 100644 --- a/frontend/src/components/assistant-editor/use-workflow-editor-state.ts +++ b/frontend/src/components/assistant-editor/use-workflow-editor-state.ts @@ -1,14 +1,19 @@ "use client"; -import { useState } from "react"; +import { useCallback, useMemo, useState } from "react"; import type { WorkflowSettings } from "@/components/workflow/types"; import { defaultGraph, type WorkflowGraph } from "@/components/workflow/specs"; import type { DynamicVariableDefinition } from "@/lib/api"; -import { defaultTurnConfig } from "@/lib/turn-config"; -function initialWorkflowSettings(): WorkflowSettings { - const graph = defaultGraph(); +type WorkflowPanel = + | { type: "debug" } + | { type: "settings" } + | { type: "node"; nodeId: string } + | { type: "edge"; edgeId: string } + | null; + +function settingsFromGraph(graph: WorkflowGraph): WorkflowSettings { return { globalPrompt: graph.settings.globalPrompt, llm: graph.settings.defaultLlmResourceId, @@ -21,8 +26,8 @@ function initialWorkflowSettings(): WorkflowSettings { topN: graph.settings.knowledgeTopN, scoreThreshold: graph.settings.knowledgeScoreThreshold, }, - allowInterrupt: true, - turnConfig: defaultTurnConfig(), + allowInterrupt: graph.settings.enableInterrupt, + turnConfig: graph.settings.turnConfig, }; } @@ -32,19 +37,67 @@ export function useWorkflowEditorState() { const [workflowGraph, setWorkflowGraph] = useState(() => defaultGraph(), ); - const [workflowSettings, setWorkflowSettings] = useState( - initialWorkflowSettings, + const workflowSettings = useMemo( + () => settingsFromGraph(workflowGraph), + [workflowGraph], ); + const setWorkflowSettings = useCallback((settings: WorkflowSettings) => { + setWorkflowGraph((graph) => ({ + ...graph, + settings: { + globalPrompt: settings.globalPrompt, + defaultLlmResourceId: settings.llm ?? "", + defaultAsrResourceId: settings.asr ?? "", + defaultTtsResourceId: settings.tts ?? "", + toolIds: settings.toolIds, + knowledgeBaseId: settings.knowledgeBaseId, + knowledgeMode: settings.knowledgeRetrievalConfig.mode, + knowledgeTopN: settings.knowledgeRetrievalConfig.topN, + knowledgeScoreThreshold: + settings.knowledgeRetrievalConfig.scoreThreshold, + enableInterrupt: settings.allowInterrupt, + turnConfig: settings.turnConfig, + }, + })); + }, []); const [workflowDynamicVariableDefinitions, setWorkflowDynamicVariableDefinitions] = useState>({}); - const [workflowDebugOpen, setWorkflowDebugOpen] = useState(false); - const [workflowSettingsOpen, setWorkflowSettingsOpen] = useState(false); - const [workflowEditingNodeId, setWorkflowEditingNodeId] = useState( - null, - ); - const [workflowEditingEdgeId, setWorkflowEditingEdgeId] = useState( - null, - ); + const [workflowPanel, setWorkflowPanel] = useState(null); + const workflowDebugOpen = workflowPanel?.type === "debug"; + const workflowSettingsOpen = workflowPanel?.type === "settings"; + const workflowEditingNodeId = + workflowPanel?.type === "node" ? workflowPanel.nodeId : null; + const workflowEditingEdgeId = + workflowPanel?.type === "edge" ? workflowPanel.edgeId : null; + + const setWorkflowDebugOpen = useCallback((open: boolean) => { + setWorkflowPanel((panel) => + open ? { type: "debug" } : panel?.type === "debug" ? null : panel, + ); + }, []); + const setWorkflowSettingsOpen = useCallback((open: boolean) => { + setWorkflowPanel((panel) => + open ? { type: "settings" } : panel?.type === "settings" ? null : panel, + ); + }, []); + const setWorkflowEditingNodeId = useCallback((nodeId: string | null) => { + setWorkflowPanel((panel) => + nodeId + ? { type: "node", nodeId } + : panel?.type === "node" + ? null + : panel, + ); + }, []); + const setWorkflowEditingEdgeId = useCallback((edgeId: string | null) => { + setWorkflowPanel((panel) => + edgeId + ? { type: "edge", edgeId } + : panel?.type === "edge" + ? null + : panel, + ); + }, []); const [activeNodeId, setActiveNodeId] = useState(null); const [dynamicVariablesOpen, setDynamicVariablesOpen] = useState(false); @@ -65,6 +118,7 @@ export function useWorkflowEditorState() { setWorkflowEditingNodeId, workflowEditingEdgeId, setWorkflowEditingEdgeId, + workflowPanel, activeNodeId, setActiveNodeId, dynamicVariablesOpen, diff --git a/frontend/src/components/workflow/AddNodeDialog.tsx b/frontend/src/components/workflow/AddNodeDialog.tsx new file mode 100644 index 0000000..c296ec3 --- /dev/null +++ b/frontend/src/components/workflow/AddNodeDialog.tsx @@ -0,0 +1,116 @@ +"use client"; + +import { Plus } from "lucide-react"; + +import { + Dialog, + DialogContent, + DialogDescription, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; + +import { accentVar, type RuntimeNodeSpec } from "./specs"; + +export function AddNodeDialog({ + open, + connectingFromSource, + specs, + canAdd, + onOpenChange, + onSelect, +}: { + open: boolean; + connectingFromSource: boolean; + specs: RuntimeNodeSpec[]; + canAdd: (spec: RuntimeNodeSpec) => boolean; + onOpenChange: (open: boolean) => void; + onSelect: (spec: RuntimeNodeSpec) => void; +}) { + return ( + + + +
+
+
+ +
+
+
节点目录
+ + 添加节点 + + + {connectingFromSource + ? "选择节点类型，将在当前节点下方创建并自动连接。" + : "选择节点类型并添加到画布中央，随后可编辑内容并建立连线。"} + +
+
+ + +
+ {specs.length === 0 && ( +

+ 暂无可添加的节点类型。 +

+ )} + {specs.map((spec) => { + const Icon = spec.icon; + const allowed = canAdd(spec); + return ( + + ); + })} +
+ +
+ ); +} diff --git a/frontend/src/components/workflow/WorkflowCanvas.tsx b/frontend/src/components/workflow/WorkflowCanvas.tsx index 18555c2..f5827f0 100644 --- a/frontend/src/components/workflow/WorkflowCanvas.tsx +++ b/frontend/src/components/workflow/WorkflowCanvas.tsx @@ -16,7 +16,7 @@ import { useNodesState, useReactFlow, } from "@xyflow/react"; -import { Braces, Plus, Settings2, X } from "lucide-react"; +import { Braces, Plus, Settings2 } from "lucide-react"; import { useCallback, useEffect, @@ -26,13 +26,6 @@ import { } from "react"; import { Button } from "@/components/ui/button"; -import { - Dialog, - DialogContent, - DialogDescription, - DialogHeader, - DialogTitle, -} from "@/components/ui/dialog"; import { Tooltip, TooltipContent, @@ -40,6 +33,7 @@ import { TooltipTrigger, } from "@/components/ui/tooltip"; +import { AddNodeDialog } from "./AddNodeDialog"; import { edgeTypes } from "./ConditionEdge"; import { hasDefaultPath, newEdgeData } from "./edge-rules"; import { @@ -52,8 +46,8 @@ import { nodeTypes } from "./GenericNode"; import { EdgeSettingsPanel } from "./panels/EdgeSettingsPanel"; import { GlobalSettingsPanel } from "./panels/GlobalSettingsPanel"; import { NodeSettingsPanel } from "./panels/NodeSettingsPanel"; +import { WorkflowSidePanel } from "./WorkflowSidePanel"; import { - accentVar, defaultGraph, type NodeSpecMap, type RuntimeNodeSpec, @@ -195,7 +189,9 @@ export function WorkflowCanvas({ settings.tts, settings.toolIds, settings.knowledgeBaseId, - settings.knowledgeRetrievalConfig, + settings.knowledgeRetrievalConfig.mode, + settings.knowledgeRetrievalConfig.topN, + settings.knowledgeRetrievalConfig.scoreThreshold, settings.allowInterrupt, settings.turnConfig, ]); @@ -203,6 +199,9 @@ export function WorkflowCanvas({ const onConnect = useCallback( (connection: Connection) => { if (!connection.source || !connection.target) return; + const sourceType = nodes.find( + (node) => node.id === connection.source, + )?.type; setEdges((eds) => addEdge( { @@ -210,13 +209,13 @@ export function WorkflowCanvas({ id: `e-${connection.source}-${connection.target}-${Date.now()}`, type: "condition", animated: true, - data: newEdgeData(eds, connection.source), + data: newEdgeData(eds, connection.source, sourceType), }, eds, ), ); }, - [setEdges], + [nodes, setEdges], ); // 连线约束:不能连入开始节点(无入边句柄),不能自连。 @@ -277,7 +276,7 @@ export function WorkflowCanvas({ target: id, type: "condition", animated: true, - data: newEdgeData(currentEdges, source.id), + data: newEdgeData(currentEdges, source.id, source.type), }, currentEdges, ); @@ -577,9 +576,11 @@ export function WorkflowCanvas({ - {/* 添加节点弹窗 */} - { setAddOpen(open); if (!open) { @@ -587,131 +588,43 @@ export function WorkflowCanvas({ setAddPosition(null); } }} - > - - -
-
-
- -
-
-
- 节点目录 -
- - 添加节点 - - - {addSourceId - ? "选择节点类型，将在当前节点下方创建并自动连接。" - : "选择节点类型并添加到画布中央，随后可编辑内容并建立连线。"} - -
-
- -
- {addableSpecs.length === 0 ? ( -

- 暂无可添加的节点类型。 -

- ) : null} - {addableSpecs.map((spec) => { - const Icon = spec.icon; - const canAdd = canAddSpec(spec); - return ( - - ); - })} -
- -
+ onSelect={addNode} + /> {(debugOpen || settingsOpen || (editingNode && editingSpec) || editingEdge) && ( - + )} + )} diff --git a/frontend/src/components/workflow/WorkflowSidePanel.tsx b/frontend/src/components/workflow/WorkflowSidePanel.tsx new file mode 100644 index 0000000..823b373 --- /dev/null +++ b/frontend/src/components/workflow/WorkflowSidePanel.tsx @@ -0,0 +1,38 @@ +"use client"; + +import { X } from "lucide-react"; +import type { ReactNode } from "react"; + +export function WorkflowSidePanel({ + title, + closeLabel, + onClose, + children, +}: { + title?: string; + closeLabel?: string; + onClose?: () => void; + children: ReactNode; +}) { + return ( + + ); +} diff --git a/frontend/src/components/workflow/edge-rules.ts b/frontend/src/components/workflow/edge-rules.ts index a74d203..63aa6e5 100644 --- a/frontend/src/components/workflow/edge-rules.ts +++ b/frontend/src/components/workflow/edge-rules.ts @@ -31,11 +31,19 @@ export function nextEdgePriority(edges: Edge[], sourceId: string): number { } /** - * The first path from a node is deterministic. Further paths start as an - * incomplete LLM condition so the graph never silently gains two defaults. + * Agent nodes need a real condition before they can leave; a lone default path + * would consume the next user turn without letting the Agent answer it. + * Automatic nodes may use a default as their first deterministic path. */ -export function newEdgeData(edges: Edge[], sourceId: string): WorkflowEdgeData { +export function newEdgeData( + edges: Edge[], + sourceId: string, + sourceType?: string, +): WorkflowEdgeData { const priority = nextEdgePriority(edges, sourceId); + if (sourceType === "agent" && !hasDefaultPath(edges, sourceId)) { + return { mode: "llm", priority, condition: "" }; + } 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 af089f0..dfcd95b 100644 --- a/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx +++ b/frontend/src/components/workflow/panels/EdgeSettingsPanel.tsx @@ -27,10 +27,14 @@ import type { ExpressionRule, WorkflowEdgeData } from "../specs"; export function EdgeSettingsPanel({ edge, + sourceType, + isOnlyOutgoing, hasOtherDefaultPath, onChange, }: { edge: Edge; + sourceType?: string; + isOnlyOutgoing: boolean; hasOtherDefaultPath: boolean; onChange: (patch: WorkflowEdgeData) => void; }) { @@ -109,7 +113,10 @@ export function EdgeSettingsPanel({ { value: "always", label: "默认路径", - disabled: mode !== "always" && hasOtherDefaultPath, + disabled: + mode !== "always" && + (hasOtherDefaultPath || + (sourceType === "agent" && isOnlyOutgoing)), }, { value: "llm", label: "大模型判断" }, { value: "expression", label: "表达式" }, @@ -127,24 +134,31 @@ export function EdgeSettingsPanel({ 当前节点已经有一条默认路径；其它出边需要使用大模型判断或表达式。 )} - + )} + {mode !== "always" && ( + + )}