diff --git a/backend/schemas.py b/backend/schemas.py index 60b445a..2ecb038 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -100,9 +100,18 @@ class StartupAction(CamelModel): required: bool = True +class OpeningMessageConfig(CamelModel): + """Built-in Prompt opening interaction; it is not a reusable LLM tool.""" + + title: str = Field(default="重要提示", min_length=1, max_length=120) + message: str = Field(min_length=1, max_length=2000) + confirm_label: str = Field(default="确认", min_length=1, max_length=40) + + class StartupConfig(CamelModel): execution_mode: Literal["sequential"] = "sequential" actions: list[StartupAction] = Field(default_factory=list, max_length=5) + opening_message: OpeningMessageConfig | None = None @model_validator(mode="after") def validate_unique_action_ids(self): @@ -163,13 +172,17 @@ class AssistantUpsert(CamelModel): setattr(self, field, "") if "graph" not in allowed: self.graph = {} + if self.type == "workflow": + self.greeting = "" if self.type not in {"prompt", "workflow"}: self.tool_ids = [] self.dynamic_variable_definitions = {} if self.type != "prompt": self.startup = StartupConfig() - if self.runtime_mode == "realtime" and self.startup.actions: - raise ValueError("Prompt Realtime 模式暂不支持启动 Action") + if self.runtime_mode == "realtime" and ( + self.startup.actions or self.startup.opening_message is not None + ): + raise ValueError("Prompt Realtime 模式暂不支持启动阶段") # 外部托管大脑只能 cascade,拦住不兼容的 realtime if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES: raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式") diff --git a/backend/services/action_stage.py b/backend/services/action_stage.py new file mode 100644 index 0000000..1de949f --- /dev/null +++ b/backend/services/action_stage.py @@ -0,0 +1,100 @@ +"""Shared deterministic stage for one or more tool Actions.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any, Literal + +from models import RuntimeTool +from services.action_runtime import ActionOutcome, ActionRunner, ActionStatus + + +InputPolicy = Literal["queue", "block"] +OutcomeHook = Callable[["StageAction", ActionOutcome], Awaitable[None]] +StartedHook = Callable[[], Awaitable[None]] + + +@dataclass(frozen=True) +class StageAction: + """One deterministic tool invocation inside an Action stage.""" + + id: str + tool: RuntimeTool | None + arguments: dict[str, Any] = field(default_factory=dict) + result_assignments: dict[str, str] | None = None + required: bool = True + invocation_id: str | None = None + + +@dataclass(frozen=True) +class ActionStageSpec: + """Mode-independent description produced by Prompt or Workflow config.""" + + actions: tuple[StageAction, ...] = () + input_policy: InputPolicy = "queue" + + +@dataclass(frozen=True) +class ActionStageResult: + """Ordered Action outcomes; optional failures do not fail the stage.""" + + succeeded: bool + outcomes: tuple[ActionOutcome, ...] + + +class ActionStageRunner: + """Run deterministic tool Actions under one optional user-input gate.""" + + def __init__(self, actions: ActionRunner) -> None: + self._actions = actions + + async def run( + self, + spec: ActionStageSpec, + *, + set_input_enabled: Callable[[bool], None] | None = None, + input_already_blocked: bool = False, + release_input_on_failure: bool = True, + on_started: StartedHook | None = None, + on_outcome: OutcomeHook | None = None, + ) -> ActionStageResult: + input_setter = set_input_enabled + block_input = spec.input_policy == "block" and input_setter is not None + if block_input and not input_already_blocked: + input_setter(False) + + result: ActionStageResult | None = None + try: + if on_started is not None: + await on_started() + + outcomes: list[ActionOutcome] = [] + succeeded = True + for action in spec.actions: + outcome = await self._actions.execute( + action.tool, + action.arguments, + result_assignments=action.result_assignments, + invocation_id=action.invocation_id, + ) + outcomes.append(outcome) + if on_outcome is not None: + await on_outcome(action, outcome) + if outcome.status == ActionStatus.SUCCESS: + continue + if outcome.status == ActionStatus.CANCELLED or action.required: + succeeded = False + break + + result = ActionStageResult( + succeeded=succeeded, + outcomes=tuple(outcomes), + ) + return result + finally: + if block_input and ( + release_input_on_failure + or (result is not None and result.succeeded) + ): + input_setter(True) diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 13d0a56..bc92a5c 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -64,7 +64,7 @@ class CallEndPort(Protocol): def arm_after_speech(self) -> None: ... - def track_speech(self) -> None: ... + def track_speech(self) -> Awaitable[None] | None: ... async def arm_after_tracked_speech(self) -> None: ... diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index c52a207..c699b79 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +from collections.abc import Awaitable from typing import Any from uuid import uuid4 @@ -25,10 +26,18 @@ from services.brains.base import ( SessionVariableUpdate, ) from services.action_runtime import ( + ActionOutcome, ActionInvocationCancelled, ActionRunner, ActionStatus, ) +from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction +from services.fixed_speech import FixedSpeechOutput +from services.message_stage import ( + MessageDisplaySpec, + MessageStageRunner, + MessageStageSpec, +) from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool @@ -50,17 +59,31 @@ class PromptBrain(BaseBrain): self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store) self._actions = ActionRunner(self._tools) + self._action_stages = ActionStageRunner(self._actions) + self._message_stages = MessageStageRunner() self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._runtime: BrainRuntime | None = None + self._output: FixedSpeechOutput | None = None self._waiting_for_generated_end_speech = False - self._greeting_finished = True self._preflight_finished = False self._opening_started = False self._opening_finished = False + self._opening_input_blocked = False self._startup_failed = False async def greeting(self, cfg: AssistantConfig) -> str: - return self._store.render(cfg.greeting) if self._dynamic_enabled else cfg.greeting + # The built-in opening Message owns the greeting so speech and the + # client dialog can start as one atomic stage. + if self._opening_message() is not None: + return "" + return self._render_greeting(cfg) + + def _render_greeting(self, cfg: AssistantConfig) -> str: + return ( + self._store.render(cfg.greeting) + if self._dynamic_enabled + else cfg.greeting + ) def system_prompt(self, cfg: AssistantConfig) -> str: return self._store.render(cfg.prompt) if self._dynamic_enabled else cfg.prompt @@ -77,12 +100,15 @@ class PromptBrain(BaseBrain): self._tools, is_session_ending=lambda: runtime.call_end.ending, ) + self._action_stages = ActionStageRunner(self._actions) + self._message_stages = MessageStageRunner(runtime.client_tools) + self._output = FixedSpeechOutput(self._store, runtime) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._waiting_for_generated_end_speech = False - self._greeting_finished = True self._preflight_finished = False self._opening_started = False - self._opening_finished = not bool(self._startup_actions("opening")) + self._opening_finished = not self._has_opening_stage() + self._opening_input_blocked = False self._startup_failed = False llm_tool_ids = ( set(cfg.llm_tool_ids) if cfg.llm_tool_ids is not None else None @@ -119,32 +145,83 @@ class PromptBrain(BaseBrain): self._preflight_finished = True async def on_connected(self, *, greeting_pending: bool = False) -> None: - self._greeting_finished = not greeting_pending if ( - self._startup_actions("opening") + self._has_opening_stage() and self._runtime is not None and self._runtime.set_input_enabled is not None ): self._runtime.set_input_enabled(False) + self._opening_input_blocked = True async def on_client_ready(self) -> None: + if self._output is not None: + await self._output.mark_client_ready() if self._opening_started or self._opening_finished or self._startup_failed: return self._opening_started = True + runtime = self._runtime + if runtime is None: + raise RuntimeError("PromptBrain 尚未初始化") + opening_message = self._opening_message() + opening_actions = self._startup_actions("opening") + speech = ( + self._render_greeting(self._cfg).strip() + if opening_message is not None + else "" + ) + if speech: + self.prepare_greeting_context(speech, runtime.context) try: - succeeded = await self._run_startup_actions("opening") + if opening_message is not None: + message_result = await self._message_stages.run( + self._opening_message_stage_spec(speech, opening_message), + speak=self._speak_opening, + set_input_enabled=runtime.set_input_enabled, + input_already_blocked=self._opening_input_blocked, + release_input_on_success=not bool(opening_actions), + release_input_on_failure=False, + ) + if not message_result.succeeded: + await self._fail_opening( + message_result.error or "开场消息显示失败" + ) + return + + if opening_actions: + result = await self._action_stages.run( + self._opening_actions_stage_spec(), + set_input_enabled=runtime.set_input_enabled, + input_already_blocked=self._opening_input_blocked, + release_input_on_failure=False, + on_outcome=self._publish_opening_outcome, + ) + if not result.succeeded: + await self._fail_opening("必需的开场 Action 执行失败") + return except ActionInvocationCancelled: self._startup_failed = True raise - if not succeeded: - await self._fail_opening("必需的开场 Action 执行失败") - return self._opening_finished = True - self._release_startup_gate_if_ready() + self._opening_input_blocked = False - async def on_greeting_finished(self) -> None: - self._greeting_finished = True - self._release_startup_gate_if_ready() + async def _speak_opening(self, content: str) -> Awaitable[None] | None: + if self._output is None: + raise RuntimeError("Prompt 固定播报输出尚未初始化") + return await self._output.speak( + content, + source="prompt-opening-speech", + record_history=False, + ) + + def _opening_message(self) -> dict[str, Any] | None: + startup = self._cfg.startup if isinstance(self._cfg.startup, dict) else {} + value = startup.get("opening_message", startup.get("openingMessage")) + return value if isinstance(value, dict) else None + + def _has_opening_stage(self) -> bool: + return self._opening_message() is not None or bool( + self._startup_actions("opening") + ) def _startup_actions(self, phase: str) -> list[dict[str, Any]]: startup = self._cfg.startup if isinstance(self._cfg.startup, dict) else {} @@ -154,6 +231,78 @@ class PromptBrain(BaseBrain): if isinstance(action, dict) and action.get("phase", "opening") == phase ] + def _opening_actions_stage_spec(self) -> ActionStageSpec: + actions = tuple( + StageAction( + id=str(action.get("id") or "startup_action"), + tool=self._tool_by_id.get( + str(action.get("tool_id") or action.get("toolId") or "") + ), + arguments=action.get("arguments") or {}, + required=bool(action.get("required", True)), + invocation_id=self._actions.new_invocation_id(), + ) + for action in self._startup_actions("opening") + ) + return ActionStageSpec( + actions=actions, + input_policy="block", + ) + + def _opening_message_stage_spec( + self, + speech: str, + config: dict[str, Any], + ) -> MessageStageSpec: + return MessageStageSpec( + speech=speech, + display=MessageDisplaySpec( + title=self._store.render( + str(config.get("title") or "重要提示") + ).strip(), + message=self._store.render( + str(config.get("message") or "") + ).strip(), + confirm_label=self._store.render( + str( + config.get("confirm_label") + or config.get("confirmLabel") + or "确认" + ) + ).strip(), + ), + require_confirmation=True, + ) + + async def _publish_opening_outcome( + self, + action: StageAction, + outcome: ActionOutcome, + ) -> None: + if outcome.updated_variables: + self._refresh_prompt() + if self._runtime is not None: + await self._runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "startup-action-result", + "actionId": action.id, + "phase": "opening", + "outcome": outcome.trace_payload(), + } + ) + ) + if outcome.status == ActionStatus.FAILURE and action.required: + logger.warning( + f"必需的 Prompt opening Action 失败: " + f"action={action.id} error={outcome.error}" + ) + elif outcome.status == ActionStatus.FAILURE: + logger.warning( + f"忽略可选 Prompt opening Action 失败: " + f"action={action.id} error={outcome.error}" + ) + async def _run_startup_actions(self, phase: str) -> bool: for action in self._startup_actions(phase): action_id = str(action.get("id") or "startup_action") @@ -170,17 +319,6 @@ class PromptBrain(BaseBrain): ) if outcome.updated_variables: self._refresh_prompt() - if phase == "opening" and self._runtime is not None: - await self._runtime.queue_frame( - OutputTransportMessageUrgentFrame( - message={ - "type": "startup-action-result", - "actionId": action_id, - "phase": phase, - "outcome": outcome.trace_payload(), - } - ) - ) if outcome.status == ActionStatus.SUCCESS: continue if outcome.status == ActionStatus.CANCELLED: @@ -197,18 +335,6 @@ class PromptBrain(BaseBrain): ) return True - def _release_startup_gate_if_ready(self) -> None: - runtime = self._runtime - if ( - runtime is not None - and runtime.set_input_enabled is not None - and self._greeting_finished - and self._opening_finished - and not self._startup_failed - and not runtime.call_end.ending - ): - runtime.set_input_enabled(True) - async def _fail_opening(self, message: str) -> None: self._startup_failed = True runtime = self._runtime diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index d59ee50..eb09cd7 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +from collections.abc import Awaitable from copy import deepcopy from dataclasses import replace from typing import Any @@ -40,7 +41,14 @@ from services.action_runtime import ( ActionRunner, ActionStatus, ) +from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction from services.knowledge import search as search_knowledge +from services.message_stage import ( + MessageDisplaySpec, + MessageStageResult, + MessageStageRunner, + MessageStageSpec, +) from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool @@ -110,6 +118,8 @@ class WorkflowBrain(BaseBrain): self._store = DynamicVariableStore.from_config(cfg or AssistantConfig(type="workflow")) self._tools = ToolExecutor(self._store) self._actions = ActionRunner(self._tools) + self._action_stages = ActionStageRunner(self._actions) + self._message_stages = MessageStageRunner() self._tool_by_id: dict[str, RuntimeTool] = { tool.id: tool for tool in (cfg.tools if cfg else []) } @@ -126,11 +136,10 @@ class WorkflowBrain(BaseBrain): 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 - async def greeting(self, cfg: AssistantConfig) -> str: - return self._engine.greeting(self._store) or cfg.greeting + async def greeting(self, _cfg: AssistantConfig) -> str: + """Workflow opening speech belongs to an explicit Message or Agent.""" + return "" def system_prompt(self, cfg: AssistantConfig) -> str: return self._store.render(self._engine.global_prompt()) @@ -151,6 +160,8 @@ class WorkflowBrain(BaseBrain): self._tools, is_session_ending=lambda: runtime.call_end.ending, ) + self._action_stages = ActionStageRunner(self._actions) + self._message_stages = MessageStageRunner(runtime.client_tools) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) self._edge_evaluator = WorkflowEdgeEvaluator( @@ -168,8 +179,6 @@ class WorkflowBrain(BaseBrain): runtime=runtime, ) self._ended = False - self._greeting_context_message = None - self._startup_waiting_for_greeting = False self._manager = ConfiguredFlowManager( worker=runtime.worker, llm=runtime.llm, @@ -179,15 +188,6 @@ class WorkflowBrain(BaseBrain): ) self._manager.state["variables"] = self._store.values - def prepare_greeting_context( - self, - greeting: str, - context: LLMContext, - ) -> dict[str, str] | None: - message = super().prepare_greeting_context(greeting, context) - self._greeting_context_message = deepcopy(message) if message else None - 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) @@ -198,39 +198,11 @@ class WorkflowBrain(BaseBrain): if self._manager is None: raise RuntimeError("Workflow FlowManager 尚未初始化") - self._startup_waiting_for_greeting = greeting_pending - if greeting_pending: - # Keep the Workflow on Start until the transport confirms that the - # shared greeting has finished. This prevents an initial Agent's - # fixed speech (or generated reply) from racing the greeting. - await self._manager.initialize( - self._passive_node_config(self._engine.start_id) - ) - logger.info("工作流等待 Start 开场白播放完毕") - return - node_config = await self._initial_node_config() await self._manager.initialize(node_config) await self._after_node_activated(node_config) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") - async def on_greeting_finished(self) -> None: - """Enter the first node only after Start's greeting reaches playback end.""" - if not self._startup_waiting_for_greeting or self._ended: - return - self._startup_waiting_for_greeting = False - manager = self._require_manager() - if manager.current_node != self._engine.start_id: - return - - node_config = await self._initial_node_config() - if node_config.get("name") == self._engine.start_id: - 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: """Only a default-only Start advances before the first user turn.""" outgoing = self._engine.outgoing(self._engine.start_id) @@ -409,7 +381,6 @@ class WorkflowBrain(BaseBrain): return self._require_agent_stage().node_config( node_id, functions=functions, - greeting_context_message=self._greeting_context_message, leading_messages=leading_messages, ) @@ -455,8 +426,8 @@ class WorkflowBrain(BaseBrain): *, source: str = "workflow-speech", node_id: str | None = None, - ) -> None: - await self._require_output().speak( + ) -> Awaitable[None] | None: + return await self._require_output().speak( text, source=source, node_id=node_id, @@ -700,6 +671,14 @@ class WorkflowBrain(BaseBrain): outcome = await self._enter_action(node_id) 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} + ) elif node_type == "handoff": await self._enter_handoff(node_id) elif node_type == "start": @@ -735,29 +714,36 @@ class WorkflowBrain(BaseBrain): data = self._engine.data(node_id) runtime = self._require_runtime() invocation_id = self._actions.new_invocation_id() - block_user_input = data.get("userInputPolicy") == "block" - if block_user_input and runtime.set_input_enabled: - # Blocking only suppresses new audio/text input while the Action - # runs. It deliberately does not cancel the tool. The default - # queue policy leaves input enabled; the turn lock serializes any - # completed user turn until this automatic path has finished. - runtime.set_input_enabled(False) tool_id = str(data.get("toolId") or "") tool = self._tool_by_id.get(tool_id) try: - await self._emit_trace( - "action_started", - nodeId=node_id, - invocationId=invocation_id, - toolId=tool_id, - toolType=tool.type if tool else None, - ) - outcome = await self._actions.execute( - tool, - data.get("arguments") or {}, - result_assignments=self._action_result_assignments(data), - invocation_id=invocation_id, + stage_result = await self._action_stages.run( + ActionStageSpec( + actions=( + StageAction( + id=node_id, + tool=tool, + arguments=data.get("arguments") or {}, + result_assignments=self._action_result_assignments(data), + invocation_id=invocation_id, + ), + ), + input_policy=( + "block" + if data.get("userInputPolicy") == "block" + else "queue" + ), + ), + set_input_enabled=runtime.set_input_enabled, + on_started=lambda: self._emit_trace( + "action_started", + nodeId=node_id, + invocationId=invocation_id, + toolId=tool_id, + toolType=tool.type if tool else None, + ), ) + outcome = stage_result.outcomes[0] updated_variables = list(outcome.updated_variables) if updated_variables: await self._emit_variables( @@ -770,13 +756,74 @@ class WorkflowBrain(BaseBrain): self._set_last_action(outcome) await self._emit_action_outcome(node_id, outcome) raise - finally: - if block_user_input and runtime.set_input_enabled: - runtime.set_input_enabled(True) self._set_last_action(outcome) 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) + data = self._engine.data(node_id) + runtime = self._require_runtime() + speech = self._store.render(str(data.get("speech") or "")).strip() + show_message = bool(data.get("showMessage", False)) + require_confirmation = bool(data.get("requireConfirmation", False)) + display = ( + MessageDisplaySpec( + title=self._store.render( + str(data.get("title") or "重要提示") + ).strip(), + message=self._store.render( + str(data.get("message") or "") + ).strip(), + confirm_label=self._store.render( + str(data.get("confirmLabel") or "确认") + ).strip(), + ) + if show_message + else None + ) + result = await self._message_stages.run( + MessageStageSpec( + speech=speech, + display=display, + require_confirmation=require_confirmation, + ), + speak=lambda content: self._queue_visible_speech( + content, + source="workflow-message-speech", + node_id=node_id, + ), + set_input_enabled=runtime.set_input_enabled, + on_started=lambda: self._emit_trace( + "message_started", + nodeId=node_id, + hasSpeech=bool(speech), + showsMessage=show_message, + requiresConfirmation=require_confirmation, + ), + ) + if result.succeeded: + await self._emit_trace( + "message_completed", + nodeId=node_id, + action=result.action, + ) + return result + + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + await self._emit_trace( + "message_failed", + nodeId=node_id, + error=result.error or "Message 节点执行失败", + ) + await self._require_output().emit_error( + result.error or "Message 节点执行失败", + node_id=node_id, + code="workflow_message_error", + ) + return result + def _set_last_action(self, outcome: ActionOutcome) -> None: legacy_status = { ActionStatus.SUCCESS: "ok", diff --git a/backend/services/fixed_speech.py b/backend/services/fixed_speech.py new file mode 100644 index 0000000..da84f0f --- /dev/null +++ b/backend/services/fixed_speech.py @@ -0,0 +1,73 @@ +"""Shared client-visible output for deterministic fixed speech.""" + +from __future__ import annotations + +from collections.abc import Awaitable +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 FixedSpeechOutput: + """Display and synthesize fixed speech without waiting for playback.""" + + 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, + record_history: bool = True, + ) -> Awaitable[None] | None: + content = text.strip() + if not content: + return None + if record_history: + 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) + playback_completion: Awaitable[None] | None = None + if callable(track_speech): + playback_completion = track_speech() + await self._runtime.queue_frame( + TTSSpeakFrame(content, append_to_context=False) + ) + return playback_completion + + async def emit(self, message: dict[str, Any]) -> None: + await self._runtime.queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) diff --git a/backend/services/message_stage.py b/backend/services/message_stage.py new file mode 100644 index 0000000..6e5afb9 --- /dev/null +++ b/backend/services/message_stage.py @@ -0,0 +1,197 @@ +"""Deterministic speech and client-message interaction shared by all brains.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from dataclasses import dataclass + +from services.client_tools import ClientToolError, ClientToolPort + + +BUILTIN_SHOW_MESSAGE = "show_message" +SpeechCompletion = Awaitable[None] | None +Speak = Callable[[str], Awaitable[SpeechCompletion]] +StartedHook = Callable[[], Awaitable[None]] + + +@dataclass(frozen=True) +class MessageDisplaySpec: + """Content rendered by the platform-provided client message dialog.""" + + title: str + message: str + confirm_label: str = "确认" + + +@dataclass(frozen=True) +class MessageStageSpec: + """Mode-independent fixed speech and optional client interaction.""" + + speech: str = "" + display: MessageDisplaySpec | None = None + require_confirmation: bool = False + + +@dataclass(frozen=True) +class MessageStageResult: + """Result used by Workflow routing and Prompt opening failure handling.""" + + succeeded: bool + speech: str = "" + action: str | None = None + error: str | None = None + + +class MessageStageRunner: + """Run one atomic user-visible message stage. + + Speech is queued before the client message is dispatched. A confirmation + stage completes when the user confirms, even if audio is still playing. A + speech-only stage completes at the real transport playback boundary. + """ + + def __init__(self, client_tools: ClientToolPort | None = None) -> None: + self._client_tools = client_tools + + def set_client_tools(self, client_tools: ClientToolPort | None) -> None: + self._client_tools = client_tools + + async def run( + self, + spec: MessageStageSpec, + *, + speak: Speak | None = None, + set_input_enabled: Callable[[bool], None] | None = None, + input_already_blocked: bool = False, + release_input_on_success: bool = True, + release_input_on_failure: bool = True, + on_started: StartedHook | None = None, + ) -> MessageStageResult: + input_setter = set_input_enabled + if input_setter is not None and not input_already_blocked: + input_setter(False) + + result: MessageStageResult | None = None + try: + if on_started is not None: + await on_started() + + speech = spec.speech.strip() + if spec.require_confirmation and spec.display is None: + result = MessageStageResult( + succeeded=False, + speech=speech, + error="等待用户确认时必须显示客户端消息", + ) + return result + if not speech and spec.display is None: + result = MessageStageResult( + succeeded=False, + error="Message 阶段至少需要播报或客户端消息", + ) + return result + playback_completion: SpeechCompletion = None + if speech and speak is not None: + playback_completion = await speak(speech) + + action: str | None = None + if spec.display is not None: + result = await self._show_message(spec, speech=speech) + if not result.succeeded: + return result + action = result.action + + # Confirmation is the gate. It deliberately does not wait for the + # audio completion future, so the user can continue immediately. + if ( + playback_completion is not None + and not spec.require_confirmation + ): + await playback_completion + + result = MessageStageResult( + succeeded=True, + speech=speech, + action=action, + ) + return result + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 - surface deterministic stage failure + result = MessageStageResult( + succeeded=False, + speech=spec.speech.strip(), + error=str(exc), + ) + return result + finally: + should_release = ( + result is not None + and ( + (result.succeeded and release_input_on_success) + or (not result.succeeded and release_input_on_failure) + ) + ) + if input_setter is not None and should_release: + input_setter(True) + + async def _show_message( + self, + spec: MessageStageSpec, + *, + speech: str, + ) -> MessageStageResult: + display = spec.display + if display is None: + return MessageStageResult(succeeded=True, speech=speech) + if self._client_tools is None: + return MessageStageResult( + succeeded=False, + speech=speech, + error="当前运行模式不支持客户端消息", + ) + try: + response = await self._client_tools.call( + BUILTIN_SHOW_MESSAGE, + { + "title": display.title, + "message": display.message, + "actions": [ + { + "id": "confirmed", + "label": display.confirm_label, + "style": "primary", + } + ], + "dismissible": not spec.require_confirmation, + }, + timeout_seconds=3, + wait_for_response=spec.require_confirmation, + response_wait_mode=( + "session" if spec.require_confirmation else "timeout" + ), + ) + except ClientToolError as exc: + return MessageStageResult( + succeeded=False, + speech=speech, + error=str(exc), + ) + if response.get("status") != "ok": + return MessageStageResult( + succeeded=False, + speech=speech, + error=str(response.get("message") or "客户端消息显示失败"), + ) + data = response.get("data") + action = ( + str(data.get("action") or "") or None + if isinstance(data, dict) + else None + ) + return MessageStageResult( + succeeded=True, + speech=speech, + action=action, + ) diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 72ee34c..51d0e1b 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -8,12 +8,12 @@ from typing import Any SPEC_VERSION = "3" -NODE_TYPES = {"start", "agent", "action", "handoff", "end"} +NODE_TYPES = {"start", "agent", "message", "action", "handoff", "end"} EDGE_MODES = {"llm", "expression", "always"} AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"} ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"} ACTION_USER_INPUT_POLICIES = {"queue", "block"} -AUTOMATIC_NODE_TYPES = {"start", "action", "handoff"} +AUTOMATIC_NODE_TYPES = {"start", "message", "action", "handoff"} EXPRESSION_OPERATORS = { "eq", "neq", @@ -31,7 +31,7 @@ NODE_SPECS: list[dict[str, Any]] = [ "name": "start", "displayName": "Start", "category": "control_node", - "description": "初始化会话、动态变量和全局观察器，可播放固定开场白。", + "description": "初始化会话、动态变量和全局观察器。", "icon": "Play", "accent": "mint", "addable": False, @@ -44,7 +44,6 @@ NODE_SPECS: list[dict[str, Any]] = [ }, "fields": [ {"key": "name", "label": "节点名称", "type": "text", "default": "Start"}, - {"key": "greeting", "label": "固定开场白", "type": "textarea", "default": ""}, ], }, { @@ -67,6 +66,19 @@ NODE_SPECS: list[dict[str, Any]] = [ }, ], }, + { + "name": "message", + "displayName": "Message", + "category": "interaction_node", + "description": "固定播报，并可同时显示内置客户端消息、等待用户确认。", + "icon": "MessageSquareText", + "accent": "lavender", + "addable": True, + "constraints": {"minIncoming": 1, "minOutgoing": 0}, + "fields": [ + {"key": "name", "label": "节点名称", "type": "text", "default": "Message"}, + ], + }, { "name": "action", "displayName": "Action", @@ -172,6 +184,17 @@ def _normalize_action_data(data: dict[str, Any]) -> None: ) data.setdefault("resultAssignments", {}) data.setdefault("userInputPolicy", "queue") + data.pop("speech", None) + + +def _normalize_message_data(data: dict[str, Any]) -> None: + """Fill the small built-in Message contract used by runtime and editor.""" + data.setdefault("speech", "") + data.setdefault("showMessage", False) + data.setdefault("title", "重要提示") + data.setdefault("message", "") + data.setdefault("confirmLabel", "确认") + data.setdefault("requireConfirmation", False) def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: @@ -200,8 +223,12 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: source.setdefault("edges", []) for node in source["nodes"]: data = node.setdefault("data", {}) - if node.get("type") == "agent": + if node.get("type") == "start": + data.pop("greeting", None) + elif node.get("type") == "agent": _normalize_agent_data(data) + elif node.get("type") == "message": + _normalize_message_data(data) elif node.get("type") == "action": _normalize_action_data(data) return source @@ -218,6 +245,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: "endCall": "end", "start": "start", "agent": "agent", + "message": "message", "action": "action", "handoff": "handoff", "end": "end", @@ -236,10 +264,13 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: data.setdefault("scope", "session") elif new_type == "agent": _normalize_agent_data(data) + elif new_type == "message": + _normalize_message_data(data) elif new_type == "action": _normalize_action_data(data) elif new_type == "start": prompt = str(data.pop("prompt", "") or "").strip() + data.pop("greeting", None) if prompt: start_prompt_nodes[str(node.get("id"))] = prompt for key in ("allowInterrupt", "addGlobalPrompt"): @@ -349,6 +380,49 @@ def validate_graph(graph: dict[str, Any]) -> list[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") + show_message = data.get("showMessage") + require_confirmation = data.get("requireConfirmation") + if not isinstance(speech, str): + errors.append(f"Message 节点 {node_id} 的播报内容必须是文本") + if not isinstance(show_message, bool): + errors.append(f"Message 节点 {node_id} 的弹窗开关必须是布尔值") + if not isinstance(require_confirmation, bool): + errors.append(f"Message 节点 {node_id} 的确认开关必须是布尔值") + if require_confirmation and show_message is not True: + errors.append(f"Message 节点 {node_id} 等待确认时必须显示弹窗") + if not str(speech or "").strip() and show_message is not True: + errors.append(f"Message 节点 {node_id} 至少需要播报或显示弹窗") + if show_message is True: + title = data.get("title") + message = data.get("message") + confirm_label = data.get("confirmLabel") + if ( + not isinstance(title, str) + or not title.strip() + or len(title) > 120 + ): + errors.append( + f"Message 节点 {node_id} 的弹窗标题必须为 1-120 个字符" + ) + if ( + not isinstance(message, str) + or not message.strip() + or len(message) > 2000 + ): + errors.append( + f"Message 节点 {node_id} 的弹窗消息必须为 1-2000 个字符" + ) + if ( + not isinstance(confirm_label, str) + or not confirm_label.strip() + or len(confirm_label) > 40 + ): + errors.append( + f"Message 节点 {node_id} 的按钮文字必须为 1-40 个字符" + ) elif node_type == "action": data = node.get("data") or {} assignment_mode = data.get("resultAssignmentMode") @@ -477,7 +551,7 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if node.get("type") != "agent" ) if any(visit(node_id) for node_id in automatic_node_ids): - errors.append("Start/Action/Handoff/End 之间不能形成无等待循环") + errors.append("自动节点之间不能形成无等待循环") return list(dict.fromkeys(errors)) diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index 99f66d7..e1a863f 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -2,6 +2,8 @@ from __future__ import annotations +import asyncio +from collections import deque from collections.abc import Awaitable, Callable from loguru import logger @@ -19,6 +21,7 @@ class CallEndCoordinator: self._speaking = False self._response_speech_started = False self._tracked_speeches = 0 + self._tracked_speech_completions: deque[asyncio.Future[None]] = deque() self._finish_after_tracked_speech = False self._finished = False self._reason = "completed" @@ -39,9 +42,12 @@ 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.""" + def track_speech(self) -> Awaitable[None]: + """Register fixed speech and return its transport completion signal.""" + completion = asyncio.get_running_loop().create_future() + self._tracked_speech_completions.append(completion) self._tracked_speeches += 1 + return completion async def arm_after_tracked_speech(self) -> None: """Finish after every already queued fixed utterance has played.""" @@ -73,6 +79,9 @@ class CallEndCoordinator: self._speaking = False if self._tracked_speeches > 0: self._tracked_speeches -= 1 + completion = self._tracked_speech_completions.popleft() + if not completion.done(): + completion.set_result(None) if ( self._finish_after_tracked_speech and self._tracked_speeches == 0 diff --git a/backend/services/workflow/agent.py b/backend/services/workflow/agent.py index 8ecf41b..1fdc01a 100644 --- a/backend/services/workflow/agent.py +++ b/backend/services/workflow/agent.py @@ -2,8 +2,6 @@ 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 @@ -108,7 +106,6 @@ class WorkflowAgentStage: 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) @@ -119,11 +116,6 @@ class WorkflowAgentStage: 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 @@ -133,7 +125,6 @@ class WorkflowAgentStage: "name": node_id, "role_message": self.role_message(node_id), "task_messages": [ - *greeting_messages, *(leading_messages or []), *fixed_reply_messages, ], diff --git a/backend/services/workflow/models.py b/backend/services/workflow/models.py index e8a0c17..53d9ba9 100644 --- a/backend/services/workflow/models.py +++ b/backend/services/workflow/models.py @@ -15,6 +15,7 @@ class WorkflowStatus(StrEnum): ROUTING = "routing" RUNNING_AGENT = "running_agent" RUNNING_ACTION = "running_action" + RUNNING_MESSAGE = "running_message" HANDOFF = "handoff" ENDED = "ended" diff --git a/backend/services/workflow/output.py b/backend/services/workflow/output.py index b3d5994..4b291f7 100644 --- a/backend/services/workflow/output.py +++ b/backend/services/workflow/output.py @@ -5,65 +5,15 @@ from __future__ import annotations from typing import Any from uuid import uuid4 -from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame +from pipecat.frames.frames import OutputTransportMessageUrgentFrame from pipecat.utils.time import time_now_iso8601 -from services.brains.base import BrainRuntime -from services.runtime_variables import DynamicVariableStore +from services.fixed_speech import FixedSpeechOutput -class WorkflowOutput: +class WorkflowOutput(FixedSpeechOutput): """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}) diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index fe33641..d9bb96c 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -190,20 +190,12 @@ class WorkflowEngine: sections.append(f"[当前阶段任务]\n{prompt}") return "\n\n".join(sections) - def greeting(self, store: DynamicVariableStore) -> str: - return store.render(str(self.data(self.start_id).get("greeting") or "")) - def routing_prompt(self, node_id: str, store: DynamicVariableStore) -> str: """Describe the current node to the small LLM edge router.""" if self.node_type(node_id) == "agent": return self.prompt_for(node_id, store) data = self.data(node_id) - details = ( - data.get("greeting") - or data.get("message") - or data.get("target") - or "" - ) + details = data.get("message") or data.get("target") or "" rendered = store.render(str(details)).strip() return f"{self.node_type(node_id) or 'workflow'} 节点：{rendered or self.name(node_id)}" diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 0745f42..5487d60 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -29,7 +30,7 @@ from services.brains.dify_llm import ( ) from services.brains.workflow_brain import WorkflowBrain from services.runtime_variables import prepare_dynamic_config -from services.action_runtime import ActionError, ActionOutcome, ActionStatus +from services.action_runtime import ActionOutcome, ActionStatus from services.workflow.models import ( LLMRouteResult, RouteStatus, @@ -350,30 +351,32 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): ["preflight_1", "preflight_2"], ) - async def test_opening_actions_wait_for_confirmation_and_greeting(self): - tools = [ - RuntimeTool( - id=f"opening_{index}", - name=f"开场动作 {index}", - function_name="show_message" if index == 1 else "load_opening_data", - type="client" if index == 1 else "http", - ) - for index in (1, 2) - ] + async def test_opening_stage_starts_speech_and_releases_on_confirmation(self): + tool = RuntimeTool( + id="opening_data", + name="加载开场数据", + function_name="load_opening_data", + type="http", + ) cfg = AssistantConfig( type="prompt", - tools=tools, + greeting="请阅读并确认重要信息", + tools=[tool], startup={ "execution_mode": "sequential", + "opening_message": { + "title": "重要提示", + "message": "请确认已阅读。", + "confirm_label": "确认", + }, "actions": [ { - "id": f"opening_{index}", + "id": "opening_data", "phase": "opening", - "tool_id": f"opening_{index}", + "tool_id": "opening_data", "arguments": {}, "required": True, } - for index in (1, 2) ], }, ) @@ -384,6 +387,17 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): async def queue_frame(frame): queued.append(frame) + confirmation_started = asyncio.Event() + user_confirmed = asyncio.Event() + client_calls = [] + + class FakeClientTools: + async def call(self, function_name, arguments, **options): + client_calls.append((function_name, arguments, options)) + confirmation_started.set() + await user_confirmed.wait() + return {"status": "ok", "data": {"action": "confirmed"}} + await brain.setup( cfg, BrainRuntime( @@ -393,28 +407,51 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): set_system_prompt=lambda _prompt: None, set_tools=lambda _tools: None, call_end=FakeCallEnd(), + client_tools=FakeClientTools(), set_input_enabled=input_states.append, ), ) - brain._actions.execute = AsyncMock( - side_effect=[ - ActionOutcome( - invocation_id=f"act_{index}", - status=ActionStatus.SUCCESS, - duration_ms=index, - ) - for index in (1, 2) - ] - ) + called_tool_ids = [] - await brain.on_connected(greeting_pending=True) - await brain.on_client_ready() + async def execute(tool, *_args, **_kwargs): + called_tool_ids.append(tool.id) + return ActionOutcome( + invocation_id=f"act_{len(called_tool_ids)}", + status=ActionStatus.SUCCESS, + duration_ms=len(called_tool_ids), + ) + + brain._actions.execute = AsyncMock(side_effect=execute) + + self.assertEqual(await brain.greeting(cfg), "") + await brain.on_connected(greeting_pending=False) + opening_task = asyncio.create_task(brain.on_client_ready()) + await confirmation_started.wait() self.assertEqual(input_states, [False]) - called_tool_ids = [ - call.args[0].id for call in brain._actions.execute.await_args_list - ] - self.assertEqual(called_tool_ids, ["opening_1", "opening_2"]) + self.assertEqual(called_tool_ids, []) + self.assertEqual(client_calls[0][0], "show_message") + self.assertFalse(client_calls[0][1]["dismissible"]) + self.assertTrue( + any( + isinstance(frame, TTSSpeakFrame) + and frame.text == "请阅读并确认重要信息" + for frame in queued + ) + ) + self.assertTrue( + any( + isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "transcript" + and frame.message.get("content") == "请阅读并确认重要信息" + for frame in queued + ) + ) + + user_confirmed.set() + await opening_task + self.assertEqual(input_states, [False, True]) + self.assertEqual(called_tool_ids, ["opening_data"]) self.assertEqual( len( [ @@ -424,36 +461,22 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): and frame.message.get("type") == "startup-action-result" ] ), - 2, + 1, ) - await brain.on_greeting_finished() - self.assertEqual(input_states, [False, True]) - # Replayed client-ready must not execute startup actions twice. await brain.on_client_ready() - self.assertEqual(brain._actions.execute.await_count, 2) + self.assertEqual(brain._actions.execute.await_count, 1) async def test_required_opening_failure_keeps_input_blocked_and_ends_call(self): - tool = RuntimeTool( - id="opening_message", - name="开场确认", - function_name="show_message", - type="client", - ) cfg = AssistantConfig( type="prompt", - tools=[tool], startup={ - "actions": [ - { - "id": "opening_message", - "phase": "opening", - "tool_id": tool.id, - "arguments": {}, - "required": True, - } - ] + "opening_message": { + "title": "重要提示", + "message": "请确认已阅读。", + "confirm_label": "确认", + } }, ) brain = build_brain(cfg) @@ -463,6 +486,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): async def queue_frame(_frame): pass + class FailingClientTools: + async def call(self, *_args, **_kwargs): + return {"status": "error", "message": "客户端未显示消息"} + await brain.setup( cfg, BrainRuntime( @@ -472,20 +499,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): set_system_prompt=lambda _prompt: None, set_tools=lambda _tools: None, call_end=call_end, + client_tools=FailingClientTools(), set_input_enabled=input_states.append, ), ) - brain._actions.execute = AsyncMock( - return_value=ActionOutcome( - invocation_id="act_failed", - status=ActionStatus.FAILURE, - duration_ms=1, - error=ActionError( - code="tool_error", - message="用户未确认", - ), - ) - ) await brain.on_connected(greeting_pending=False) await brain.on_client_ready() @@ -948,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_waits_for_start_greeting_to_finish(self): + async def test_initial_fixed_speech_starts_without_workflow_greeting(self): brain = WorkflowBrain( { "specVersion": 3, @@ -1006,12 +1023,14 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) brain._manager = FakeManager() - await brain.on_connected(greeting_pending=True) - - self.assertEqual(brain._manager.current_node, "start") - self.assertFalse(any(isinstance(frame, TTSSpeakFrame) for frame in queued)) - - await brain.on_greeting_finished() + self.assertNotIn("greeting", brain._engine.data("start")) + self.assertEqual( + await brain.greeting( + AssistantConfig(type="workflow", greeting="旧助手级开场白") + ), + "", + ) + await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") fixed_speech_frames = [ @@ -1020,8 +1039,8 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(fixed_speech_frames), 1) self.assertEqual(fixed_speech_frames[0].text, "请问您怎么称呼？") - # Playback notifications may be duplicated by a transport reconnect; - # the initial entry behavior must still run only once. + # Workflow no longer owns a greeting playback lifecycle. Stray generic + # transport notifications must not repeat Agent entry behavior. await brain.on_greeting_finished() self.assertEqual( len([frame for frame in queued if isinstance(frame, TTSSpeakFrame)]), @@ -1372,6 +1391,148 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): await brain._enter_action("queue_action") self.assertEqual(input_states, ["executing"]) + async def test_message_starts_speech_and_releases_on_confirmation(self): + brain = WorkflowBrain( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "message", + "type": "message", + "data": { + "speech": "请先确认 {{customer}} 的重要信息。", + "showMessage": True, + "title": "重要提示", + "message": "请核对客户信息。", + "confirmLabel": "确认", + "requireConfirmation": True, + }, + }, + ], + "edges": [], + }, + ) + ) + brain._store.values["customer"] = "王先生" + events = [] + + async def queue_frame(frame): + if isinstance(frame, TTSSpeakFrame): + events.append(("speech", frame.text)) + + class OrderedCallEnd(FakeCallEnd): + def __init__(self): + super().__init__() + self.playback_completion = None + + def track_speech(self): + self.tracked_speeches += 1 + events.append("tracked") + self.playback_completion = asyncio.get_running_loop().create_future() + return self.playback_completion + + message_started = asyncio.Event() + user_confirmed = asyncio.Event() + + class FakeClientTools: + async def call(self, function_name, arguments, **options): + self.function_name = function_name + self.arguments = arguments + self.options = options + events.append("message_displayed") + message_started.set() + await user_confirmed.wait() + return {"status": "ok", "data": {"action": "confirmed"}} + + input_states = [] + call_end = OrderedCallEnd() + client_tools = FakeClientTools() + 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, + client_tools=client_tools, + set_input_enabled=input_states.append, + ) + brain._message_stages.set_client_tools(client_tools) + + message_task = asyncio.create_task(brain._enter_message("message")) + await message_started.wait() + + self.assertEqual( + events, + [ + "tracked", + ("speech", "请先确认 王先生 的重要信息。"), + "message_displayed", + ], + ) + self.assertEqual(input_states, [False]) + self.assertEqual(client_tools.function_name, "show_message") + self.assertEqual(client_tools.options["response_wait_mode"], "session") + + user_confirmed.set() + result = await message_task + self.assertTrue(result.succeeded) + self.assertEqual(result.action, "confirmed") + self.assertFalse(call_end.playback_completion.done()) + self.assertEqual(input_states, [False, True]) + + async def test_speech_only_message_waits_for_transport_playback(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "message", + "type": "message", + "data": {"speech": "正在为您准备服务。"}, + }, + ], + "edges": [], + } + ) + + class PlaybackCallEnd(FakeCallEnd): + def __init__(self): + super().__init__() + self.completion = None + + def track_speech(self): + self.completion = asyncio.get_running_loop().create_future() + return self.completion + + call_end = PlaybackCallEnd() + input_states = [] + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=noop_queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + set_input_enabled=input_states.append, + ) + + message_task = asyncio.create_task(brain._enter_message("message")) + await asyncio.sleep(0) + self.assertFalse(message_task.done()) + self.assertEqual(input_states, [False]) + + call_end.completion.set_result(None) + result = await message_task + self.assertTrue(result.succeeded) + self.assertEqual(input_states, [False, True]) + async def test_nodes_without_outgoing_edges_remain_active(self): queued = [] @@ -1501,7 +1662,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): { "id": "start", "type": "start", - "data": {"name": "Start", "greeting": "你想做什么？"}, + "data": {"name": "Start"}, }, { "id": "eat", @@ -1815,10 +1976,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): { "id": "start", "type": "start", - "data": { - "name": "Start", - "greeting": "欢迎，{{user_name}}", - }, + "data": {"name": "Start"}, }, { "id": "agent", @@ -1925,13 +2083,8 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) await brain.setup(cfg, runtime) greeting = await brain.greeting(cfg) - self.assertEqual(greeting, "欢迎，王先生") - greeting_message = { - "role": "system", - "content": f"{GREETING_CONTEXT_MARKER}\n欢迎，王先生", - } - brain.prepare_greeting_context(greeting, context) - self.assertEqual(context.get_messages(), [greeting_message]) + self.assertEqual(greeting, "") + self.assertEqual(context.get_messages(), []) await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") variable_events = [ @@ -1980,7 +2133,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): agent_config = brain._agent_config("agent") self.assertIn("王先生", agent_config["role_message"]) self.assertIn("工作流路由已在用户一轮输入结束时完成", agent_config["role_message"]) - self.assertEqual(agent_config["task_messages"], [greeting_message]) + self.assertEqual(agent_config["task_messages"], []) self.assertFalse(agent_config["respond_immediately"]) self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) self.assertEqual( @@ -2005,10 +2158,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertNotIn("pre_actions", fixed_config) self.assertEqual( fixed_config["task_messages"], - [ - greeting_message, - {"role": "assistant", "content": "您好，王先生"}, - ], + [{"role": "assistant", "content": "您好，王先生"}], ) self.assertEqual( brain._agent_config( @@ -2016,7 +2166,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): [{"role": "assistant", "content": "正在进入下一阶段"}], )["task_messages"], [ - greeting_message, {"role": "assistant", "content": "正在进入下一阶段"}, {"role": "assistant", "content": "您好，王先生"}, ], @@ -2034,10 +2183,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ] self.assertEqual( context_updates[-1].messages, - [ - greeting_message, - {"role": "assistant", "content": "您好，王先生"}, - ], + [{"role": "assistant", "content": "您好，王先生"}], ) self.assertFalse( any( diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py index 683fd3c..8a9ebc9 100644 --- a/backend/tests/test_call_lifecycle.py +++ b/backend/tests/test_call_lifecycle.py @@ -45,17 +45,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() + first_completion = self.coordinator.track_speech() + second_completion = 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.assertTrue(first_completion.done()) + self.assertFalse(second_completion.done()) self.assertEqual(self.reasons, []) await self.coordinator.observe(BotStartedSpeakingFrame()) await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.assertTrue(second_completion.done()) self.assertEqual(self.reasons, ["workflow_completed"]) diff --git a/backend/tests/test_runtime_variables.py b/backend/tests/test_runtime_variables.py index f6ed90c..4453efa 100644 --- a/backend/tests/test_runtime_variables.py +++ b/backend/tests/test_runtime_variables.py @@ -89,7 +89,7 @@ class DynamicVariableTests(unittest.TestCase): { "id": "start", "type": "start", - "data": {"greeting": "您好 {{nickname}}"}, + "data": {"name": "Start"}, } ], "edges": [], diff --git a/backend/tests/test_startup_actions.py b/backend/tests/test_startup_actions.py index c329bdc..755f039 100644 --- a/backend/tests/test_startup_actions.py +++ b/backend/tests/test_startup_actions.py @@ -77,6 +77,39 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): startup=startup_body().startup, ) + def test_realtime_rejects_builtin_opening_message(self): + with self.assertRaisesRegex(ValueError, "Realtime"): + AssistantUpsert( + name="Realtime 开场消息", + type="prompt", + runtimeMode="realtime", + startup={ + "openingMessage": { + "title": "重要提示", + "message": "请确认已阅读。", + "confirmLabel": "确认", + } + }, + ) + + async def test_builtin_opening_message_does_not_reference_a_tool(self): + body = AssistantUpsert( + name="内置开场消息", + type="prompt", + startup={ + "openingMessage": { + "title": "重要提示", + "message": "请确认已阅读。", + "confirmLabel": "确认", + } + }, + ) + + await _validate_startup_actions(FakeSession(None), body) + + self.assertEqual(body.startup.actions, []) + self.assertEqual(body.startup.opening_message.confirm_label, "确认") + async def test_startup_tool_does_not_need_conversation_binding(self): tool = SimpleNamespace( id="tool_message", diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 70936c3..1f56711 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -55,6 +55,20 @@ def valid_graph(): class WorkflowGraphTests(unittest.TestCase): + def test_workflow_removes_assistant_and_start_greetings(self): + body = AssistantUpsert( + name="无开场白工作流", + type="workflow", + greeting="旧助手级开场白", + graph=valid_graph(), + ) + body.graph["nodes"][0]["data"]["greeting"] = "旧 Start 开场白" + + normalized = normalize_graph(body.graph) + + self.assertEqual(body.greeting, "") + self.assertNotIn("greeting", normalized["nodes"][0]["data"]) + def test_revision_pins_the_normalized_workflow_snapshot(self): first = WorkflowEngine(valid_graph()) same_graph = deepcopy(valid_graph()) @@ -147,6 +161,7 @@ class WorkflowGraphTests(unittest.TestCase): ) self.assertEqual(actions["legacy_none"]["resultAssignments"], {}) self.assertEqual(actions["legacy_none"]["userInputPolicy"], "queue") + self.assertNotIn("speech", actions["legacy_none"]) def test_action_advanced_settings_are_validated(self): graph = valid_graph() @@ -167,6 +182,46 @@ class WorkflowGraphTests(unittest.TestCase): self.assertTrue(any("结果变量映射必须是对象" in error for error in errors)) self.assertTrue(any("用户输入策略无效" in error for error in errors)) + def test_message_defaults_and_validation(self): + graph = valid_graph() + graph["nodes"].append( + { + "id": "message", + "type": "message", + "data": {"speech": "请确认重要信息"}, + } + ) + normalized = normalize_graph(graph) + message = next( + node for node in normalized["nodes"] if node["type"] == "message" + ) + self.assertFalse(message["data"]["showMessage"]) + self.assertFalse(message["data"]["requireConfirmation"]) + self.assertEqual(message["data"]["confirmLabel"], "确认") + + message["data"].update( + { + "speech": "", + "showMessage": True, + "title": "重要提示", + "message": "", + "confirmLabel": "确认", + "requireConfirmation": True, + } + ) + errors = validate_graph(normalized) + self.assertTrue(any("弹窗消息必须为" in error for error in errors)) + + message["data"].update( + { + "speech": "请确认", + "showMessage": False, + "requireConfirmation": True, + } + ) + errors = validate_graph(normalized) + self.assertTrue(any("等待确认时必须显示弹窗" in error for error in errors)) + def test_voice_resource_creates_isolated_runtime_config(self): base = AssistantConfig(type="workflow", asr="default", voice="default") asr = RuntimeModelResource( diff --git a/frontend/src/components/assistant-editor/prompt-editor.tsx b/frontend/src/components/assistant-editor/prompt-editor.tsx index 40887df..bd1fa8a 100644 --- a/frontend/src/components/assistant-editor/prompt-editor.tsx +++ b/frontend/src/components/assistant-editor/prompt-editor.tsx @@ -85,55 +85,18 @@ export function PromptEditor({ handlePromptVisionEnabledChange, handlePromptModelChange, }: PromptEditorProps) { - const openingMessage = form.startup.actions.find( - (action) => action.id === "opening_message" && action.phase === "opening", - ); - const openingArguments = openingMessage?.arguments ?? {}; - const openingButtons = Array.isArray(openingArguments.actions) - ? openingArguments.actions - : []; - const openingButton = - openingButtons[0] && typeof openingButtons[0] === "object" - ? (openingButtons[0] as Record) - : {}; - const showMessageTools = tools - .filter( - (tool) => - tool.status === "active" && - tool.type === "client" && - tool.functionName === "show_message" && - tool.definition.type === "client" && - tool.definition.config.waitForResponse && - tool.definition.config.responseWaitMode === "session", - ) - .map((tool) => ({ value: tool.id, label: tool.name })); + const openingMessage = form.startup.openingMessage; function setOpeningMessage(enabled: boolean) { - const otherActions = form.startup.actions.filter( - (action) => action.id !== "opening_message", - ); - const defaultToolId = showMessageTools[0]?.value ?? ""; updateForm("startup", { - executionMode: "sequential", - actions: enabled - ? [ - ...otherActions, - { - id: "opening_message", - phase: "opening", - toolId: defaultToolId, - required: true, - arguments: { - title: "重要提示", - message: "请确认已阅读以上信息。", - actions: [ - { id: "confirmed", label: "确认", style: "primary" }, - ], - dismissible: false, - }, - }, - ] - : otherActions, + ...form.startup, + openingMessage: enabled + ? { + title: "重要提示", + message: "请确认已阅读以上信息。", + confirmLabel: "确认", + } + : null, }); } @@ -143,16 +106,10 @@ export function PromptEditor({ if (!openingMessage) return; updateForm("startup", { ...form.startup, - actions: form.startup.actions.map((action) => - action.id === openingMessage.id ? { ...action, ...patch } : action, - ), + openingMessage: { ...openingMessage, ...patch }, }); } - function updateOpeningArguments(patch: Record) { - updateOpeningMessage({ arguments: { ...openingArguments, ...patch } }); - } - return (
@@ -235,33 +192,21 @@ export function PromptEditor({
{openingMessage && (
- updateOpeningMessage({ toolId })} - /> - {showMessageTools.length === 0 && ( -

- 请先在组件管理中创建 functionName 为 show_message、会话内等待的 Client Tool。 -

- )} updateOpeningArguments({ message })} + value={openingMessage.message} + onChange={(message) => updateOpeningMessage({ message })} placeholder="请输入需要用户确认的重要信息" rows={4} /> @@ -279,16 +224,10 @@ export function PromptEditor({ 确认按钮 - updateOpeningArguments({ - actions: [ - { - id: "confirmed", - label: event.target.value, - style: "primary", - }, - ], + updateOpeningMessage({ + confirmLabel: event.target.value, }) } placeholder="确认" @@ -296,8 +235,8 @@ export function PromptEditor({ />

- 弹窗不可跳过；失败时将结束本次通话。该工具仅用于开场， - 如需允许模型后续调用，请另外在“工具”区域勾选。 + 弹窗不可跳过；失败时将结束本次通话。这是平台内置开场阶段， + 不会作为工具暴露给模型。

)} @@ -318,10 +257,14 @@ export function PromptEditor({ value={form.runtimeMode} onChange={(runtimeMode) => { updateForm("runtimeMode", runtimeMode); - if (runtimeMode === "realtime" && form.startup.actions.length) { + if ( + runtimeMode === "realtime" && + (form.startup.actions.length || form.startup.openingMessage) + ) { updateForm("startup", { executionMode: "sequential", actions: [], + openingMessage: null, }); } }} diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 5600262..ce9e518 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -177,7 +177,11 @@ function blankPromptForm(name: string): AssistantForm { knowledgeRetrievalConfig: defaultKnowledgeRetrievalConfig(), enableInterrupt: true, turnConfig: defaultTurnConfig(), - startup: { executionMode: "sequential", actions: [] }, + startup: { + executionMode: "sequential", + actions: [], + openingMessage: null, + }, visionEnabled: false, visionModelResourceId: "", toolIds: [], @@ -466,7 +470,11 @@ export function AssistantPage(props: AssistantPageProps) { a.knowledgeRetrievalConfig ?? defaultKnowledgeRetrievalConfig(), enableInterrupt: a.enableInterrupt, turnConfig: normalizeTurnConfig(a.turnConfig), - startup: a.startup ?? { executionMode: "sequential", actions: [] }, + startup: { + executionMode: "sequential", + actions: a.startup?.actions ?? [], + openingMessage: a.startup?.openingMessage ?? null, + }, visionEnabled: a.visionEnabled, visionModelResourceId: a.visionModelResourceId ?? "", toolIds: a.toolIds ?? [], @@ -536,7 +544,11 @@ export function AssistantPage(props: AssistantPageProps) { greeting: "", enableInterrupt: true, turnConfig: defaultTurnConfig(), - startup: { executionMode: "sequential", actions: [] }, + startup: { + executionMode: "sequential", + actions: [], + openingMessage: null, + }, visionEnabled: false, visionModelResourceId: null, modelResourceIds: {}, diff --git a/frontend/src/components/workflow/GenericNode.tsx b/frontend/src/components/workflow/GenericNode.tsx index 304873c..7bca604 100644 --- a/frontend/src/components/workflow/GenericNode.tsx +++ b/frontend/src/components/workflow/GenericNode.tsx @@ -27,7 +27,12 @@ export function GenericNode({ id, type, data, selected }: NodeProps) { const nodeData = data as WorkflowNodeData; const Icon = spec.icon; - const preview = (nodeData.greeting || nodeData.prompt || nodeData.message || "") + const preview = ( + nodeData.prompt || + nodeData.speech || + nodeData.message || + "" + ) .toString() .trim(); const entryModeLabel = { @@ -50,8 +55,17 @@ export function GenericNode({ id, type, data, selected }: NodeProps) { nodeData.visionEnabled ? "视觉理解" : null, ].filter(Boolean) : type === "action" && nodeData.toolId - ? ["确定性工具"] - : []; + ? [ + "确定性工具", + nodeData.userInputPolicy === "block" ? "禁止输入" : null, + ].filter(Boolean) + : type === "message" + ? [ + nodeData.speech ? "固定播报" : null, + nodeData.showMessage ? "客户端消息" : null, + nodeData.requireConfirmation ? "等待确认" : null, + ].filter(Boolean) + : []; return (
void; + setPatch: (patch: Partial) => void; +}) { + const showMessage = Boolean(draft.showMessage); + const requireConfirmation = Boolean(draft.requireConfirmation); + + return ( + } + title="播报与确认" + description="播放固定话术，并可同时显示平台内置的客户端消息" + > +