From f5c36a62aa1443cad4715640ef5249b810a3c8e1 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Mon, 3 Aug 2026 07:54:51 +0800 Subject: [PATCH] feat: make workflow messages resumable --- backend/services/brains/workflow_brain.py | 251 +++++++++++++---- backend/services/node_specs.py | 14 +- backend/services/workflow/agent.py | 12 +- backend/tests/test_brains.py | 254 ++++++++++++++---- backend/tests/test_workflow_v3.py | 19 +- .../src/components/workflow/GenericNode.tsx | 1 - .../components/workflow/WorkflowCanvas.tsx | 1 - .../workflow/panels/AgentNodePanel.tsx | 21 +- .../workflow/panels/NodeSettingsPanel.tsx | 21 +- frontend/src/components/workflow/specs.ts | 4 +- 10 files changed, 423 insertions(+), 175 deletions(-) diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index eb09cd7..064da76 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -5,7 +5,7 @@ from __future__ import annotations import asyncio from collections.abc import Awaitable from copy import deepcopy -from dataclasses import replace +from dataclasses import dataclass, replace from typing import Any from loguru import logger @@ -63,9 +63,29 @@ from services.workflow_router import WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 +@dataclass +class _MessageContinuation: + """Resume one Workflow path after its visible Message gate completes.""" + + token: int + node_id: str + context_messages: list[dict[str, str]] + triggering_user_text: str + task: asyncio.Task[None] | None = None + + class ConfiguredFlowManager(FlowManager): """Preserve Flow transitions while suppressing late async-tool replies.""" + ENTRY_ACTION_TYPE = "workflow_function_transition_entry" + + async def _set_node(self, node_id: str, node_config: NodeConfig) -> None: + """Notify Workflow only after FlowManager committed the active node.""" + await super()._set_node(node_id, node_config) + after_activation = node_config.get("workflow_after_activation") + if callable(after_activation): + await after_activation(node_id) + async def _create_transition_func(self, name, handler): transition = await super()._create_transition_func(name, handler) if not getattr(handler, "_suppress_followup_llm", False): @@ -136,6 +156,8 @@ class WorkflowBrain(BaseBrain): self._output: WorkflowOutput | None = None self._agent_stage: WorkflowAgentStage | None = None self._ended = False + self._next_message_token = 1 + self._pending_message: _MessageContinuation | None = None async def greeting(self, _cfg: AssistantConfig) -> str: """Workflow opening speech belongs to an explicit Message or Agent.""" @@ -179,6 +201,8 @@ class WorkflowBrain(BaseBrain): runtime=runtime, ) self._ended = False + self._next_message_token = 1 + self._pending_message = None self._manager = ConfiguredFlowManager( worker=runtime.worker, llm=runtime.llm, @@ -199,8 +223,7 @@ class WorkflowBrain(BaseBrain): raise RuntimeError("Workflow FlowManager 尚未初始化") node_config = await self._initial_node_config() - await self._manager.initialize(node_config) - await self._after_node_activated(node_config) + await self._activate_node_config(node_config, initialize=True) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") async def _initial_node_config(self) -> NodeConfig: @@ -298,8 +321,7 @@ class WorkflowBrain(BaseBrain): decision.edge, triggering_user_text=content, ) - await manager.set_node_from_config(next_config) - await self._after_node_activated( + await self._activate_node_config( next_config, triggering_user_text=content, ) @@ -390,9 +412,12 @@ class WorkflowBrain(BaseBrain): *, triggering_user_text: str = "", ) -> None: - """Publish activation and perform exactly one Agent entry behavior.""" + """Run the entry behavior owned by the newly active node.""" node_id = str(node_config.get("name") or "") node_type = self._engine.node_type(node_id) + if node_type == "message": + await self._activate_message_continuation(node_id) + return if node_type != "agent": if node_type == "start": self._state.enter(node_id, WorkflowStatus.WAITING_USER) @@ -401,17 +426,6 @@ class WorkflowBrain(BaseBrain): 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) @@ -420,6 +434,24 @@ class WorkflowBrain(BaseBrain): self._state.enter(node_id, WorkflowStatus.WAITING_USER) + async def _activate_node_config( + self, + node_config: NodeConfig, + *, + triggering_user_text: str = "", + initialize: bool = False, + ) -> None: + """Install one node and dispatch its entry behavior exactly once.""" + manager = self._require_manager() + if initialize: + await manager.initialize(node_config) + else: + await manager.set_node_from_config(node_config) + await self._after_node_activated( + node_config, + triggering_user_text=triggering_user_text, + ) + async def _queue_visible_speech( self, text: str, @@ -529,21 +561,25 @@ class WorkflowBrain(BaseBrain): state; it never performs a second LLM run. """ node_id = str(node_config.get("name") or "") - if self._engine.node_type(node_id) != "agent": + node_type = self._engine.node_type(node_id) + if node_type == "message": + configured = dict(node_config) + configured["workflow_after_activation"] = ( + self._activate_message_continuation + ) + return configured + if node_type != "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["respond_immediately"] = should_run configured["pre_actions"] = [ { - "type": "workflow_function_transition_entry", + "type": ConfiguredFlowManager.ENTRY_ACTION_TYPE, "node_id": node_id, - "entry_mode": entry_mode, "should_run": should_run, "handler": self._activate_from_flow_transition, } @@ -558,19 +594,7 @@ class WorkflowBrain(BaseBrain): """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"): + if action.get("should_run"): self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT) else: self._state.enter(node_id, WorkflowStatus.WAITING_USER) @@ -619,10 +643,11 @@ class WorkflowBrain(BaseBrain): self, edge: dict, *, + leading_messages: list[dict[str, str]] | None = None, triggering_user_text: str = "", ) -> NodeConfig: await self._begin_edge_transition(edge) - leading_messages: list[dict[str, str]] = [] + context_messages = list(leading_messages or []) speech = self._engine.edge_transition_speech(edge) if speech: content = self._store.render(speech).strip() @@ -632,12 +657,12 @@ class WorkflowBrain(BaseBrain): source="workflow-edge-transition", node_id=str(edge.get("target") or "") or None, ) - leading_messages.append( + context_messages.append( {"role": "assistant", "content": content} ) return await self._resolve_path( str(edge.get("target") or ""), - leading_messages=leading_messages, + leading_messages=context_messages, triggering_user_text=triggering_user_text, ) @@ -672,13 +697,12 @@ class WorkflowBrain(BaseBrain): if not outcome.should_route: return self._passive_node_config(node_id, context_messages) elif node_type == "message": - message_result = await self._enter_message(node_id) - if not message_result.succeeded: - return self._passive_node_config(node_id, context_messages) - if message_result.speech: - context_messages.append( - {"role": "assistant", "content": message_result.speech} - ) + self._prepare_message_continuation( + node_id, + context_messages=context_messages, + triggering_user_text=triggering_user_text, + ) + return self._passive_node_config(node_id, context_messages) elif node_type == "handoff": await self._enter_handoff(node_id) elif node_type == "start": @@ -708,6 +732,127 @@ class WorkflowBrain(BaseBrain): node_id = str(edge.get("target") or "") raise RuntimeError("工作流连续自动跳转超过安全上限") + def _prepare_message_continuation( + self, + node_id: str, + *, + context_messages: list[dict[str, str]], + triggering_user_text: str, + ) -> None: + """Save the path state without waiting inside the pipeline call stack.""" + token = self._next_message_token + self._next_message_token += 1 + self._pending_message = _MessageContinuation( + token=token, + node_id=node_id, + context_messages=[dict(message) for message in context_messages], + triggering_user_text=triggering_user_text, + ) + self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) + + async def _activate_message_continuation(self, node_id: str) -> None: + """Start a prepared Message only after FlowManager made it active.""" + continuation = self._pending_message + if ( + continuation is None + or continuation.node_id != node_id + or continuation.task is not None + ): + return + self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) + await self._emit_node_active(node_id) + runtime = self._require_runtime() + if runtime.set_input_enabled is not None: + runtime.set_input_enabled(False) + continuation.task = asyncio.create_task( + self._complete_message_continuation(continuation), + name=f"workflow-message-{node_id}-{continuation.token}", + ) + + async def _complete_message_continuation( + self, + continuation: _MessageContinuation, + ) -> None: + """Wait for playback/confirmation outside routing, then resume once.""" + try: + result = await self._enter_message( + continuation.node_id, + already_active=True, + input_already_blocked=True, + ) + if not result.succeeded: + if self._pending_message is continuation: + self._pending_message = None + return + await self._resume_message_continuation(continuation, result) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - keep callback failures visible + logger.exception(f"Message 节点续跑失败:{exc}") + if self._pending_message is continuation: + self._pending_message = None + self._state.enter( + continuation.node_id, + WorkflowStatus.WAITING_USER, + ) + runtime = self._require_runtime() + if runtime.set_input_enabled is not None: + runtime.set_input_enabled(True) + await self._require_output().emit_error( + "Message 节点完成后无法继续工作流", + node_id=continuation.node_id, + code="workflow_message_resume_error", + ) + + async def _resume_message_continuation( + self, + continuation: _MessageContinuation, + result: MessageStageResult, + ) -> None: + """Advance from the completed Message without blocking media frames.""" + async with self._turn_lock: + manager = self._require_manager() + if ( + self._ended + or self._pending_message is not continuation + or self._state.current_node_id != continuation.node_id + or str(manager.current_node or "") != continuation.node_id + ): + return + + self._pending_message = None + context_messages = [ + dict(message) for message in continuation.context_messages + ] + if result.speech: + context_messages.append( + {"role": "assistant", "content": result.speech} + ) + + if not self._engine.has_outgoing(continuation.node_id): + self._state.enter( + continuation.node_id, + WorkflowStatus.WAITING_USER, + ) + return + edge = await self._select_edge(continuation.node_id) + if not edge: + self._state.enter( + continuation.node_id, + WorkflowStatus.WAITING_USER, + ) + return + + next_config = await self._follow_edge( + edge, + leading_messages=context_messages, + triggering_user_text=continuation.triggering_user_text, + ) + await self._activate_node_config( + next_config, + triggering_user_text=continuation.triggering_user_text, + ) + async def _enter_action(self, node_id: str) -> ActionOutcome: self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_node_active(node_id) @@ -760,9 +905,16 @@ class WorkflowBrain(BaseBrain): await self._emit_action_outcome(node_id, outcome) return outcome - async def _enter_message(self, node_id: str) -> MessageStageResult: - self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) - await self._emit_node_active(node_id) + async def _enter_message( + self, + node_id: str, + *, + already_active: bool = False, + input_already_blocked: bool = False, + ) -> MessageStageResult: + if not already_active: + self._state.enter(node_id, WorkflowStatus.RUNNING_MESSAGE) + await self._emit_node_active(node_id) data = self._engine.data(node_id) runtime = self._require_runtime() speech = self._store.render(str(data.get("speech") or "")).strip() @@ -795,6 +947,7 @@ class WorkflowBrain(BaseBrain): node_id=node_id, ), set_input_enabled=runtime.set_input_enabled, + input_already_blocked=input_already_blocked, on_started=lambda: self._emit_trace( "message_started", nodeId=node_id, diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 51d0e1b..dbbb3f4 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -10,7 +10,7 @@ from typing import Any SPEC_VERSION = "3" NODE_TYPES = {"start", "agent", "message", "action", "handoff", "end"} EDGE_MODES = {"llm", "expression", "always"} -AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"} +AGENT_ENTRY_MODES = {"wait_user", "generate"} ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"} ACTION_USER_INPUT_POLICIES = {"queue", "block"} AUTOMATIC_NODE_TYPES = {"start", "message", "action", "handoff"} @@ -149,10 +149,11 @@ def _edge_data_v3(edge: dict) -> dict: def _normalize_agent_data(data: dict[str, Any]) -> None: - """Add v3 Agent defaults without changing existing node-level behavior.""" + """Keep Agent entry focused on conversation, not fixed interaction.""" data.setdefault("contextPolicy", "inherit") - data.setdefault("entryMode", "wait_user") - data.setdefault("entrySpeech", "") + if data.get("entryMode") not in AGENT_ENTRY_MODES: + data["entryMode"] = "wait_user" + data.pop("entrySpeech", None) if "inheritGlobalConfig" not in data: has_node_overrides = any( ( @@ -300,7 +301,6 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: "contextPolicy": "inherit", "inheritGlobalConfig": True, "entryMode": "wait_user", - "entrySpeech": "", }, } ) @@ -376,10 +376,6 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: entry_mode = data.get("entryMode", "wait_user") if entry_mode not in AGENT_ENTRY_MODES: errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}") - elif entry_mode == "fixed_speech" and not str( - data.get("entrySpeech") or "" - ).strip(): - errors.append(f"Agent 节点 {node_id} 的固定进入语不能为空") elif node_type == "message": data = node.get("data") or {} speech = data.get("speech") diff --git a/backend/services/workflow/agent.py b/backend/services/workflow/agent.py index 1fdc01a..adbd2a6 100644 --- a/backend/services/workflow/agent.py +++ b/backend/services/workflow/agent.py @@ -109,25 +109,15 @@ class WorkflowAgentStage: 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 ) - 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": [ - *(leading_messages or []), - *fixed_reply_messages, - ], + "task_messages": list(leading_messages or []), "functions": functions, "context_strategy": ContextStrategyConfig(strategy=strategy), # Direct node activations let WorkflowRuntime decide whether to run diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 5487d60..dba1b8f 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -965,7 +965,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertNotIn("fetch_user_image", custom_config["role_message"]) self.assertFalse(scopes[-1]["enabled"]) - async def test_initial_fixed_speech_starts_without_workflow_greeting(self): + async def test_initial_message_starts_without_workflow_greeting(self): brain = WorkflowBrain( { "specVersion": 3, @@ -976,20 +976,30 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): "type": "start", "data": {"greeting": "欢迎使用"}, }, + { + "id": "message", + "type": "message", + "data": { + "speech": "请问您怎么称呼？", + "showMessage": False, + }, + }, { "id": "agent", "type": "agent", - "data": { - "prompt": "收集用户信息", - "entryMode": "fixed_speech", - "entrySpeech": "请问您怎么称呼？", - }, + "data": {"prompt": "收集用户信息"}, }, ], "edges": [ { "id": "begin", "source": "start", + "target": "message", + "data": {"mode": "always", "priority": 0}, + }, + { + "id": "after_message", + "source": "message", "target": "agent", "data": {"mode": "always", "priority": 0}, } @@ -1032,15 +1042,17 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) await brain.on_connected() + self.assertEqual(brain._manager.current_node, "message") + for _ in range(3): + await asyncio.sleep(0) self.assertEqual(brain._manager.current_node, "agent") - fixed_speech_frames = [ + message_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, "请问您怎么称呼？") + self.assertEqual(len(message_speech_frames), 1) + self.assertEqual(message_speech_frames[0].text, "请问您怎么称呼？") - # Workflow no longer owns a greeting playback lifecycle. Stray generic - # transport notifications must not repeat Agent entry behavior. + # Stray generic greeting notifications must not replay the Message. await brain.on_greeting_finished() self.assertEqual( len([frame for frame in queued if isinstance(frame, TTSSpeakFrame)]), @@ -1533,6 +1545,175 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(result.succeeded) self.assertEqual(input_states, [False, True]) + async def test_message_between_agents_resumes_after_playback(self): + graph = { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "opening", + "type": "message", + "data": {"speech": "欢迎使用。"}, + }, + { + "id": "agent1", + "type": "agent", + "data": {"prompt": "收集基本信息"}, + }, + { + "id": "middle", + "type": "message", + "data": {"speech": "现在进入信息确认。"}, + }, + { + "id": "agent2", + "type": "agent", + "data": {"prompt": "确认信息", "contextPolicy": "fresh"}, + }, + { + "id": "end", + "type": "end", + "data": {"scope": "session"}, + }, + ], + "edges": [ + { + "id": "start-opening", + "source": "start", + "target": "opening", + "data": {"mode": "always"}, + }, + { + "id": "opening-agent1", + "source": "opening", + "target": "agent1", + "data": {"mode": "always"}, + }, + { + "id": "agent1-middle", + "source": "agent1", + "target": "middle", + "data": { + "mode": "llm", + "priority": 10, + "condition": "基本信息已经收集完成", + }, + }, + { + "id": "middle-agent2", + "source": "middle", + "target": "agent2", + "data": {"mode": "always"}, + }, + { + "id": "agent2-end", + "source": "agent2", + "target": "end", + "data": { + "mode": "llm", + "priority": 10, + "condition": "用户确认可以结束通话", + }, + }, + ], + } + brain = WorkflowBrain(graph) + queued = [] + input_states = [] + + class PlaybackCallEnd(FakeCallEnd): + def __init__(self): + super().__init__() + self.completions = [] + + def track_speech(self): + completion = asyncio.get_running_loop().create_future() + self.completions.append(completion) + return completion + + class FakeManager: + def __init__(self): + self.current_node = None + self.configs = [] + + async def initialize(self, config): + self.current_node = config["name"] + self.configs.append(config) + + async def set_node_from_config(self, config): + self.current_node = config["name"] + self.configs.append(config) + + async def queue_frame(frame): + queued.append(frame) + + call_end = PlaybackCallEnd() + manager = FakeManager() + + class MatchingRouter: + async def select_edge(self, **kwargs): + edge = kwargs["edges"][0] + return LLMRouteResult( + status=RouteStatus.MATCHED, + function_name=kwargs["edge_name"](edge), + ) + + brain._router = MatchingRouter() + 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=call_end, + set_input_enabled=input_states.append, + ) + brain._manager = manager + + await brain.on_connected() + await asyncio.sleep(0) + self.assertEqual(manager.current_node, "opening") + self.assertEqual(len(call_end.completions), 1) + + call_end.completions[0].set_result(None) + for _ in range(5): + await asyncio.sleep(0) + if manager.current_node == "agent1": + break + self.assertEqual(manager.current_node, "agent1") + + # The user-turn processor must return while the second Message is + # still waiting for its transport playback boundary. + await asyncio.wait_for( + brain.on_user_turn_end("基本信息已经收集完成"), + timeout=0.1, + ) + await asyncio.sleep(0) + self.assertEqual(manager.current_node, "middle") + self.assertEqual(len(call_end.completions), 2) + self.assertFalse(call_end.completions[1].done()) + + call_end.completions[1].set_result(None) + for _ in range(5): + await asyncio.sleep(0) + if manager.current_node == "agent2": + break + self.assertEqual(manager.current_node, "agent2") + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) + self.assertEqual( + manager.configs[-1]["task_messages"], + [ + {"role": "user", "content": "基本信息已经收集完成"}, + {"role": "assistant", "content": "现在进入信息确认。"}, + ], + ) + + await brain.on_assistant_text_end("agent2-turn", "信息确认完成", False) + await brain.on_user_turn_end("结束通话") + self.assertEqual(manager.current_node, "end") + self.assertTrue(call_end.finished) + async def test_nodes_without_outgoing_edges_remain_active(self): queued = [] @@ -2087,6 +2268,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(context.get_messages(), []) await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") + await brain.on_client_ready() variable_events = [ frame.message for frame in queued @@ -2150,58 +2332,14 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): 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.assertNotIn("pre_actions", fixed_config) - self.assertEqual( - fixed_config["task_messages"], - [{"role": "assistant", "content": "您好，王先生"}], - ) + brain._engine.data("agent")["entryMode"] = "wait_user" self.assertEqual( brain._agent_config( "agent", [{"role": "assistant", "content": "正在进入下一阶段"}], )["task_messages"], - [ - {"role": "assistant", "content": "正在进入下一阶段"}, - {"role": "assistant", "content": "您好，王先生"}, - ], + [{"role": "assistant", "content": "正在进入下一阶段"}], ) - 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 = [ - frame - for frame in worker.frames - if isinstance(frame, LLMMessagesUpdateFrame) - ] - self.assertEqual( - context_updates[-1].messages, - [{"role": "assistant", "content": "您好，王先生"}], - ) - self.assertFalse( - any( - isinstance(frame, OutputTransportMessageUrgentFrame) - and frame.message.get("source") == "workflow-fixed-reply" - for frame in queued - ) - ) - await brain.on_client_ready() - fixed_reply_events = [ - frame.message - for frame in queued - if isinstance(frame, OutputTransportMessageUrgentFrame) - and frame.message.get("source") == "workflow-fixed-reply" - ] - self.assertEqual(fixed_reply_events[0]["content"], "您好，王先生") - self.assertEqual(fixed_reply_events[0]["nodeId"], "agent") - self.assertIn("您好，王先生", brain._store.values["system__conversation_history"]) self.assertFalse( any( @@ -2252,7 +2390,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ] self.assertEqual( assistant_transcripts, - ["您好，王先生", "正在为你结束流程", "感谢来电"], + ["正在为你结束流程", "感谢来电"], ) self.assertIn( "正在为你结束流程", diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 1f56711..74ca90e 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -111,22 +111,27 @@ class WorkflowGraphTests(unittest.TestCase): self.assertTrue(body.vision_enabled) self.assertIsNone(body.vision_model_resource_id) - def test_agent_entry_mode_defaults_and_validation(self): + def test_agent_entry_modes_only_control_conversation_start(self): graph = valid_graph() normalized = normalize_graph(graph) agent = next(node for node in normalized["nodes"] if node["type"] == "agent") self.assertEqual(agent["data"]["entryMode"], "wait_user") - self.assertEqual(agent["data"]["entrySpeech"], "") + self.assertNotIn("entrySpeech", agent["data"]) self.assertTrue(agent["data"]["inheritGlobalConfig"]) self.assertEqual(agent["data"]["contextPolicy"], "fresh") - agent["data"]["entryMode"] = "fixed_speech" - self.assertTrue( - any("固定进入语不能为空" in error for error in validate_graph(normalized)) - ) - agent["data"]["entrySpeech"] = "您好，{{customer}}" + agent["data"]["entryMode"] = "generate" self.assertEqual(validate_graph(normalized), []) + agent["data"]["entryMode"] = "fixed_speech" + agent["data"]["entrySpeech"] = "您好，{{customer}}" + cleaned = normalize_graph(normalized) + cleaned_agent = next( + node for node in cleaned["nodes"] if node["type"] == "agent" + ) + self.assertEqual(cleaned_agent["data"]["entryMode"], "wait_user") + self.assertNotIn("entrySpeech", cleaned_agent["data"]) + def test_action_defaults_preserve_legacy_result_assignment_behavior(self): graph = valid_graph() graph["nodes"].extend( diff --git a/frontend/src/components/workflow/GenericNode.tsx b/frontend/src/components/workflow/GenericNode.tsx index 2ee64eb..288cf26 100644 --- a/frontend/src/components/workflow/GenericNode.tsx +++ b/frontend/src/components/workflow/GenericNode.tsx @@ -38,7 +38,6 @@ export function GenericNode({ id, type, data, selected }: NodeProps) { const entryModeLabel = { wait_user: "等待用户", generate: "立即回复", - fixed_speech: "固定进入语", }[nodeData.entryMode ?? "wait_user"]; const inheritsGlobal = nodeData.inheritGlobalConfig !== false; const meta = type === "agent" diff --git a/frontend/src/components/workflow/WorkflowCanvas.tsx b/frontend/src/components/workflow/WorkflowCanvas.tsx index 708494f..873165d 100644 --- a/frontend/src/components/workflow/WorkflowCanvas.tsx +++ b/frontend/src/components/workflow/WorkflowCanvas.tsx @@ -69,7 +69,6 @@ function defaultNodeData(spec: RuntimeNodeSpec): WorkflowNodeData { contextPolicy: "inherit", inheritGlobalConfig: true, entryMode: "wait_user", - entrySpeech: "", }); } else if (spec.type === "action") { Object.assign(data, { diff --git a/frontend/src/components/workflow/panels/AgentNodePanel.tsx b/frontend/src/components/workflow/panels/AgentNodePanel.tsx index 8fa8754..ad8249d 100644 --- a/frontend/src/components/workflow/panels/AgentNodePanel.tsx +++ b/frontend/src/components/workflow/panels/AgentNodePanel.tsx @@ -170,28 +170,13 @@ export function AgentNodePanel({ options={[ { value: "wait_user", label: "等待用户说话（默认）" }, { value: "generate", label: "立即让 LLM 回复" }, - { value: "fixed_speech", label: "播放固定进入语" }, ]} onChange={(value) => set("entryMode", value || "wait_user")} allowNone={false} /> - {draft.entryMode === "fixed_speech" && ( -