diff --git a/backend/db/seed_assistants.sql b/backend/db/seed_assistants.sql index 8ae9c5b..4893e67 100644 --- a/backend/db/seed_assistants.sql +++ b/backend/db/seed_assistants.sql @@ -9,24 +9,24 @@ ON CONFLICT (id) DO NOTHING; INSERT INTO assistants ( id, name, type, runtime_mode, greeting, enable_interrupt, vision_enabled, vision_model_resource_id, - knowledge_base_id, prompt, api_url, api_key, app_id, graph + knowledge_base_id, prompt, api_url, api_key, app_id, graph, startup ) VALUES ('asst_001', '政务咨询助手', 'prompt', 'pipeline', '您好,我是政务助手,请问有什么可以帮您?', TRUE, FALSE, NULL, - 'kb_001', '你是一名专业的政务咨询助手,回答准确、简洁,不编造政策内容。', '', '', '', '{}'), - ('asst_002', '热线工单助手', 'workflow', 'pipeline', '', TRUE, - FALSE, NULL, - NULL, '', '', '', '', - '{"nodes":[{"id":"1","type":"startCall","position":{"x":0,"y":0},"data":{"name":"开场","prompt":"你好,请问需要办理什么业务?"}}],"edges":[]}'), + 'kb_001', '你是一名专业的政务咨询助手,回答准确、简洁,不编造政策内容。', '', '', '', '{}', + '{"execution_mode":"sequential","opening_mode":"interruptible","entry_mode":"wait_user","actions":[],"opening_message":null}'), ('asst_003', 'Dify 客服助手', 'dify', 'pipeline', '', TRUE, FALSE, NULL, - NULL, '', 'https://api.dify.ai/v1', 'app-dify-demo-key', '', '{}'), + NULL, '', 'https://api.dify.ai/v1', 'app-dify-demo-key', '', '{}', + '{"execution_mode":"sequential","opening_mode":"interruptible","entry_mode":"wait_user","actions":[],"opening_message":null}'), ('asst_004', 'FastGPT 售后助手', 'fastgpt', 'pipeline', '', TRUE, FALSE, NULL, - NULL, '', 'https://api.fastgpt.in/api/v1/chat/completions', 'fastgpt-demo-key', 'app-fastgpt-001', '{}'), + NULL, '', 'https://api.fastgpt.in/api/v1/chat/completions', 'fastgpt-demo-key', 'app-fastgpt-001', '{}', + '{"execution_mode":"sequential","opening_mode":"interruptible","entry_mode":"wait_user","actions":[],"opening_message":null}'), ('asst_005', 'OpenCode 代码助手', 'opencode', 'pipeline', '', TRUE, FALSE, NULL, - NULL, '你是一个代码助手的语音界面,用简洁口语回答工程问题。', 'http://localhost:4096', 'opencode-demo-key', '', '{}') + NULL, '你是一个代码助手的语音界面,用简洁口语回答工程问题。', 'http://localhost:4096', 'opencode-demo-key', '', '{}', + '{"execution_mode":"sequential","opening_mode":"interruptible","entry_mode":"wait_user","actions":[],"opening_message":null}') ON CONFLICT (id) DO NOTHING; INSERT INTO assistant_model_bindings @@ -35,8 +35,6 @@ VALUES ('asst_001', 'LLM', 'model_001', '{}'), ('asst_001', 'ASR', 'model_002', '{}'), ('asst_001', 'TTS', 'model_004', '{}'), - ('asst_002', 'ASR', 'model_002', '{}'), - ('asst_002', 'TTS', 'model_004', '{}'), ('asst_003', 'ASR', 'model_002', '{}'), ('asst_003', 'TTS', 'model_004', '{}'), ('asst_003', 'Agent', 'model_014', '{}'), diff --git a/backend/models.py b/backend/models.py index 9eb22ff..7b8d572 100644 --- a/backend/models.py +++ b/backend/models.py @@ -122,7 +122,15 @@ class AssistantConfig(BaseModel): enableInterrupt: bool = True turnConfig: dict = Field(default_factory=dict) - startup: dict = Field(default_factory=dict) + startup: dict = Field( + default_factory=lambda: { + "execution_mode": "sequential", + "opening_mode": "interruptible", + "entry_mode": "wait_user", + "actions": [], + "opening_message": None, + } + ) analysis_config: dict = Field(default_factory=dict) built_in_tools: dict = Field(default_factory=dict) diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index ae803bf..6776dd6 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -20,7 +20,7 @@ from schemas import ( ) from services.auth import require_admin from services.masking import mask, resolve_incoming_key -from services.node_specs import graph_references, normalize_graph, validate_graph +from services.node_specs import copy_graph, graph_references, validate_graph from services.state_expressions import StateExpressionError, validate_update_rules from services.system_tools import RESERVED_TOOL_NAMES from services.tool_policy import policy_for_tool @@ -98,9 +98,9 @@ def _validate_workflow(body: AssistantUpsert) -> None: if body.graph and str(body.graph.get("specVersion") or "") != "4": raise HTTPException( 400, - "Workflow API 仅接受 specVersion=4;旧图请先执行数据库迁移", + "Workflow API 仅接受当前 specVersion=4 图结构", ) - body.graph = normalize_graph(body.graph or {}) + body.graph = copy_graph(body.graph or {}) errors = validate_graph(body.graph) settings = body.graph.get("settings") or {} graph_runtime_mode = str(settings.get("runtimeMode") or "pipeline") @@ -636,7 +636,7 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut: api_key=mask(assistant.api_key), app_id=assistant.app_id, graph=( - normalize_graph(assistant.graph or {}) + copy_graph(assistant.graph or {}) if assistant.type == "workflow" else {} ), diff --git a/backend/routes/voice_webrtc.py b/backend/routes/voice_webrtc.py index 37fd55b..0582c37 100644 --- a/backend/routes/voice_webrtc.py +++ b/backend/routes/voice_webrtc.py @@ -63,24 +63,23 @@ async def _wait_for_pipeline_close( async def shutdown_active_sessions() -> None: - """Compatibility proxy; now drains RTVI and OpenAI Realtime sessions.""" + """Drain RTVI and OpenAI Realtime sessions.""" _http_peers.clear() await realtime_lifecycle.shutdown_active_sessions( timeout=PIPELINE_CLOSE_GRACE_SECONDS, ) -def _decode_client_tool_names(encoded: str | None) -> list[str] | None: +def _decode_client_tool_names(encoded: str | None) -> list[str]: """Decode client capabilities carried outside the WebRTC SDP body. Pipecat's browser transport replaces the body of a ``Request`` endpoint with its SDP offer and drops ``APIRequest.requestData``. Request headers are - preserved, so they are the authoritative metadata channel for this path. - ``None`` means the header was absent and callers should use the legacy body. + preserved, so they are the single metadata channel for this path. """ if encoded is None: - return None + return [] try: padding = "=" * (-len(encoded) % 4) value = json.loads( @@ -134,21 +133,16 @@ async def http_offer(request: Request, payload: dict = Body(...)): header_variables = json.loads( base64.urlsafe_b64decode(encoded_variables + padding).decode("utf-8") ) - header_client_tools = _decode_client_tool_names( + client_tool_names = _decode_client_tool_names( request.headers.get("x-pipecat-client-tools") ) - client_tool_names = ( - header_client_tools - if header_client_tools is not None - else request_data.get("client_tool_names") or [] - ) logger.debug(f"WebRTC 客户端工具上报: {client_tool_names or ''}") offer_payload = { **payload, "assistant_id": request.headers.get("x-pipecat-assistant-id"), "dynamic_variables": header_variables, **request_data, - # Keep header metadata authoritative when both paths are present. + # Client capabilities have one canonical transport: the request header. "client_tool_names": client_tool_names, } answer = await _handle_offer_payload(offer_payload, _http_peers) diff --git a/backend/schemas.py b/backend/schemas.py index e7c05e4..982c906 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -9,7 +9,7 @@ from __future__ import annotations from datetime import UTC, datetime import re -from typing import Annotated, Any, ClassVar, Literal, Union +from typing import Annotated, Any, Literal, Union from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from pydantic.alias_generators import to_camel @@ -197,8 +197,8 @@ class OpeningMessageConfig(CamelModel): class StartupConfig(CamelModel): execution_mode: Literal["sequential"] = "sequential" - opening_mode: PromptOpeningMode | None = None - entry_mode: PromptEntryMode | None = None + opening_mode: PromptOpeningMode = "interruptible" + entry_mode: PromptEntryMode = "wait_user" actions: list[StartupAction] = Field(default_factory=list, max_length=5) opening_message: OpeningMessageConfig | None = None @@ -207,33 +207,15 @@ class StartupConfig(CamelModel): ids = [action.id for action in self.actions] if len(ids) != len(set(ids)): raise ValueError("启动 Action 的 id 不能重复") - legacy_confirmation = ( - self.opening_mode is None and self.opening_message is not None - ) - if self.opening_mode is None: - self.opening_mode = ( - "confirmation" if legacy_confirmation else "interruptible" - ) if self.opening_mode == "confirmation" and self.opening_message is None: raise ValueError("确认后继续必须配置开场确认弹窗") if self.opening_mode != "confirmation": self.opening_message = None - if self.entry_mode is None: - # Preserve the behavior introduced before entryMode became an - # explicit setting. Newly saved clients always send the field. - self.entry_mode = ( - "generate" if legacy_confirmation else "wait_user" - ) return self # ---------- 助手(单表 STI:瘦类型真列 + workflow 图 JSON 列) ---------- class AssistantUpsert(CamelModel): - # AssistantOut inherits this DTO for a stable public shape, but persisted - # historical rows must remain readable. Only request models enforce - # runtime combinations that the current editor/runtime can create. - enforce_request_constraints: ClassVar[bool] = True - name: str type: AssistantType = "prompt" runtime_mode: RuntimeMode = "pipeline" @@ -299,23 +281,20 @@ class AssistantUpsert(CamelModel): # bindings remain a Prompt-only public API. self.tool_bindings = [] self.built_in_tools = BuiltInTools() - if self.enforce_request_constraints: - if self.runtime_mode == "realtime" and ( - self.startup.actions - or self.startup.opening_message is not None - or self.startup.opening_mode != "interruptible" - or self.startup.entry_mode != "wait_user" - ): - raise ValueError("Prompt Realtime 模式暂不支持启动阶段") - # 外部托管大脑只能 cascade,拦住不兼容的 realtime + if self.runtime_mode == "realtime" and ( + self.startup.actions + or self.startup.opening_message is not None + or self.startup.opening_mode != "interruptible" + or self.startup.entry_mode != "wait_user" + ): + raise ValueError("Prompt Realtime 模式暂不支持启动阶段") + # 外部托管大脑只能使用 Pipeline,拦住不支持的 Realtime。 if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES: raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式") return self class AssistantOut(AssistantUpsert): - enforce_request_constraints: ClassVar[bool] = False - id: str updated_at: str | None = None diff --git a/backend/services/brains/fastgpt_brain.py b/backend/services/brains/fastgpt_brain.py index deb8882..3d0e416 100644 --- a/backend/services/brains/fastgpt_brain.py +++ b/backend/services/brains/fastgpt_brain.py @@ -1,6 +1,6 @@ """FastGPT 大脑:外部托管,context/KB/tools 全在 FastGPT 服务端。 -cascade-only(realtime 不兼容外部大脑)。每通电话持有一个稳定 chatId: +Pipeline-only(外部大脑不支持 Realtime)。每通电话持有一个稳定 chatId: greeting(get_chat_init)与后续每轮推理共用它,保证服务端上下文连续。 """ diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index 35c6c1b..bd48ed6 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -364,31 +364,14 @@ class PromptBrain(BaseBrain): ) 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")) + value = self._cfg.startup["opening_message"] return value if isinstance(value, dict) else None def _opening_mode(self) -> str: - startup = self._cfg.startup if isinstance(self._cfg.startup, dict) else {} - value = str(startup.get("opening_mode", startup.get("openingMode")) or "") - if value in {"interruptible", "playback", "confirmation"}: - return value - # Before openingMode existed, an opening message meant confirmation; - # ordinary Prompt greetings were interruptible. - return ( - "confirmation" - if self._opening_message() is not None - else "interruptible" - ) + return str(self._cfg.startup["opening_mode"]) def _entry_mode(self) -> str: - startup = self._cfg.startup if isinstance(self._cfg.startup, dict) else {} - value = str(startup.get("entry_mode", startup.get("entryMode")) or "") - if value in {"wait_user", "generate"}: - return value - # Raw runtime configs saved before entryMode existed generated a first - # reply after their opening confirmation. Keep that legacy behavior. - return "generate" if self._opening_mode() == "confirmation" else "wait_user" + return str(self._cfg.startup["entry_mode"]) async def _enter_prompt_if_ready(self) -> None: if ( diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 8f7631d..41597cd 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -1,4 +1,4 @@ -"""Pipecat Flows-backed Workflow v3 brain.""" +"""Pipecat Flows-backed Workflow v4 brain.""" from __future__ import annotations @@ -48,10 +48,8 @@ from services.action_stage import ActionStageRunner, ActionStageSpec, StageActio from services.fixed_speech import fixed_speech_context_message from services.knowledge import search as search_knowledge from services.message_policy import ( - MESSAGE_COMPLETION_POLICIES, MESSAGE_CONFIRMATION, MESSAGE_INTERRUPTIBLE, - MESSAGE_PLAYBACK, ) from services.message_stage import ( MessageDisplaySpec, @@ -160,12 +158,6 @@ class WorkflowBrain(BaseBrain): def __init__(self, cfg_or_graph: AssistantConfig | dict[str, Any]): cfg = cfg_or_graph if isinstance(cfg_or_graph, AssistantConfig) else None graph = deepcopy(cfg.graph if cfg is not None else cfg_or_graph) - if cfg is not None: - # Graph v3 owns Workflow defaults. Keep older saved graphs compatible - # by filling the new interaction settings from the assistant row. - settings = graph.setdefault("settings", {}) - settings.setdefault("enableInterrupt", cfg.enableInterrupt) - settings.setdefault("turnConfig", deepcopy(cfg.turnConfig)) self._engine = WorkflowEngine(graph or {}) if not self._engine.has_graph() or not self._engine.start_id: raise ValueError("WorkflowBrain 缺少有效的 Start 节点") @@ -722,7 +714,7 @@ class WorkflowBrain(BaseBrain): def _agent_runs_on_entry(self, node_id: str) -> bool: """Only explicit generate mode may start a reply on node entry.""" entry_mode = str( - self._engine.data(node_id).get("entryMode") or "wait_user" + self._engine.data(node_id).get("entryMode", "wait_user") ) return entry_mode == "generate" @@ -1604,21 +1596,17 @@ class WorkflowBrain(BaseBrain): return result def _message_completion_policy(self, node_id: str) -> str: - value = str( - self._engine.data(node_id).get("completionPolicy") - or MESSAGE_PLAYBACK - ) - return value if value in MESSAGE_COMPLETION_POLICIES else MESSAGE_PLAYBACK + return str(self._engine.data(node_id)["completionPolicy"]) def _set_last_action(self, outcome: ActionOutcome) -> None: - legacy_status = { + status_value = { ActionStatus.SUCCESS: "ok", ActionStatus.FAILURE: "error", ActionStatus.CANCELLED: "cancelled", }[outcome.status] self._store.values.update( { - "system__last_action_status": legacy_status, + "system__last_action_status": status_value, "system__last_action_invocation_id": outcome.invocation_id, "system__last_action_duration_ms": outcome.duration_ms, "system__last_action_error_code": ( diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index a77f0b8..219b2c9 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -21,7 +21,7 @@ from models import ( RuntimeTool, RuntimeToolBinding, ) -from services.node_specs import graph_references, normalize_graph +from services.node_specs import copy_graph, graph_references from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -228,7 +228,7 @@ async def resolve_runtime_config( else None ) - graph = normalize_graph(assistant.graph or {}) if assistant.type == "workflow" else {} + graph = copy_graph(assistant.graph or {}) if assistant.type == "workflow" else {} refs = graph_references(graph) if graph else { "model_resources": set(), "knowledge_bases": set(), diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 87df107..a734f96 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -10,7 +10,6 @@ from models import VISION_PROMPT_MODES from services.message_policy import ( MESSAGE_COMPLETION_POLICIES, MESSAGE_CONFIRMATION, - MESSAGE_PLAYBACK, ) @@ -170,236 +169,9 @@ def node_types_response() -> dict[str, Any]: return {"specVersion": SPEC_VERSION, "nodeTypes": NODE_SPECS} -def _edge_data_v3(edge: dict) -> dict: - data = deepcopy(edge.get("data") or {}) - data.pop("transitionSpeech", None) - data.pop("transition_speech", None) - if data.get("mode") in EDGE_MODES: - data.setdefault("priority", 10) - return data - condition = str(data.pop("condition", "") or "").strip() - data.update( - { - "mode": "llm" if condition else "always", - "priority": 10, - "condition": condition, - } - ) - return data - - -def _normalize_agent_data(data: dict[str, Any]) -> None: - """Keep Agent entry focused on conversation, not fixed interaction.""" - data.setdefault("contextPolicy", "inherit") - if data.get("entryMode") not in AGENT_ENTRY_MODES: - data["entryMode"] = "wait_user" - data.pop("entrySpeech", None) - data.setdefault("toolBindings", []) - data.setdefault("builtInTools", {}) - legacy_prompt = str(data.pop("visionFocusPrompt", "") or "").strip() - if "visionPrompt" not in data: - data["visionPrompt"] = legacy_prompt - if "visionPromptMode" not in data: - data["visionPromptMode"] = "guided" if legacy_prompt else "runtime" - if "inheritGlobalConfig" not in data: - has_node_overrides = any( - ( - data.get("llmResourceId"), - data.get("asrResourceId"), - data.get("ttsResourceId"), - data.get("visionEnabled"), - data.get("visionModelResourceId"), - data.get("visionPrompt"), - data.get("knowledgeBaseId"), - data.get("toolBindings"), - ) - ) - data["inheritGlobalConfig"] = not has_node_overrides - - -def _normalize_action_data(data: dict[str, Any]) -> None: - """Add deterministic Action defaults.""" - data.setdefault( - "toolBinding", - { - "toolId": "", - "argumentBindings": {}, - "templateBindings": {}, - "resultAssignments": {}, - }, - ) - data.setdefault("userInputPolicy", "queue") - data.pop("speech", None) - - -def _normalize_update_state_data(data: dict[str, Any]) -> None: - """Keep deterministic state updates as typed expression rules.""" - data.setdefault("updates", []) - - -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("title", "重要提示") - data.setdefault("message", "") - data.setdefault("confirmLabel", "确认") - if "completionPolicy" not in data: - data["completionPolicy"] = ( - MESSAGE_CONFIRMATION - if data.get("requireConfirmation") is True - else MESSAGE_PLAYBACK - ) - data.pop("requireConfirmation", None) - data.pop("showMessage", None) - - -def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: - # Missing values belong to graphs saved before workflow runtime/routing - # became explicit. Preserve their dedicated-router behavior. - settings.setdefault("runtimeMode", "pipeline") - settings.setdefault("llmRoutingMode", "llm_router") - settings.setdefault("globalPrompt", global_prompt) - settings.setdefault("defaultRealtimeResourceId", "") - settings.setdefault("defaultLlmResourceId", "") - settings.setdefault("defaultAsrResourceId", "") - settings.setdefault("defaultTtsResourceId", "") - settings.setdefault("visionEnabled", False) - settings.setdefault("visionModelResourceId", "") - legacy_prompt = str(settings.pop("visionFocusPrompt", "") or "").strip() - settings.setdefault("visionPrompt", legacy_prompt) - if "visionPromptMode" not in settings: - settings["visionPromptMode"] = "guided" if legacy_prompt else "runtime" - settings.setdefault("toolBindings", []) - settings.setdefault("builtInTools", {}) - settings.setdefault("knowledgeBaseId", "") - settings.setdefault("knowledgeMode", "automatic") - settings.setdefault("knowledgeTopN", 5) - settings.setdefault("knowledgeScoreThreshold", 0.0) - settings.setdefault("enableInterrupt", True) - settings.setdefault("turnConfig", {}) - - -def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: - """Return a deep-copied v4 graph.""" - source = deepcopy(graph or {}) - if str(source.get("specVersion") or "") == SPEC_VERSION: - settings = source.setdefault("settings", {}) - _normalize_settings(settings) - source.setdefault("nodes", []) - source.setdefault("edges", []) - for edge in source["edges"]: - edge["data"] = _edge_data_v3(edge) - for node in source["nodes"]: - data = node.setdefault("data", {}) - 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) - elif node.get("type") == "update_state": - _normalize_update_state_data(data) - return source - - nodes = source.get("nodes") or [] - edges = source.get("edges") or [] - global_prompt = "" - mapped_nodes: list[dict] = [] - start_prompt_nodes: dict[str, str] = {} - - type_map = { - "startCall": "start", - "agentNode": "agent", - "endCall": "end", - "start": "start", - "agent": "agent", - "message": "message", - "action": "action", - "update_state": "update_state", - "handoff": "handoff", - "end": "end", - } - for node in nodes: - old_type = str(node.get("type") or "") - data = deepcopy(node.get("data") or {}) - if old_type == "globalNode": - global_prompt = str(data.get("prompt") or "") - continue - new_type = type_map.get(old_type, old_type) - migrated = deepcopy(node) - migrated["type"] = new_type - if new_type == "end": - data["message"] = data.pop("message", data.pop("prompt", "")) - 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 == "update_state": - _normalize_update_state_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"): - data.pop(key, None) - migrated["data"] = data - mapped_nodes.append(migrated) - - mapped_edges: list[dict] = [] - for edge in edges: - migrated = deepcopy(edge) - migrated["data"] = _edge_data_v3(migrated) - mapped_edges.append(migrated) - - # A v2 Start was conversational. Insert a synthetic Agent so its prompt remains active. - for start_id, prompt in start_prompt_nodes.items(): - synthetic_id = f"{start_id}-migrated-agent" - start_node = next((n for n in mapped_nodes if n.get("id") == start_id), None) - position = (start_node or {}).get("position") or {"x": 100, "y": 120} - mapped_nodes.append( - { - "id": synthetic_id, - "type": "agent", - "position": {"x": position.get("x", 100) + 300, "y": position.get("y", 120)}, - "data": { - "name": "迁移的开场 Agent", - "prompt": prompt, - "contextPolicy": "inherit", - "inheritGlobalConfig": True, - "entryMode": "wait_user", - }, - } - ) - for edge in mapped_edges: - if edge.get("source") == start_id: - edge["source"] = synthetic_id - edge["data"] = _edge_data_v3(edge) - if str(edge["data"].get("condition") or "").strip(): - edge["data"]["mode"] = "llm" - mapped_edges.append( - { - "id": f"e-{start_id}-{synthetic_id}", - "source": start_id, - "target": synthetic_id, - "data": {"mode": "always", "priority": 0}, - } - ) - - settings = deepcopy(source.get("settings") or {}) - _normalize_settings(settings, global_prompt=global_prompt) - return { - "specVersion": 4, - "settings": settings, - "nodes": mapped_nodes, - "edges": mapped_edges, - **({"viewport": deepcopy(source["viewport"])} if source.get("viewport") else {}), - } +def copy_graph(graph: dict[str, Any] | None) -> dict[str, Any]: + """Return an isolated copy of a current Workflow graph.""" + return deepcopy(graph or {}) def _validate_expression(expression: Any) -> list[str]: @@ -421,16 +193,50 @@ def _validate_expression(expression: Any) -> list[str]: def validate_graph(graph: dict[str, Any]) -> list[str]: - graph = normalize_graph(graph) - nodes = graph.get("nodes") or [] - edges = graph.get("edges") or [] - if not nodes: - return [] - + graph = copy_graph(graph) errors: list[str] = [] - settings = graph.get("settings") or {} - runtime_mode = str(settings.get("runtimeMode") or "pipeline") - llm_routing_mode = str(settings.get("llmRoutingMode") or "llm_router") + if str(graph.get("specVersion") or "") != SPEC_VERSION: + errors.append(f"工作流只支持 specVersion {SPEC_VERSION}") + nodes = graph.get("nodes") + edges = graph.get("edges") + settings = graph.get("settings") + if not isinstance(nodes, list) or not nodes: + errors.append("工作流必须包含节点") + nodes = [] + if not isinstance(edges, list): + errors.append("工作流 edges 必须是列表") + edges = [] + if not isinstance(settings, dict): + errors.append("工作流 settings 必须是对象") + settings = {} + + required_settings = { + "runtimeMode", + "llmRoutingMode", + "globalPrompt", + "defaultRealtimeResourceId", + "defaultLlmResourceId", + "defaultAsrResourceId", + "defaultTtsResourceId", + "visionEnabled", + "visionModelResourceId", + "visionPromptMode", + "visionPrompt", + "toolBindings", + "builtInTools", + "knowledgeBaseId", + "knowledgeMode", + "knowledgeTopN", + "knowledgeScoreThreshold", + "enableInterrupt", + "turnConfig", + } + missing_settings = sorted(required_settings - settings.keys()) + if missing_settings: + errors.append("工作流 settings 缺少字段:" + ", ".join(missing_settings)) + + runtime_mode = str(settings.get("runtimeMode") or "") + llm_routing_mode = str(settings.get("llmRoutingMode") or "") if runtime_mode not in WORKFLOW_RUNTIME_MODES: errors.append(f"工作流运行模式无效:{runtime_mode}") if llm_routing_mode not in WORKFLOW_LLM_ROUTING_MODES: @@ -699,7 +505,7 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]: """Collect externally referenced IDs for save/runtime validation.""" - normalized = normalize_graph(graph) + normalized = copy_graph(graph) settings = normalized.get("settings") or {} resources = { str(value) diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index c5d4f99..24bb138 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -684,16 +684,6 @@ class KnowledgeRetrievalProcessor(FrameProcessor): self._last_signature = "" def _clear_context(self, messages: list[dict]) -> None: - # Remove the legacy Workflow knowledge message so an in-flight context - # created before this compatibility fix cannot keep sending that role. - messages[:] = [ - message - for message in messages - if not ( - message.get("role") == "developer" - and KNOWLEDGE_CONTEXT_MARKER in str(message.get("content") or "") - ) - ] system_message = next( (message for message in messages if message.get("role") == "system"), None, diff --git a/backend/services/system_tools.py b/backend/services/system_tools.py index 580c952..878a75f 100644 --- a/backend/services/system_tools.py +++ b/backend/services/system_tools.py @@ -38,7 +38,7 @@ RESERVED_TOOL_NAMES = frozenset( def system_tool_kind(definition: Mapping[str, Any] | None) -> str | None: - """Compatibility helper for old fixtures and the one-time migration.""" + """Read the platform tool kind from its runtime definition.""" config = (definition or {}).get("config") if not isinstance(config, Mapping): @@ -85,7 +85,12 @@ def runtime_system_tools( ), definition={ "type": "system", - "config": {"kind": kind, **normalized_config}, + "config": { + "kind": kind, + "allow_interruptions": True, + "execution_mode": "immediate", + **normalized_config, + }, }, ) ) diff --git a/backend/services/tool_policy.py b/backend/services/tool_policy.py index 358cf99..b98772a 100644 --- a/backend/services/tool_policy.py +++ b/backend/services/tool_policy.py @@ -1,9 +1,4 @@ -"""Runtime behavior shared by HTTP, MCP, and client tools. - -The stored definitions are intentionally plain dictionaries. Keeping policy -normalization here gives old definitions stable defaults without spreading -fallback rules across PromptBrain, WorkflowBrain, and the pipeline. -""" +"""Runtime behavior shared by HTTP, MCP, and client tools.""" from __future__ import annotations @@ -33,18 +28,14 @@ class ToolRuntimePolicy: def policy_for_tool(tool: RuntimeTool) -> ToolRuntimePolicy: - """Read policy with backwards-compatible defaults per tool kind.""" - config = (tool.definition or {}).get("config") or {} - default_mode = "async" if tool.type == "client" else "immediate" - mode = str(config.get("execution_mode") or default_mode) - if mode not in {"immediate", "async"}: - mode = default_mode - response_wait_mode = str(config.get("response_wait_mode") or "timeout") - if response_wait_mode not in {"timeout", "session"}: - response_wait_mode = "timeout" + """Read the validated policy stored on a current tool resource.""" + config = tool.definition["config"] + is_client = tool.type == "client" return ToolRuntimePolicy( - allow_interruptions=bool(config.get("allow_interruptions", True)), - execution_mode=mode, - wait_for_response=bool(config.get("wait_for_response", True)), - response_wait_mode=response_wait_mode, + allow_interruptions=bool(config["allow_interruptions"]), + execution_mode=str(config["execution_mode"]), + wait_for_response=(bool(config["wait_for_response"]) if is_client else True), + response_wait_mode=( + str(config["response_wait_mode"]) if is_client else "timeout" + ), ) diff --git a/backend/services/vision.py b/backend/services/vision.py index cdd49de..d038c6b 100644 --- a/backend/services/vision.py +++ b/backend/services/vision.py @@ -37,7 +37,7 @@ def _require(value: str, label: str) -> str: def normalize_vision_prompt_mode(value: object) -> VisionPromptMode: - """Return a supported mode while keeping older saved configs compatible.""" + """Return a supported vision prompt mode.""" mode = str(value or "runtime") if mode == "fixed": diff --git a/backend/services/workflow/realtime.py b/backend/services/workflow/realtime.py index a103417..076bb30 100644 --- a/backend/services/workflow/realtime.py +++ b/backend/services/workflow/realtime.py @@ -123,7 +123,7 @@ class RealtimeWorkflowOutput(WorkflowOutput): class WorkflowRealtimeController: - """Run Workflow v3 over one StepFun or Qwen realtime session.""" + """Run Workflow v4 over one StepFun or Qwen realtime session.""" def __init__( self, @@ -400,9 +400,10 @@ class WorkflowRealtimeController: tools = self._build_agent_tools(node_id) prompt = self._agent_prompt(node_id) await self._runtime.realtime.update_session(prompt, tools) - generate = str( - self._engine.data(node_id).get("entryMode") or "wait_user" - ) == "generate" + generate = ( + str(self._engine.data(node_id).get("entryMode", "wait_user")) + == "generate" + ) if generate: self._state.status = WorkflowStatus.RUNNING_AGENT else: @@ -1074,9 +1075,7 @@ class WorkflowRealtimeController: ) def _message_completion_policy(self, node_id: str) -> str: - return str( - self._engine.data(node_id).get("completionPolicy") or "playback" - ) + return str(self._engine.data(node_id)["completionPolicy"]) def _set_last_action(self, outcome) -> None: self._store.values.update( diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 195e3e8..905d1cc 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from hashlib import sha256 from typing import Any -from services.node_specs import normalize_graph +from services.node_specs import copy_graph from services.runtime_variables import DynamicVariableStore @@ -37,7 +37,7 @@ class AgentStageConfig: class WorkflowEngine: def __init__(self, graph: dict[str, Any]): - self.graph = normalize_graph(graph) + self.graph = copy_graph(graph) self.revision = self._revision_for(self.graph) self.settings = self.graph.get("settings") or {} self.nodes: dict[str, dict] = { @@ -272,7 +272,7 @@ class WorkflowEngine: return default if include_default else None def llm_edges(self, node_id: str) -> list[dict]: - """Edges considered by the compatibility dedicated router.""" + """Edges considered by the dedicated LLM router.""" return [ edge for edge in self.outgoing(node_id) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 2881049..f72db1c 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -119,6 +119,21 @@ async def noop_queue_frame(_frame): return None +def current_tool_definition(tool_type: str, **config) -> dict: + policy = { + "allow_interruptions": True, + "execution_mode": "async" if tool_type == "client" else "immediate", + } + if tool_type == "client": + policy.update( + { + "wait_for_response": True, + "response_wait_mode": "timeout", + } + ) + return {"type": tool_type, "config": {**policy, **config}} + + def system_runtime_tool(kind: str, **config) -> RuntimeTool: return RuntimeTool( id=f"tool_{kind}", @@ -126,7 +141,7 @@ def system_runtime_tool(kind: str, **config) -> RuntimeTool: function_name=kind, type="system", description=f"{kind} system tool", - definition={"type": "system", "config": {"kind": kind, **config}}, + definition=current_tool_definition("system", kind=kind, **config), ) @@ -306,18 +321,24 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): name="开场确认", function_name="show_message", type="client", + definition=current_tool_definition("client"), ) conversation_tool = RuntimeTool( id="lookup_order", name="查询订单", function_name="lookup_order", type="http", + definition=current_tool_definition("http"), ) cfg = AssistantConfig( type="prompt", tools=[startup_tool, conversation_tool], tool_bindings=[RuntimeToolBinding(tool_id=conversation_tool.id)], startup={ + "execution_mode": "sequential", + "opening_mode": "interruptible", + "entry_mode": "wait_user", + "opening_message": None, "actions": [ { "id": "opening_message", @@ -359,6 +380,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): name=f"预检动作 {index}", function_name=f"preflight_{index}", type="http", + definition=current_tool_definition("http"), ) for index in (1, 2) ] @@ -366,6 +388,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): type="prompt", tools=tools, startup={ + "execution_mode": "sequential", + "opening_mode": "interruptible", + "entry_mode": "wait_user", + "opening_message": None, "actions": [ { "id": tool.id, @@ -416,6 +442,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): name="加载开场数据", function_name="load_opening_data", type="http", + definition=current_tool_definition("http"), ) cfg = AssistantConfig( type="prompt", @@ -590,8 +617,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): type="prompt", greeting="请先确认", startup={ + "execution_mode": "sequential", "opening_mode": "confirmation", "entry_mode": "wait_user", + "actions": [], "opening_message": { "title": "重要提示", "message": "请确认已阅读。", @@ -634,8 +663,11 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): type="prompt", greeting="固定开场白", startup={ + "execution_mode": "sequential", "opening_mode": "playback", "entry_mode": "generate", + "actions": [], + "opening_message": None, }, ) brain = build_brain(cfg) @@ -675,8 +707,11 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): type="prompt", greeting="可以打断的开场白", startup={ + "execution_mode": "sequential", "opening_mode": "interruptible", "entry_mode": "generate", + "actions": [], + "opening_message": None, }, ) brain = build_brain(cfg) @@ -712,8 +747,11 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): type="prompt", greeting="可以打断的开场白", startup={ + "execution_mode": "sequential", "opening_mode": "interruptible", "entry_mode": "generate", + "actions": [], + "opening_message": None, }, ) brain = build_brain(cfg) @@ -749,6 +787,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): cfg = AssistantConfig( type="prompt", startup={ + "execution_mode": "sequential", + "opening_mode": "confirmation", + "entry_mode": "generate", + "actions": [], "opening_message": { "title": "重要提示", "message": "请确认已阅读。", @@ -1312,6 +1354,8 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): description="查询订单状态", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "immediate", "method": "GET", "url": "https://example.test/orders/{order_id}", "headers": {"Authorization": "Bearer {{secret__token}}"}, @@ -1389,7 +1433,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -1511,7 +1555,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -1567,7 +1611,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_workflow_end_conversation_waits_for_generated_speech(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [{"id": "start", "type": "start", "data": {}}], "edges": [], @@ -1644,7 +1688,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): def test_client_tool_session_wait_disables_flow_timeout(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [{"id": "start", "type": "start", "data": {}}], "edges": [], @@ -1657,7 +1701,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): type="client", definition={ "type": "client", - "config": {"timeout_seconds": 7}, + "config": { + "allow_interruptions": True, + "execution_mode": "async", + "wait_for_response": True, + "response_wait_mode": "timeout", + "timeout_seconds": 7, + }, }, ) session_tool = RuntimeTool( @@ -1668,6 +1718,9 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): definition={ "type": "client", "config": { + "allow_interruptions": True, + "execution_mode": "async", + "wait_for_response": True, "timeout_seconds": 7, "response_wait_mode": "session", }, @@ -1687,7 +1740,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_flow_transition_honors_strict_agent_entry_mode(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -1749,7 +1802,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "面板 {{panel_open}}"}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -1812,7 +1865,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_agent_vision_tool_is_scoped_to_effective_stage(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": { "globalPrompt": "全局规则", "defaultLlmResourceId": "llm_global", @@ -1889,16 +1942,16 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertNotIn("fetch_user_image", custom_config["role_message"]) self.assertFalse(scopes[-1]["enabled"]) - async def test_initial_message_starts_without_workflow_greeting(self): + async def test_initial_message_starts_directly(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "全局规则"}, "nodes": [ { "id": "start", "type": "start", - "data": {"greeting": "欢迎使用"}, + "data": {}, }, { "id": "message", @@ -1957,13 +2010,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ) brain._manager = FakeManager() - 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, "message") @@ -1989,12 +2035,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): name="查询订单", function_name="lookup_order", type="http", + definition=current_tool_definition("http"), ) cfg = prepare_dynamic_config( AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2082,12 +2129,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): name="客户端操作", function_name="show_message", type="client", + definition=current_tool_definition("client"), ) brain = WorkflowBrain( AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2155,12 +2203,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): name="客户端操作", function_name="show_message", type="client", + definition=current_tool_definition("client"), ) brain = WorkflowBrain( AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2209,12 +2258,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): name="客户端操作", function_name="show_message", type="client", + definition=current_tool_definition("client"), ) brain = WorkflowBrain( AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2280,12 +2330,13 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): name="客户端操作", function_name="show_message", type="client", + definition=current_tool_definition("client"), ) brain = WorkflowBrain( AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2340,7 +2391,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2441,7 +2492,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_confirmation_message_is_forwarded_to_next_agent_context(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2534,14 +2585,17 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_speech_only_message_waits_for_transport_playback(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, { "id": "message", "type": "message", - "data": {"speech": "正在为您准备服务。"}, + "data": { + "speech": "正在为您准备服务。", + "completionPolicy": "playback", + }, }, ], "edges": [], @@ -2581,7 +2635,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_interruptible_message_forwards_multimodal_input_once(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -2707,14 +2761,17 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_message_between_agents_resumes_after_playback(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, { "id": "opening", "type": "message", - "data": {"speech": "欢迎使用。"}, + "data": { + "speech": "欢迎使用。", + "completionPolicy": "playback", + }, }, { "id": "agent1", @@ -2727,7 +2784,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): { "id": "middle", "type": "message", - "data": {"speech": "现在进入信息确认。"}, + "data": { + "speech": "现在进入信息确认。", + "completionPolicy": "playback", + }, }, { "id": "agent2", @@ -2928,7 +2988,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): start_brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [{"id": "start", "type": "start", "data": {}}], "edges": [], @@ -2960,7 +3020,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): agent_brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "全局规则"}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -3000,7 +3060,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): handoff_brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -3027,7 +3087,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_waiting_agent_does_not_reply_to_transitioning_user_turn(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "全局规则"}, "nodes": [ { @@ -3167,7 +3227,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_start_expression_condition_also_waits_for_user_turn(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -3206,7 +3266,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_automatic_node_can_follow_llm_condition(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "全局规则"}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -3268,7 +3328,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_mixed_edge_conditions_follow_priority(self): brain = WorkflowBrain( { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -3352,7 +3412,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def test_transition_and_end_are_owned_by_workflow_brain(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": { "globalPrompt": "全局规则", "defaultLlmResourceId": "llm_global", diff --git a/backend/tests/test_client_tools.py b/backend/tests/test_client_tools.py index 61bce20..91e7829 100644 --- a/backend/tests/test_client_tools.py +++ b/backend/tests/test_client_tools.py @@ -26,6 +26,10 @@ def client_tool() -> RuntimeTool: "schema_version": 1, "type": "client", "config": { + "allow_interruptions": True, + "execution_mode": "async", + "wait_for_response": True, + "response_wait_mode": "timeout", "parameters": [ { "name": "visible", @@ -70,10 +74,10 @@ class FakeClientTools: class ClientToolExecutorTests(unittest.IsolatedAsyncioTestCase): def test_response_wait_mode_defaults_and_uses_camel_case(self): - legacy = ClientToolConfig() + default_config = ClientToolConfig() session = ClientToolConfig(responseWaitMode="session") - self.assertEqual(legacy.response_wait_mode, "timeout") + self.assertEqual(default_config.response_wait_mode, "timeout") self.assertEqual(session.response_wait_mode, "session") self.assertEqual( session.model_dump(by_alias=True)["responseWaitMode"], diff --git a/backend/tests/test_debug_auto_test.py b/backend/tests/test_debug_auto_test.py index c4d78ed..45a3fb1 100644 --- a/backend/tests/test_debug_auto_test.py +++ b/backend/tests/test_debug_auto_test.py @@ -79,7 +79,13 @@ def system_runtime_tool(kind: str) -> RuntimeTool: name=kind, function_name=kind, type="system", - definition={"config": {"kind": kind}}, + definition={ + "config": { + "kind": kind, + "allow_interruptions": True, + "execution_mode": "immediate", + } + }, ) diff --git a/backend/tests/test_fixed_speech_playback.py b/backend/tests/test_fixed_speech_playback.py index 0a7d411..0c93d1b 100644 --- a/backend/tests/test_fixed_speech_playback.py +++ b/backend/tests/test_fixed_speech_playback.py @@ -65,7 +65,7 @@ class FixedSpeechPlaybackTest(unittest.IsolatedAsyncioTestCase): async def test_realtime_end_ignores_unrelated_speech_boundaries(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, @@ -144,7 +144,7 @@ class FixedSpeechPlaybackTest(unittest.IsolatedAsyncioTestCase): async def test_realtime_confirmation_is_appended_without_running_model(self): graph = { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ { diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index 4092cf4..de32af2 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -46,10 +46,6 @@ class KnowledgeToolDescriptionTest(unittest.TestCase): processor = KnowledgeRetrievalProcessor(None) messages = [ {"role": "assistant", "content": "你好"}, - { - "role": "developer", - "content": f"{KNOWLEDGE_CONTEXT_MARKER}\n旧检索结果", - }, ] processor._set_context( @@ -59,7 +55,6 @@ class KnowledgeToolDescriptionTest(unittest.TestCase): self.assertEqual(messages[0]["role"], "system") self.assertIn("新检索结果", messages[0]["content"]) - self.assertFalse(any(message["role"] == "developer" for message in messages)) class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): diff --git a/backend/tests/test_prompt_realtime_tools.py b/backend/tests/test_prompt_realtime_tools.py index b98695f..d862d31 100644 --- a/backend/tests/test_prompt_realtime_tools.py +++ b/backend/tests/test_prompt_realtime_tools.py @@ -335,7 +335,16 @@ class PromptRealtimeToolControllerTests(unittest.IsolatedAsyncioTestCase): name="显示拍照按钮", function_name="set_photo_button_visible", type="client", - definition={"type": "client", "config": {"parameters": []}}, + definition={ + "type": "client", + "config": { + "allow_interruptions": True, + "execution_mode": "async", + "wait_for_response": True, + "response_wait_mode": "timeout", + "parameters": [], + }, + }, ) ], tool_bindings=[RuntimeToolBinding(tool_id="tool_photo")], @@ -502,13 +511,20 @@ class PromptRealtimeCapabilityTests(unittest.TestCase): "qwen-audio-realtime", tools=[ RuntimeTool( - id="legacy", - name="旧外部工具", + id="reserved_name_tool", + name="保留名冲突工具", function_name="end_conversation", type="http", + definition={ + "type": "http", + "config": { + "allow_interruptions": True, + "execution_mode": "immediate", + }, + }, ) ], - tool_bindings=[RuntimeToolBinding(tool_id="legacy")], + tool_bindings=[RuntimeToolBinding(tool_id="reserved_name_tool")], ) ) diff --git a/backend/tests/test_runtime_variables.py b/backend/tests/test_runtime_variables.py index 4453efa..3e5cc0e 100644 --- a/backend/tests/test_runtime_variables.py +++ b/backend/tests/test_runtime_variables.py @@ -83,7 +83,7 @@ class DynamicVariableTests(unittest.TestCase): cfg = AssistantConfig( type="workflow", graph={ - "specVersion": 3, + "specVersion": 4, "settings": {"globalPrompt": "称呼 {{nickname}}"}, "nodes": [ { diff --git a/backend/tests/test_startup_actions.py b/backend/tests/test_startup_actions.py index da6d0f3..3ccfcd4 100644 --- a/backend/tests/test_startup_actions.py +++ b/backend/tests/test_startup_actions.py @@ -91,6 +91,8 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): type="prompt", runtimeMode="realtime", startup={ + "openingMode": "confirmation", + "entryMode": "generate", "openingMessage": { "title": "重要提示", "message": "请确认已阅读。", @@ -104,6 +106,8 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): name="内置开场消息", type="prompt", startup={ + "openingMode": "confirmation", + "entryMode": "generate", "openingMessage": { "title": "重要提示", "message": "请确认已阅读。", @@ -128,6 +132,8 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): status="active", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "async", "wait_for_response": True, "response_wait_mode": "session", } @@ -147,6 +153,8 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): status="active", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "async", "wait_for_response": True, "response_wait_mode": "timeout", } @@ -168,6 +176,8 @@ class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): status="active", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "async", "wait_for_response": True, "response_wait_mode": "session", } diff --git a/backend/tests/test_tool_executor_mcp.py b/backend/tests/test_tool_executor_mcp.py index 8a541aa..d3c6c1f 100644 --- a/backend/tests/test_tool_executor_mcp.py +++ b/backend/tests/test_tool_executor_mcp.py @@ -17,6 +17,8 @@ def mcp_tool() -> RuntimeTool: "schema_version": 1, "type": "mcp", "config": { + "allow_interruptions": True, + "execution_mode": "immediate", "remote_tool_name": "get_order", "input_schema": { "type": "object", diff --git a/backend/tests/test_tool_organization_v4.py b/backend/tests/test_tool_organization_v4.py index d98f507..ff81be2 100644 --- a/backend/tests/test_tool_organization_v4.py +++ b/backend/tests/test_tool_organization_v4.py @@ -58,7 +58,7 @@ class ToolOrganizationContractTests(unittest.TestCase): request = AssistantUpsert(**data) self.assertIsNotNone(request.built_in_tools.end_conversation) - response = AssistantOut(id="asst_legacy", **data) + response = AssistantOut(id="asst_realtime", **data) self.assertIsNotNone(response.built_in_tools.end_conversation) def test_workflow_camel_case_builtin_config_is_normalized_for_runtime(self): @@ -247,6 +247,8 @@ class ToolBindingTests(unittest.IsolatedAsyncioTestCase): type="http", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "immediate", "parameters": [ {"name": "order_id", "type": "string", "required": True}, {"name": "tenant", "type": "string", "required": True}, @@ -280,6 +282,8 @@ class ToolBindingTests(unittest.IsolatedAsyncioTestCase): type="http", definition={ "config": { + "allow_interruptions": True, + "execution_mode": "immediate", "parameters": [ {"name": "tenant", "type": "string", "required": True} ] diff --git a/backend/tests/test_tool_policy.py b/backend/tests/test_tool_policy.py index 20c0c19..cadb1a5 100644 --- a/backend/tests/test_tool_policy.py +++ b/backend/tests/test_tool_policy.py @@ -24,19 +24,9 @@ def runtime_tool(tool_type: str, config: dict | None = None) -> RuntimeTool: class ToolPolicyTests(unittest.TestCase): - def test_backwards_compatible_execution_defaults(self): - self.assertEqual( - policy_for_tool(runtime_tool("http")).execution_mode, - "immediate", - ) - self.assertEqual( - policy_for_tool(runtime_tool("client")).execution_mode, - "async", - ) - self.assertEqual( - policy_for_tool(runtime_tool("client")).response_wait_mode, - "timeout", - ) + def test_missing_current_policy_is_rejected(self): + with self.assertRaises(KeyError): + policy_for_tool(runtime_tool("http")) def test_explicit_policy_is_normalized(self): policy = policy_for_tool( @@ -55,6 +45,22 @@ class ToolPolicyTests(unittest.TestCase): self.assertFalse(policy.wait_for_response) self.assertEqual(policy.response_wait_mode, "session") + def test_non_client_tools_use_fixed_response_wait_policy(self): + policy = policy_for_tool( + runtime_tool( + "http", + { + "allow_interruptions": True, + "execution_mode": "async", + }, + ) + ) + + self.assertTrue(policy.allow_interruptions) + self.assertEqual(policy.execution_mode, "async") + self.assertTrue(policy.wait_for_response) + self.assertEqual(policy.response_wait_mode, "timeout") + class ToolInterruptionStrategyTests(unittest.IsolatedAsyncioTestCase): async def test_releases_only_after_tool_and_following_speech_finish(self): diff --git a/backend/tests/test_vision_prompt_modes.py b/backend/tests/test_vision_prompt_modes.py index f60c3d8..c34231f 100644 --- a/backend/tests/test_vision_prompt_modes.py +++ b/backend/tests/test_vision_prompt_modes.py @@ -5,7 +5,7 @@ from types import SimpleNamespace from models import AssistantConfig from services.brains.base import RealtimeBrainRuntime -from services.node_specs import normalize_graph, validate_graph +from services.node_specs import validate_graph from services.vision import ( VISION_SYSTEM_HINT, resolve_vision_question, @@ -16,6 +16,35 @@ from services.workflow_engine import WorkflowEngine from services.workflow.realtime import WorkflowRealtimeController +def workflow_settings(**overrides): + settings = { + "runtimeMode": "pipeline", + "llmRoutingMode": "llm_router", + "globalPrompt": "", + "defaultRealtimeResourceId": "", + "defaultLlmResourceId": "", + "defaultAsrResourceId": "", + "defaultTtsResourceId": "", + "visionEnabled": False, + "visionModelResourceId": "", + "visionPromptMode": "runtime", + "visionPrompt": "", + "toolBindings": [], + "builtInTools": {}, + "knowledgeBaseId": "", + "knowledgeMode": "disabled", + "knowledgeTopN": 5, + "knowledgeScoreThreshold": 0.0, + "enableInterrupt": True, + "turnConfig": { + "bargeIn": {"strategy": "vad"}, + "turnDetection": {"strategy": "smart_turn"}, + }, + } + settings.update(overrides) + return settings + + class VisionPromptModeTest(unittest.TestCase): def test_fixed_mode_has_no_runtime_question_parameter(self): description, properties, required = vision_tool_schema_parts( @@ -80,12 +109,14 @@ class VisionPromptModeTest(unittest.TestCase): class WorkflowVisionPromptModeTest(unittest.TestCase): def test_workflow_resolves_global_and_node_modes(self): graph = { - "specVersion": 3, - "settings": { - "visionEnabled": True, - "visionPromptMode": "fixed", - "visionPrompt": "全局固定问题", - }, + "specVersion": 4, + "settings": workflow_settings( + **{ + "visionEnabled": True, + "visionPromptMode": "fixed", + "visionPrompt": "全局固定问题", + } + ), "nodes": [ { "id": "global-agent", @@ -116,36 +147,37 @@ class WorkflowVisionPromptModeTest(unittest.TestCase): self.assertEqual(custom_stage.vision_prompt_mode, "guided") self.assertEqual(custom_stage.vision_prompt, "节点引导") - def test_legacy_focus_is_migrated_to_guided_mode(self): - normalized = normalize_graph( - { - "specVersion": 3, - "settings": {"visionFocusPrompt": "旧全局关注目标"}, - "nodes": [], - "edges": [], - } - ) - - self.assertEqual(normalized["settings"]["visionPromptMode"], "guided") - self.assertEqual(normalized["settings"]["visionPrompt"], "旧全局关注目标") - self.assertNotIn("visionFocusPrompt", normalized["settings"]) - def test_fixed_or_guided_mode_requires_prompt_when_vision_is_enabled(self): graph = { - "specVersion": 3, - "settings": { - "visionEnabled": True, - "visionPromptMode": "fixed", - "visionPrompt": "", - }, + "specVersion": 4, + "settings": workflow_settings( + **{ + "visionEnabled": True, + "visionPromptMode": "fixed", + "visionPrompt": "", + } + ), "nodes": [ + {"id": "start", "type": "start", "data": {}}, { "id": "agent", "type": "agent", - "data": {"name": "Agent", "prompt": "处理任务"}, + "data": { + "name": "Agent", + "prompt": "处理任务", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + }, + } + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, } ], - "edges": [], } self.assertTrue( @@ -154,16 +186,29 @@ class WorkflowVisionPromptModeTest(unittest.TestCase): def test_invalid_mode_is_reported_instead_of_silently_rewritten(self): graph = { - "specVersion": 3, - "settings": {"visionPromptMode": "unexpected"}, + "specVersion": 4, + "settings": workflow_settings(visionPromptMode="unexpected"), "nodes": [ + {"id": "start", "type": "start", "data": {}}, { "id": "agent", "type": "agent", - "data": {"name": "Agent", "prompt": "处理任务"}, + "data": { + "name": "Agent", + "prompt": "处理任务", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + }, + } + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, } ], - "edges": [], } self.assertTrue( @@ -172,8 +217,8 @@ class WorkflowVisionPromptModeTest(unittest.TestCase): def test_realtime_tool_contract_changes_with_mode(self): graph = { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": workflow_settings(), "nodes": [ { "id": "fixed-agent", diff --git a/backend/tests/test_voice_webrtc_lifecycle.py b/backend/tests/test_voice_webrtc_lifecycle.py index 3fc01c3..b1dc8e9 100644 --- a/backend/tests/test_voice_webrtc_lifecycle.py +++ b/backend/tests/test_voice_webrtc_lifecycle.py @@ -52,7 +52,7 @@ class WebRTCOfferMetadataTest(unittest.IsolatedAsyncioTestCase): ["set_photo_button_visible", "show_message"], ) - async def test_client_tool_request_data_remains_a_compatibility_fallback(self): + async def test_client_tool_request_data_is_not_a_capability_channel(self): request = offer_request() payload = { "sdp": "offer", @@ -68,7 +68,7 @@ class WebRTCOfferMetadataTest(unittest.IsolatedAsyncioTestCase): await voice_webrtc.http_offer(request, payload) forwarded = handle_offer.await_args.args[0] - self.assertEqual(forwarded["client_tool_names"], ["show_message"]) + self.assertEqual(forwarded["client_tool_names"], []) async def test_invalid_client_tool_header_is_rejected(self): request = offer_request(client_tools={"name": "show_message"}) diff --git a/backend/tests/test_workflow_routing.py b/backend/tests/test_workflow_routing.py index 618f304..366b130 100644 --- a/backend/tests/test_workflow_routing.py +++ b/backend/tests/test_workflow_routing.py @@ -10,7 +10,7 @@ from services.workflow_engine import WorkflowEngine def routing_graph() -> dict: return { - "specVersion": 3, + "specVersion": 4, "settings": {}, "nodes": [ {"id": "start", "type": "start", "data": {}}, diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v4.py similarity index 80% rename from backend/tests/test_workflow_v3.py rename to backend/tests/test_workflow_v4.py index 71ff72d..ed047a3 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v4.py @@ -15,22 +15,55 @@ from routes.assistants import ( ) from schemas import AssistantUpsert from services.pipecat.service_factory import config_with_resource -from services.node_specs import graph_references, normalize_graph, validate_graph +from services.node_specs import graph_references, validate_graph from services.runtime_variables import DynamicVariableStore, prepare_dynamic_config from services.vision import config_with_vision_resource from services.workflow_engine import WorkflowEngine +def valid_settings(): + return { + "runtimeMode": "pipeline", + "llmRoutingMode": "llm_router", + "globalPrompt": "服务 {{customer}}", + "defaultRealtimeResourceId": "", + "defaultLlmResourceId": "", + "defaultAsrResourceId": "", + "defaultTtsResourceId": "", + "visionEnabled": False, + "visionModelResourceId": "", + "visionPromptMode": "runtime", + "visionPrompt": "", + "toolBindings": [], + "builtInTools": {}, + "knowledgeBaseId": "", + "knowledgeMode": "disabled", + "knowledgeTopN": 5, + "knowledgeScoreThreshold": 0.0, + "enableInterrupt": True, + "turnConfig": { + "bargeIn": {"strategy": "vad"}, + "turnDetection": {"strategy": "smart_turn"}, + }, + } + + def valid_graph(): return { "specVersion": 4, - "settings": {"globalPrompt": "服务 {{customer}}"}, + "settings": valid_settings(), "nodes": [ {"id": "start", "type": "start", "data": {"name": "Start"}}, { "id": "agent", "type": "agent", - "data": {"name": "Agent", "prompt": "处理订单", "contextPolicy": "fresh"}, + "data": { + "name": "Agent", + "prompt": "处理订单", + "contextPolicy": "fresh", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + }, }, {"id": "end", "type": "end", "data": {"name": "End", "scope": "flow"}}, ], @@ -61,19 +94,15 @@ 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 开场白" + def test_rejects_non_current_workflow_schema(self): + graph = valid_graph() + graph["specVersion"] = 3 + graph["settings"].pop("runtimeMode") - normalized = normalize_graph(body.graph) + errors = validate_graph(graph) - self.assertEqual(body.greeting, "") - self.assertNotIn("greeting", normalized["nodes"][0]["data"]) + self.assertIn("工作流只支持 specVersion 4", errors) + self.assertTrue(any("缺少字段:runtimeMode" in error for error in errors)) def test_revision_pins_the_normalized_workflow_snapshot(self): first = WorkflowEngine(valid_graph()) @@ -109,7 +138,7 @@ class WorkflowGraphTests(unittest.TestCase): type="workflow", graph=graph, visionEnabled=False, - visionModelResourceId="legacy_vision", + visionModelResourceId="ignored_top_level_vision", ) _validate_workflow(body) @@ -119,66 +148,16 @@ class WorkflowGraphTests(unittest.TestCase): 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") + agent = next(node for node in graph["nodes"] if node["type"] == "agent") self.assertEqual(agent["data"]["entryMode"], "wait_user") - self.assertNotIn("entrySpeech", agent["data"]) self.assertTrue(agent["data"]["inheritGlobalConfig"]) self.assertEqual(agent["data"]["contextPolicy"], "fresh") agent["data"]["entryMode"] = "generate" - self.assertEqual(validate_graph(normalized), []) + self.assertEqual(validate_graph(graph), []) 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_edge_speech_fields_are_removed(self): - graph = valid_graph() - graph["edges"][0]["data"]["transitionSpeech"] = "不再播放" - graph["edges"][1]["data"]["transition_speech"] = "旧字段也不再播放" - - normalized = normalize_graph(graph) - - for edge in normalized["edges"]: - self.assertNotIn("transitionSpeech", edge["data"]) - self.assertNotIn("transition_speech", edge["data"]) - - def test_action_defaults_use_one_binding_contract(self): - graph = valid_graph() - graph["nodes"].extend( - [ - { - "id": "legacy_override", - "type": "action", - "data": {"toolBinding": {"toolId": "lookup"}}, - }, - {"id": "legacy_none", "type": "action", "data": {}}, - { - "id": "explicit_inherit", - "type": "action", - "data": {"toolBinding": {"toolId": "notify"}}, - }, - ] - ) - - normalized = normalize_graph(graph) - actions = { - node["id"]: node["data"] - for node in normalized["nodes"] - if node["type"] == "action" - } - - self.assertEqual(actions["legacy_override"]["toolBinding"]["toolId"], "lookup") - self.assertEqual(actions["legacy_none"]["toolBinding"]["toolId"], "") - self.assertEqual(actions["explicit_inherit"]["toolBinding"]["toolId"], "notify") - self.assertEqual(actions["legacy_none"]["userInputPolicy"], "queue") - self.assertNotIn("speech", actions["legacy_none"]) + self.assertTrue(any("进入模式无效" in error for error in validate_graph(graph))) def test_action_advanced_settings_are_validated(self): graph = valid_graph() @@ -197,23 +176,32 @@ 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): + def test_message_contract_and_validation(self): graph = valid_graph() graph["nodes"].append( { "id": "message", "type": "message", - "data": {"speech": "请确认重要信息"}, + "data": { + "speech": "请确认重要信息", + "completionPolicy": "playback", + }, } ) - normalized = normalize_graph(graph) - message = next( - node for node in normalized["nodes"] if node["type"] == "message" + graph["edges"][0]["target"] = "message" + graph["edges"].insert( + 1, + { + "id": "after-message", + "source": "message", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, ) - self.assertEqual(message["data"]["completionPolicy"], "playback") - self.assertNotIn("requireConfirmation", message["data"]) - self.assertNotIn("showMessage", message["data"]) - self.assertEqual(message["data"]["confirmLabel"], "确认") + message = next( + node for node in graph["nodes"] if node["type"] == "message" + ) + self.assertEqual(validate_graph(graph), []) message["data"].update( { @@ -224,40 +212,20 @@ class WorkflowGraphTests(unittest.TestCase): "completionPolicy": "confirmation", } ) - errors = validate_graph(normalized) + errors = validate_graph(graph) self.assertTrue(any("弹窗消息必须为" in error for error in errors)) self.assertTrue(any("必须配置播报内容" in error for error in errors)) message["data"].update( {"speech": "", "completionPolicy": "playback"} ) - errors = validate_graph(normalized) + errors = validate_graph(graph) self.assertTrue(any("必须配置播报内容" in error for error in errors)) message["data"]["completionPolicy"] = "unknown" - errors = validate_graph(normalized) + errors = validate_graph(graph) self.assertTrue(any("完成策略无效" in error for error in errors)) - legacy = valid_graph() - legacy["nodes"].append( - { - "id": "legacy-message", - "type": "message", - "data": { - "speech": "请确认", - "showMessage": True, - "title": "提示", - "message": "请确认", - "confirmLabel": "确认", - "requireConfirmation": True, - }, - } - ) - legacy_message = normalize_graph(legacy)["nodes"][-1]["data"] - self.assertEqual(legacy_message["completionPolicy"], "confirmation") - self.assertNotIn("requireConfirmation", legacy_message) - self.assertNotIn("showMessage", legacy_message) - def test_voice_resource_creates_isolated_runtime_config(self): base = AssistantConfig(type="workflow", asr="default", voice="default") asr = RuntimeModelResource( @@ -327,16 +295,6 @@ class WorkflowGraphTests(unittest.TestCase): self.assertEqual(refs["tools"], {"tool_global", "tool_agent"}) self.assertEqual(refs["knowledge_bases"], {"kb_global", "kb_agent"}) - def test_existing_agent_override_disables_implicit_inheritance(self): - graph = valid_graph() - agent = next(node for node in graph["nodes"] if node["type"] == "agent") - agent["data"]["toolBindings"] = [{"toolId": "legacy_tool"}] - normalized = normalize_graph(graph) - normalized_agent = next( - node for node in normalized["nodes"] if node["type"] == "agent" - ) - self.assertFalse(normalized_agent["data"]["inheritGlobalConfig"]) - def test_inherited_agent_ignores_stale_custom_references(self): graph = valid_graph() agent = next(node for node in graph["nodes"] if node["type"] == "agent") @@ -525,20 +483,24 @@ class WorkflowGraphTests(unittest.TestCase): def test_start_agent_action_and_handoff_may_have_no_outgoing_edge(self): terminal_graphs = [ { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": valid_settings(), "nodes": [{"id": "start", "type": "start", "data": {}}], "edges": [], }, { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": valid_settings(), "nodes": [ {"id": "start", "type": "start", "data": {}}, { "id": "agent", "type": "agent", - "data": {"prompt": "持续处理用户问题"}, + "data": { + "prompt": "持续处理用户问题", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + }, }, ], "edges": [ @@ -551,8 +513,8 @@ class WorkflowGraphTests(unittest.TestCase): ], }, { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": valid_settings(), "nodes": [ {"id": "start", "type": "start", "data": {}}, {"id": "handoff", "type": "handoff", "data": {}}, @@ -573,11 +535,18 @@ class WorkflowGraphTests(unittest.TestCase): self.assertEqual(validate_graph(graph), []) action_without_exit = { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": valid_settings(), "nodes": [ {"id": "start", "type": "start", "data": {}}, - {"id": "action", "type": "action", "data": {"toolBinding": {"toolId": "tool"}}}, + { + "id": "action", + "type": "action", + "data": { + "toolBinding": {"toolId": "tool"}, + "userInputPolicy": "queue", + }, + }, ], "edges": [ { @@ -592,11 +561,18 @@ class WorkflowGraphTests(unittest.TestCase): def test_agent_cannot_have_only_one_default_path(self): graph = { - "specVersion": 3, - "settings": {}, + "specVersion": 4, + "settings": valid_settings(), "nodes": [ {"id": "start", "type": "start", "data": {}}, - {"id": "agent", "type": "agent", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": { + "inheritGlobalConfig": True, + "entryMode": "wait_user", + }, + }, {"id": "end", "type": "end", "data": {}}, ], "edges": [ @@ -699,7 +675,17 @@ class WorkflowGraphTests(unittest.TestCase): } for source_type, parts in cases.items(): - graph = {"specVersion": 3, "settings": {}, **parts} + for node in parts["nodes"]: + if node["type"] == "agent": + node["data"].update( + { + "inheritGlobalConfig": True, + "entryMode": "wait_user", + } + ) + elif node["type"] == "action": + node["data"]["userInputPolicy"] = "queue" + graph = {"specVersion": 4, "settings": valid_settings(), **parts} with self.subTest(source_type=source_type): self.assertEqual(validate_graph(graph), []) @@ -715,41 +701,6 @@ class WorkflowGraphTests(unittest.TestCase): any("最多只能有一条默认路径" in error for error in validate_graph(graph)) ) - def test_v2_condition_on_non_agent_becomes_llm_path(self): - graph = normalize_graph( - { - "nodes": [ - {"id": "start", "type": "start", "data": {}}, - {"id": "handoff", "type": "handoff", "data": {}}, - ], - "edges": [ - { - "id": "route", - "source": "start", - "target": "handoff", - "data": {"condition": "用户需要人工服务"}, - } - ], - } - ) - self.assertEqual(graph["edges"][0]["data"]["mode"], "llm") - - def test_v2_start_prompt_is_preserved_in_synthetic_agent(self): - graph = normalize_graph( - { - "nodes": [ - {"id": "s", "type": "startCall", "data": {"prompt": "询问需求"}}, - {"id": "e", "type": "endCall", "data": {"prompt": "再见"}}, - ], - "edges": [ - {"id": "done", "source": "s", "target": "e", "data": {"condition": "完成"}} - ], - } - ) - migrated = next(node for node in graph["nodes"] if node["type"] == "agent") - self.assertEqual(migrated["data"]["prompt"], "询问需求") - self.assertEqual(validate_graph(graph), []) - def test_rejects_automatic_cycle_and_duplicate_priorities(self): graph = valid_graph() graph["nodes"].insert( @@ -757,7 +708,10 @@ class WorkflowGraphTests(unittest.TestCase): { "id": "action", "type": "action", - "data": {"toolBinding": {"toolId": "tool"}}, + "data": { + "toolBinding": {"toolId": "tool"}, + "userInputPolicy": "queue", + }, }, ) graph["edges"] = [ diff --git a/frontend/src/components/assistant-editor/debug-auto-test.tsx b/frontend/src/components/assistant-editor/debug-auto-test.tsx index 9f22354..a4bfaf4 100644 --- a/frontend/src/components/assistant-editor/debug-auto-test.tsx +++ b/frontend/src/components/assistant-editor/debug-auto-test.tsx @@ -447,34 +447,6 @@ function startupAssistantMessages(startup: BatchStartupResult): string[] { return [greeting, ...assistantMessages].filter(Boolean); } -function legacyStartupRuntimeRecords( - startup: BatchStartupResult, -): Array> { - return startup.autoConfirmedMessages.flatMap((message, index) => { - const confirmedAction = - message.actions?.find((action) => action.id === "confirmed") ?? - message.actions?.[0]; - const legacyId = `${index}-${message.title ?? message.message ?? "message"}`; - return [ - { - eventId: `legacy-message-started-${legacyId}`, - event: "message_started", - phase: "opening", - requiresConfirmation: true, - messageTitle: message.title?.trim() || "未命名消息", - messageContent: message.message?.trim() || "(没有消息正文)", - messageActions: message.actions ?? [], - }, - { - eventId: `legacy-message-completed-${legacyId}`, - event: "message_completed", - phase: "opening", - action: confirmedAction?.id?.trim() || "confirmed", - }, - ]; - }); -} - function runtimeRecordKey( event: Record, prefix: string, @@ -514,10 +486,7 @@ function AutoTestTimeline({ }); const openingMessages = startupAssistantMessages(startup); - const records = - (startup.runtimeRecords?.length ?? 0) > 0 - ? startup.runtimeRecords ?? [] - : legacyStartupRuntimeRecords(startup); + const records = startup.runtimeRecords ?? []; const firstMessageStartedIndex = records.findIndex( (event) => event.event === "message_started", ); diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index cee7f10..ca7795a 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -727,8 +727,8 @@ export function AssistantPage(props: AssistantPageProps) { graph.settings?.turnConfig ?? assistant.turnConfig, ), }; - // 编辑器会把兼容字段和默认值写回 graph。状态与保存基线必须使用 - // 同一个规范化结果,否则页面刚打开就会被误判为有未保存改动。 + // 状态与保存基线使用同一个完整 graph,避免页面刚打开就被误判 + // 为有未保存改动。 const normalizedGraph = workflowGraphWithSettings(graph, wfSettings); setWorkflowName(assistant.name); setWorkflowGraph(normalizedGraph); diff --git a/frontend/src/components/workflow/ConditionEdge.tsx b/frontend/src/components/workflow/ConditionEdge.tsx index f088d26..5001836 100644 --- a/frontend/src/components/workflow/ConditionEdge.tsx +++ b/frontend/src/components/workflow/ConditionEdge.tsx @@ -1,7 +1,7 @@ "use client"; /** - * Workflow v3 edge: LLM judgement, variable expression, or deterministic default. + * Workflow v4 edge: LLM judgement, variable expression, or deterministic default. */ import { diff --git a/frontend/src/components/workflow/GenericNode.tsx b/frontend/src/components/workflow/GenericNode.tsx index 5cb4bd3..333e62b 100644 --- a/frontend/src/components/workflow/GenericNode.tsx +++ b/frontend/src/components/workflow/GenericNode.tsx @@ -18,7 +18,6 @@ import { } from "./context"; import { accentVar, - messageCompletionPolicy, type WorkflowNodeData, } from "./specs"; @@ -31,7 +30,7 @@ export function GenericNode({ id, type, data, selected }: NodeProps) { if (!spec) return null; const nodeData = data as WorkflowNodeData; - const messagePolicy = messageCompletionPolicy(nodeData); + const messagePolicy = nodeData.completionPolicy ?? "playback"; const Icon = spec.icon; const preview = ( nodeData.prompt || diff --git a/frontend/src/components/workflow/panels/ActionNodePanel.tsx b/frontend/src/components/workflow/panels/ActionNodePanel.tsx index 83a1c87..c319713 100644 --- a/frontend/src/components/workflow/panels/ActionNodePanel.tsx +++ b/frontend/src/components/workflow/panels/ActionNodePanel.tsx @@ -6,10 +6,7 @@ import { SectionCard } from "@/components/editor/section-card"; import { ToolBindingEditor } from "@/components/tools/tool-binding-editor"; import { NodeSelect } from "./controls"; -import { - actionUserInputPolicy, - type WorkflowNodeData, -} from "../specs"; +import { type WorkflowNodeData } from "../specs"; import type { ModelOption } from "../types"; import type { DynamicVariableDefinition } from "@/lib/api"; @@ -28,7 +25,7 @@ export function ActionNodePanel({ dynamicVariableDefinitions, runtimeMode, }: ActionNodePanelProps) { - const userInputPolicy = actionUserInputPolicy(draft); + const userInputPolicy = draft.userInputPolicy ?? "queue"; return ( diff --git a/frontend/src/components/workflow/panels/MessageNodePanel.tsx b/frontend/src/components/workflow/panels/MessageNodePanel.tsx index 37deef7..a88becf 100644 --- a/frontend/src/components/workflow/panels/MessageNodePanel.tsx +++ b/frontend/src/components/workflow/panels/MessageNodePanel.tsx @@ -14,7 +14,6 @@ import { import { Textarea } from "@/components/ui/textarea"; import { - messageCompletionPolicy, type MessageCompletionPolicy, type WorkflowNodeData, } from "../specs"; @@ -51,17 +50,13 @@ export function MessageNodePanel({ set: (key: string, value: unknown) => void; setPatch: (patch: Partial) => void; }) { - const completionPolicy = messageCompletionPolicy(draft); + const completionPolicy = draft.completionPolicy ?? "playback"; const completionDescription = COMPLETION_OPTIONS.find( (option) => option.value === completionPolicy, )?.description; const selectCompletionPolicy = (value: MessageCompletionPolicy) => { - setPatch({ - completionPolicy: value, - requireConfirmation: undefined, - showMessage: undefined, - }); + setPatch({ completionPolicy: value }); }; return ( diff --git a/frontend/src/components/workflow/specs.ts b/frontend/src/components/workflow/specs.ts index 068fc60..c28fb72 100644 --- a/frontend/src/components/workflow/specs.ts +++ b/frontend/src/components/workflow/specs.ts @@ -1,4 +1,4 @@ -/** Workflow v3 public graph types and defaults. */ +/** Current Workflow graph types and defaults. */ import * as LucideIcons from "lucide-react"; import { Circle, type LucideIcon } from "lucide-react"; @@ -68,13 +68,9 @@ export type WorkflowNodeData = { updates?: StateUpdateRule[]; userInputPolicy?: ActionUserInputPolicy; speech?: string; - /** Legacy field read only while an older graph is open in the editor. */ - showMessage?: boolean; title?: string; confirmLabel?: string; completionPolicy?: MessageCompletionPolicy; - /** Legacy field read only while an older graph is open in the editor. */ - requireConfirmation?: boolean; targetType?: "ai" | "human" | "queue" | "phone"; target?: string; message?: string; @@ -82,27 +78,6 @@ export type WorkflowNodeData = { [key: string]: unknown; }; -/** Infer the new Message policy for graphs saved before it was explicit. */ -export function messageCompletionPolicy( - data: WorkflowNodeData, -): MessageCompletionPolicy { - if ( - data.completionPolicy === "interruptible" || - data.completionPolicy === "playback" || - data.completionPolicy === "confirmation" - ) { - return data.completionPolicy; - } - return data.requireConfirmation === true ? "confirmation" : "playback"; -} - -/** Older Action nodes allowed input to continue, which matches queue. */ -export function actionUserInputPolicy( - data: WorkflowNodeData, -): ActionUserInputPolicy { - return data.userInputPolicy === "block" ? "block" : "queue"; -} - export type ExpressionRule = { variable: string; operator: ExpressionOperator; diff --git a/frontend/src/hooks/use-voice-preview.ts b/frontend/src/hooks/use-voice-preview.ts index 70b4db9..d13688f 100644 --- a/frontend/src/hooks/use-voice-preview.ts +++ b/frontend/src/hooks/use-voice-preview.ts @@ -827,8 +827,7 @@ export function useVoicePreview( void refreshDevices(); // SmallWebRTCTransport discards APIRequest.requestData when endpoint is a - // Request object. Keep capabilities in a header as the reliable path; - // requestData remains below for compatibility with other transports. + // Request object, so client capabilities use the request header only. const clientToolNames = Array.from(clientToolHandlersRef.current.keys()); const request = new Request(`${API_BASE}/api/webrtc/offer`, { method: "POST", @@ -849,7 +848,6 @@ export function useVoicePreview( assistant_id: assistantId, vision_enabled: Boolean(options.visionEnabled), dynamic_variables: options.dynamicVariables ?? {}, - client_tool_names: clientToolNames, }, }, });