From 731a372df94fa1359a3ab09886dd9dc5692454f8 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Tue, 4 Aug 2026 15:27:39 +0800 Subject: [PATCH] feat: add system tools and state updates --- backend/db/models.py | 2 + backend/db/session.py | 5 +- ...0260804_0010_add_assistant_system_tools.py | 32 ++ .../20260804_0011_tool_end_call_to_system.py | 44 ++ backend/models.py | 2 + backend/routes/assistants.py | 31 ++ backend/schemas.py | 32 +- backend/services/brains/prompt_brain.py | 180 +++++- backend/services/brains/workflow_brain.py | 238 +++++++- backend/services/config_resolver.py | 1 + backend/services/node_specs.py | 70 ++- backend/services/system_tools.py | 54 ++ backend/services/workflow_engine.py | 13 + backend/tests/test_brains.py | 527 +++++++++++++++++- backend/tests/test_workflow_v3.py | 91 +++ docs/features.mdx | 3 +- .../assistant-editor/dynamic-variables.tsx | 31 +- .../assistant-editor/editor-controls.tsx | 224 ++++++-- .../assistant-editor/prompt-editor.tsx | 8 + .../src/components/assistant-editor/types.ts | 2 + .../assistant-editor/workflow-page.tsx | 6 + .../src/components/pages/AssistantPage.tsx | 8 +- .../components/pages/ComponentsToolsPage.tsx | 39 +- .../src/components/workflow/GenericNode.tsx | 2 + .../components/workflow/WorkflowCanvas.tsx | 6 + .../workflow/panels/AgentNodePanel.tsx | 99 +++- .../workflow/panels/NodeSettingsPanel.tsx | 27 + .../workflow/panels/UpdateStateNodePanel.tsx | 79 +++ frontend/src/components/workflow/specs.ts | 8 +- frontend/src/components/workflow/types.ts | 1 + frontend/src/hooks/use-voice-preview.ts | 2 + frontend/src/lib/api.ts | 16 +- frontend/src/lib/system-tools.ts | 28 + 33 files changed, 1787 insertions(+), 124 deletions(-) create mode 100644 backend/migrations/versions/20260804_0010_add_assistant_system_tools.py create mode 100644 backend/migrations/versions/20260804_0011_tool_end_call_to_system.py create mode 100644 backend/services/system_tools.py create mode 100644 frontend/src/components/workflow/panels/UpdateStateNodePanel.tsx create mode 100644 frontend/src/lib/system-tools.ts diff --git a/backend/db/models.py b/backend/db/models.py index f12fb63..da4a8b7 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -154,6 +154,8 @@ class Assistant(Base): enable_interrupt: Mapped[bool] = mapped_column(Boolean, default=True) turn_config: Mapped[dict] = mapped_column(JSON, default=dict) startup: Mapped[dict] = mapped_column(JSON, default=dict) + # Prompt 助手级系统工具；Workflow 的权限保存在各 Agent 节点中。 + system_tools: Mapped[list] = mapped_column(JSON, default=list) vision_enabled: Mapped[bool] = mapped_column(Boolean, default=False) vision_model_resource_id: Mapped[str | None] = mapped_column( String(40), diff --git a/backend/db/session.py b/backend/db/session.py index 58ff9eb..2887134 100644 --- a/backend/db/session.py +++ b/backend/db/session.py @@ -54,12 +54,13 @@ async def sync_default_tools() -> None: "id": "tool_end_call_default", "name": "结束对话", "function_name": "end_call", - "type": "end_call", + "type": "system", "description": "当用户明确要求结束对话，或任务已完成时调用。", "definition": { "schema_version": 1, - "type": "end_call", + "type": "system", "config": { + "kind": "end_conversation", "message_type": "none", "custom_message": "", "capture_reason": True, diff --git a/backend/migrations/versions/20260804_0010_add_assistant_system_tools.py b/backend/migrations/versions/20260804_0010_add_assistant_system_tools.py new file mode 100644 index 0000000..db348b9 --- /dev/null +++ b/backend/migrations/versions/20260804_0010_add_assistant_system_tools.py @@ -0,0 +1,32 @@ +"""add assistant system tools + +Revision ID: 20260804_0010 +Revises: 20260801_0009 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260804_0010" +down_revision: str | Sequence[str] | None = "20260801_0009" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "assistants", + sa.Column( + "system_tools", + sa.JSON(), + server_default=sa.text("'[]'"), + nullable=False, + ), + ) + + +def downgrade() -> None: + op.drop_column("assistants", "system_tools") diff --git a/backend/migrations/versions/20260804_0011_tool_end_call_to_system.py b/backend/migrations/versions/20260804_0011_tool_end_call_to_system.py new file mode 100644 index 0000000..5d6ba2e --- /dev/null +++ b/backend/migrations/versions/20260804_0011_tool_end_call_to_system.py @@ -0,0 +1,44 @@ +"""move end conversation tools into the system category + +Revision ID: 20260804_0011 +Revises: 20260804_0010 +""" + +from collections.abc import Sequence + +from alembic import op + + +revision: str = "20260804_0011" +down_revision: str | Sequence[str] | None = "20260804_0010" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + """ + UPDATE tools + SET type = 'system', + definition = jsonb_set( + jsonb_set(definition, '{type}', '"system"'::jsonb), + '{config,kind}', + '"end_conversation"'::jsonb, + true + ) + WHERE type = 'end_call' + """ + ) + + +def downgrade() -> None: + op.execute( + """ + UPDATE tools + SET type = 'end_call', + definition = (definition #- '{config,kind}') + || '{"type":"end_call"}'::jsonb + WHERE type = 'system' + AND definition #>> '{config,kind}' = 'end_conversation' + """ + ) diff --git a/backend/models.py b/backend/models.py index c71f6da..538f62c 100644 --- a/backend/models.py +++ b/backend/models.py @@ -114,6 +114,8 @@ class AssistantConfig(BaseModel): # every item in ``tools``. tools: list[RuntimeTool] = Field(default_factory=list) llm_tool_ids: list[str] | None = None + # 助手级系统工具仅供 Prompt Pipeline；Workflow 在 Agent 节点中配置。 + system_tools: list[str] = Field(default_factory=list) knowledge_base_id: str | None = None knowledge_base_name: str = "" knowledge_base_description: str = "" diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index b80c9cd..e61b53f 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -36,6 +36,35 @@ def _validate_workflow(body: AssistantUpsert) -> None: return body.graph = normalize_graph(body.graph or {}) errors = validate_graph(body.graph) + declared_variables = set(body.dynamic_variable_definitions) + for node in body.graph.get("nodes") or []: + node_id = str(node.get("id") or "") + node_type = node.get("type") + data = node.get("data") or {} + if node_type == "update_state": + unknown = sorted( + set((data.get("assignments") or {}).keys()) - declared_variables + ) + if unknown: + errors.append( + f"Update State 节点 {node_id} 引用了未声明变量:" + + ",".join(unknown) + ) + elif node_type == "agent": + authorized = set(data.get("stateVariableNames") or []) + unknown = sorted(authorized - declared_variables) + if unknown: + errors.append( + f"Agent 节点 {node_id} 授权了未声明变量:" + + ",".join(unknown) + ) + if ( + "update_state" in (data.get("systemTools") or []) + and not authorized + ): + errors.append( + f"Agent 节点 {node_id} 启用更新状态工具时必须授权至少一个变量" + ) if errors: raise HTTPException(400, "工作流校验失败:" + ";".join(errors)) # Graph settings are the source of truth. The flat flag is only a session @@ -283,6 +312,7 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut: enable_interrupt=assistant.enable_interrupt, turn_config=assistant.turn_config or {}, startup=assistant.startup or {}, + system_tools=assistant.system_tools or [], vision_enabled=assistant.vision_enabled, vision_model_resource_id=assistant.vision_model_resource_id, model_resource_ids=await _resource_ids(session, assistant.id), @@ -359,6 +389,7 @@ async def duplicate_assistant( enable_interrupt=source.enable_interrupt, turn_config=dict(source.turn_config or {}), startup=dict(source.startup or {}), + system_tools=list(source.system_tools or []), vision_enabled=source.vision_enabled, vision_model_resource_id=source.vision_model_resource_id, knowledge_base_id=source.knowledge_base_id, diff --git a/backend/schemas.py b/backend/schemas.py index 99e8f84..3918295 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -19,7 +19,7 @@ ModelType = Literal["LLM", "ASR", "TTS", "Realtime", "Embedding", "Agent"] AssistantType = Literal["prompt", "workflow", "dify", "fastgpt", "opencode"] TurnEndStrategy = Literal["silence", "smart_turn"] KnowledgeRetrievalMode = Literal["automatic", "on_demand"] -ToolType = Literal["end_call", "http", "mcp", "client"] +ToolType = Literal["system", "http", "mcp", "client"] ToolStatus = Literal["active", "archived", "draft"] McpServerStatus = Literal["active", "archived", "draft"] McpTransport = Literal["streamable_http", "sse"] @@ -30,6 +30,12 @@ ClientToolResponseWaitMode = Literal["timeout", "session"] DynamicVariableType = Literal["string", "number", "boolean"] PromptEntryMode = Literal["wait_user", "generate"] PromptOpeningMode = Literal["interruptible", "playback", "confirmation"] +SystemToolKind = Literal[ + "end_conversation", + "update_state", + "skip_turn", + "request_human_handoff", +] # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵 EXTERNAL_TYPES = {"dify", "fastgpt", "opencode"} @@ -153,6 +159,8 @@ class AssistantUpsert(CamelModel): startup: StartupConfig = Field(default_factory=StartupConfig) vision_enabled: bool = False vision_model_resource_id: str | None = None + # 内置系统工具开关(仅 prompt 类型可用,类似 ElevenLabs Agent 的系统工具)。 + system_tools: list[SystemToolKind] = Field(default_factory=list, max_length=4) model_resource_ids: dict[ModelType, str] = Field(default_factory=dict) knowledge_base_id: str | None = None @@ -191,6 +199,8 @@ class AssistantUpsert(CamelModel): for field in ("prompt", "api_url", "api_key", "app_id"): if field not in allowed: setattr(self, field, "") + if self.type != "prompt": + self.system_tools = [] if "graph" not in allowed: self.graph = {} if self.type == "workflow": @@ -210,6 +220,15 @@ class AssistantUpsert(CamelModel): # 外部托管大脑只能 cascade,拦住不兼容的 realtime if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES: raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式") + if self.type == "prompt": + self.system_tools = list(dict.fromkeys(self.system_tools)) + if self.runtime_mode == "realtime" and self.system_tools: + raise ValueError("Prompt Realtime 模式暂不支持系统工具") + if ( + "update_state" in self.system_tools + and not self.dynamic_variable_definitions + ): + raise ValueError("启用更新状态工具前必须声明至少一个动态变量") return self @@ -250,7 +269,8 @@ class ToolParameter(CamelModel): required: bool = True -class EndCallToolConfig(CamelModel): +class SystemToolConfig(CamelModel): + kind: Literal["end_conversation"] = "end_conversation" message_type: Literal["none", "custom"] = "none" custom_message: str = "" capture_reason: bool = True @@ -275,10 +295,10 @@ class HttpToolConfig(CamelModel): return value -class EndCallToolDefinition(CamelModel): +class SystemToolDefinition(CamelModel): schema_version: int = 1 - type: Literal["end_call"] = "end_call" - config: EndCallToolConfig = Field(default_factory=EndCallToolConfig) + type: Literal["system"] = "system" + config: SystemToolConfig = Field(default_factory=SystemToolConfig) class HttpToolDefinition(CamelModel): @@ -326,7 +346,7 @@ class McpToolDefinition(CamelModel): ToolDefinition = Annotated[ Union[ - EndCallToolDefinition, + SystemToolDefinition, HttpToolDefinition, ClientToolDefinition, McpToolDefinition, diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index b5b18cd..f05fa33 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -43,14 +43,14 @@ from services.message_stage import ( MessageStageRunner, MessageStageSpec, ) -from services.runtime_variables import DynamicVariableStore +from services.runtime_variables import DynamicVariableError, DynamicVariableStore +from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool PREFLIGHT_TIMEOUT_SECONDS = 30 - class PromptBrain(BaseBrain): spec = BrainSpec( type="prompt", @@ -123,22 +123,37 @@ class PromptBrain(BaseBrain): set(cfg.llm_tool_ids) if cfg.llm_tool_ids is not None else None ) schemas: list[FunctionSchema] = [] + registered_names: set[str] = set() for tool in cfg.tools: if llm_tool_ids is not None and tool.id not in llm_tool_ids: continue - if tool.type == "end_call": + if tool.type == "system": schema, handler = self._make_end_call_tool(tool, runtime) elif tool.type in {"http", "mcp", "client"}: schema, handler = self._make_remote_tool(tool, runtime) else: continue schemas.append(schema) + registered_names.add(schema.name) policy = policy_for_tool(tool) runtime.llm.register_function( tool.function_name, handler, cancel_on_interruption=policy.cancel_on_interruption, ) + for kind in cfg.system_tools or []: + if kind not in SYSTEM_TOOL_KINDS: + logger.warning(f"忽略未知系统工具: {kind}") + continue + schema, handler = self._make_system_tool(kind, runtime) + if schema.name in registered_names: + logger.warning( + f"跳过系统工具 {schema.name}: 与已绑定工具函数名冲突" + ) + continue + registered_names.add(schema.name) + schemas.append(schema) + runtime.llm.register_function(schema.name, handler) runtime.set_tools(schemas) async def run_preflight(self) -> None: @@ -582,3 +597,162 @@ class PromptBrain(BaseBrain): required=["reason"] if capture_reason else [], ) return schema, end_call + + # ---------- 内置系统工具(助手配置 system_tools 开关) ---------- + + def _make_system_tool(self, kind: str, runtime: BrainRuntime): + if kind == "end_conversation": + return self._make_end_conversation_tool(runtime) + if kind == "update_state": + return self._make_update_state_tool(runtime) + if kind == "skip_turn": + return self._make_skip_turn_tool() + if kind == "request_human_handoff": + return self._make_handoff_tool(runtime) + raise ValueError(f"未知系统工具: {kind}") + + def _make_end_conversation_tool(self, runtime: BrainRuntime): + """结束本次对话,等待模型已生成的告别语播完后再挂断。""" + + async def end_conversation(params: FunctionCallParams) -> None: + reason = str( + params.arguments.get("reason") or "end_conversation" + ).strip() + self._waiting_for_generated_end_speech = True + runtime.call_end.begin(reason) + await params.result_callback( + {"status": "success", "action": "ending_call"}, + properties=FunctionCallResultProperties(run_llm=False), + ) + + schema = FunctionSchema( + name="end_conversation", + description=( + "礼貌地结束本次对话。当用户明确告别、表示任务已完成" + "或要求挂断时调用。" + ), + properties={ + "reason": { + "type": "string", + "description": "结束对话的简短原因。", + } + }, + required=[], + ) + return schema, end_conversation + + def _make_update_state_tool(self, runtime: BrainRuntime): + """更新已声明的动态变量(会话状态),并让模型继续当前回答。""" + + writable = state_update_properties(self._cfg.dynamic_variable_definitions) + + async def update_state(params: FunctionCallParams) -> None: + state = dict(params.arguments or {}) + try: + changed = self._store.assign_declared_many(state) + except DynamicVariableError as exc: + await params.result_callback( + {"status": "error", "message": f"状态更新失败: {exc}"} + ) + return + if changed: + self._refresh_prompt() + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "session-variables", + "reason": "update_state", + "variables": self._store.public_values(), + "changed": changed, + } + ) + ) + await params.result_callback( + { + "status": "success", + "changed": changed, + "variables": self._store.public_values(), + } + ) + + schema = FunctionSchema( + name="update_state", + description=( + "静默更新本次对话中已经声明并明确列出的动态变量。" + "只提交本轮获得或确认的信息，更新后继续当前回答。" + ), + properties=writable, + required=[], + ) + return schema, update_state + + def _make_skip_turn_tool(self): + """跳过当前轮次,不生成任何语音回复。""" + + async def skip_turn(params: FunctionCallParams) -> None: + reason = str(params.arguments.get("reason") or "").strip() + result = {"status": "success", "action": "skip_turn"} + if reason: + result["reason"] = reason + await params.result_callback( + result, + properties=FunctionCallResultProperties(run_llm=False), + ) + + schema = FunctionSchema( + name="skip_turn", + description=( + "跳过当前轮次,不生成任何语音回复。" + "仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。" + ), + properties={ + "reason": { + "type": "string", + "description": "跳过本轮的原因(可选)。", + } + }, + required=[], + ) + return schema, skip_turn + + def _make_handoff_tool(self, runtime: BrainRuntime): + """提交人工接管请求；请求完成前保持当前 AI 会话可用。""" + + async def request_human_handoff(params: FunctionCallParams) -> None: + reason = str( + params.arguments.get("reason") or "human_handoff" + ).strip() + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "handoff-requested", + "source": "prompt-system-tool", + "reason": reason, + "message": "用户请求转接人工服务。", + } + ) + ) + await params.result_callback( + { + "status": "requested", + "action": "human_handoff_requested", + "message": "人工接管请求已提交，请告知用户正在等待人工响应。", + } + ) + + schema = FunctionSchema( + name="request_human_handoff", + description=( + "提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法" + "解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续" + "回复用户并说明正在等待人工响应。" + ), + properties={ + "reason": { + "type": "string", + "description": "转接人工的原因。", + } + }, + required=[], + ) + return schema, request_human_handoff diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 46dc6a3..28c2ff6 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -56,7 +56,8 @@ from services.message_stage import ( MessageStageRunner, MessageStageSpec, ) -from services.runtime_variables import DynamicVariableStore +from services.runtime_variables import DynamicVariableError, DynamicVariableStore +from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool from services.workflow.agent import WorkflowAgentStage @@ -184,6 +185,7 @@ class WorkflowBrain(BaseBrain): self._output: WorkflowOutput | None = None self._agent_stage: WorkflowAgentStage | None = None self._ended = False + self._waiting_for_generated_end_speech = False self._next_message_token = 1 self._pending_message: _MessageContinuation | None = None @@ -229,6 +231,7 @@ class WorkflowBrain(BaseBrain): runtime=runtime, ) self._ended = False + self._waiting_for_generated_end_speech = False self._next_message_token = 1 self._pending_message = None self._manager = ConfiguredFlowManager( @@ -494,18 +497,30 @@ class WorkflowBrain(BaseBrain): return None return decision.edge + async def on_assistant_text_start(self, _turn_id: str) -> None: + if self._runtime is not None: + self._runtime.call_end.begin_response() + async def on_assistant_text_end( self, _turn_id: str, content: str, interrupted: bool, ) -> None: - if not content or interrupted or self._ended: - return - self._store.record("agent", content, completed_agent_turn=True) - self._state.consume_user_turn() - if self._engine.node_type(self._state.current_node_id) == "agent": - self._state.status = WorkflowStatus.WAITING_USER + if content and not interrupted and not self._ended: + self._store.record("agent", content, completed_agent_turn=True) + self._state.consume_user_turn() + if self._engine.node_type(self._state.current_node_id) == "agent": + self._state.status = WorkflowStatus.WAITING_USER + if ( + self._waiting_for_generated_end_speech + and self._runtime is not None + and self._runtime.call_end.ending + ): + self._waiting_for_generated_end_speech = False + await self._runtime.call_end.finish_after_current_speech( + has_text=bool(content.strip()) and not interrupted + ) async def _refresh_agent_prompt(self, node_id: str) -> None: await self._require_agent_stage().refresh_prompt(node_id) @@ -528,15 +543,43 @@ class WorkflowBrain(BaseBrain): ) -> NodeConfig: stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] + registered_names: set[str] = set() + + def append_function(function: FlowsFunctionSchema | None) -> None: + if function is None: + return + function_name = getattr(function, "name", "") + if not function_name: + # Runtime-provided global functions are opaque in a few + # adapters; FlowManager remains responsible for those names. + functions.append(function) + return + if function_name in registered_names: + logger.warning( + f"跳过 Agent {node_id} 的函数 {function_name}: 函数名冲突" + ) + return + registered_names.add(function_name) + functions.append(function) + for tool_id in stage.tool_ids: tool = self._tool_by_id.get(str(tool_id)) if tool and tool.type in {"http", "mcp", "client"}: - functions.append(self._flow_tool(tool, node_id)) - knowledge_function = self._knowledge_function(node_id) - if knowledge_function: - functions.append(knowledge_function) + append_function(self._flow_tool(tool, node_id)) + append_function(self._knowledge_function(node_id)) if stage.vision_enabled and self._require_runtime().vision_function: - functions.append(self._require_runtime().vision_function) + append_function(self._require_runtime().vision_function) + for kind in stage.system_tools: + if kind not in SYSTEM_TOOL_KINDS: + logger.warning(f"忽略 Agent {node_id} 的未知系统工具: {kind}") + continue + append_function( + self._workflow_system_tool( + kind, + node_id=node_id, + state_variable_names=stage.state_variable_names, + ) + ) return self._require_agent_stage().node_config( node_id, functions=functions, @@ -693,6 +736,159 @@ class WorkflowBrain(BaseBrain): ), ) + def _workflow_system_tool( + self, + kind: str, + *, + node_id: str, + state_variable_names: tuple[str, ...], + ) -> FlowsFunctionSchema: + """Build one platform-owned tool scoped to the active Agent node.""" + if kind == "update_state": + return self._workflow_update_state_tool( + node_id, + state_variable_names=state_variable_names, + ) + if kind == "skip_turn": + return self._workflow_skip_turn_tool() + if kind == "request_human_handoff": + return self._workflow_handoff_tool(node_id) + if kind == "end_conversation": + return self._workflow_end_conversation_tool() + raise ValueError(f"未知系统工具: {kind}") + + def _workflow_update_state_tool( + self, + node_id: str, + *, + state_variable_names: tuple[str, ...], + ) -> FlowsFunctionSchema: + allowed = frozenset(state_variable_names) + + async def handler(args, _flow_manager): + values = dict(args or {}) + unauthorized = sorted(set(values) - allowed) + if unauthorized: + return { + "status": "error", + "message": "状态变量未获当前节点授权: " + ",".join(unauthorized), + } + try: + changed = self._store.assign_declared_many(values) + except DynamicVariableError as exc: + return {"status": "error", "message": f"状态更新失败: {exc}"} + if changed: + await self._emit_variables( + reason="update_state", + node_id=node_id, + changed=changed, + ) + await self._refresh_agent_prompt(node_id) + return { + "status": "success", + "changed": changed, + "variables": self._store.public_values(), + } + + return FlowsFunctionSchema( + name="update_state", + description=( + "静默更新当前阶段明确授权的动态变量。" + "只提交本轮获得或确认的信息，更新后继续当前回答。" + ), + properties=state_update_properties( + self._cfg.dynamic_variable_definitions if self._cfg else {}, + allowed_names=state_variable_names, + ), + required=[], + handler=handler, + ) + + @staticmethod + def _workflow_skip_turn_tool() -> FlowsFunctionSchema: + async def handler(args, _flow_manager): + reason = str((args or {}).get("reason") or "").strip() + result = {"status": "success", "action": "skip_turn"} + if reason: + result["reason"] = reason + return result + + setattr(handler, "_suppress_followup_llm", True) + return FlowsFunctionSchema( + name="skip_turn", + description=( + "跳过当前轮次，不生成任何语音回复。" + "仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。" + ), + properties={ + "reason": { + "type": "string", + "description": "跳过本轮的原因（可选）。", + } + }, + required=[], + handler=handler, + ) + + def _workflow_handoff_tool(self, node_id: str) -> FlowsFunctionSchema: + async def handler(args, _flow_manager): + reason = str((args or {}).get("reason") or "human_handoff").strip() + await self._require_runtime().queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "handoff-requested", + "source": "workflow-system-tool", + "nodeId": node_id, + "reason": reason, + "message": "用户请求转接人工服务。", + } + ) + ) + self._store.values["system__handoff_status"] = "requested" + return { + "status": "requested", + "action": "human_handoff_requested", + "message": "人工接管请求已提交，请告知用户正在等待人工响应。", + } + + return FlowsFunctionSchema( + name="request_human_handoff", + description=( + "提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法" + "解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续" + "回复用户并说明正在等待人工响应。" + ), + properties={ + "reason": {"type": "string", "description": "转接人工的原因。"} + }, + required=[], + handler=handler, + ) + + def _workflow_end_conversation_tool(self) -> FlowsFunctionSchema: + async def handler(args, _flow_manager): + reason = str((args or {}).get("reason") or "end_conversation").strip() + self._waiting_for_generated_end_speech = True + self._require_runtime().call_end.begin(reason) + return {"status": "success", "action": "ending_call"} + + setattr(handler, "_suppress_followup_llm", True) + return FlowsFunctionSchema( + name="end_conversation", + description=( + "礼貌地结束本次对话。当用户明确告别、表示任务已完成" + "或要求挂断时调用。" + ), + properties={ + "reason": { + "type": "string", + "description": "结束对话的简短原因。", + } + }, + required=[], + handler=handler, + ) + def _flow_managed_transition_config( self, node_config: NodeConfig, @@ -838,6 +1034,8 @@ 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 == "update_state": + await self._enter_update_state(node_id) elif node_type == "message": self._prepare_message_continuation( node_id, @@ -862,6 +1060,22 @@ class WorkflowBrain(BaseBrain): node_id = str(edge.get("target") or "") raise RuntimeError("工作流连续自动跳转超过安全上限") + async def _enter_update_state(self, node_id: str) -> None: + """Apply one deterministic, atomic dynamic-variable update.""" + self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) + await self._emit_node_active(node_id) + raw_assignments = self._engine.data(node_id).get("assignments") or {} + rendered = self._store.render_data(raw_assignments) + if not isinstance(rendered, dict): + raise DynamicVariableError("Update State 节点赋值必须是对象") + changed = self._store.assign_declared_many(rendered) + if changed: + await self._emit_variables( + reason="update_state", + node_id=node_id, + changed=changed, + ) + def _prepare_message_continuation( self, node_id: str, diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index c551ca2..63a1529 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -266,6 +266,7 @@ async def resolve_runtime_config( enableInterrupt=assistant.enable_interrupt, turnConfig=assistant.turn_config or {}, startup=assistant.startup or {}, + system_tools=assistant.system_tools or [], tools=runtime_tools, llm_tool_ids=llm_tool_ids, knowledge_base_id=assistant.knowledge_base_id, diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index d12a02b..7e9fe2d 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -11,15 +11,30 @@ from services.message_policy import ( MESSAGE_CONFIRMATION, MESSAGE_PLAYBACK, ) +from services.system_tools import SYSTEM_TOOL_KINDS, normalize_system_tools SPEC_VERSION = "3" -NODE_TYPES = {"start", "agent", "message", "action", "handoff", "end"} +NODE_TYPES = { + "start", + "agent", + "message", + "action", + "update_state", + "handoff", + "end", +} EDGE_MODES = {"llm", "expression", "always"} AGENT_ENTRY_MODES = {"wait_user", "generate"} ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"} ACTION_USER_INPUT_POLICIES = {"queue", "block"} -AUTOMATIC_NODE_TYPES = {"start", "message", "action", "handoff"} +AUTOMATIC_NODE_TYPES = { + "start", + "message", + "action", + "update_state", + "handoff", +} EXPRESSION_OPERATORS = { "eq", "neq", @@ -98,6 +113,24 @@ NODE_SPECS: list[dict[str, Any]] = [ {"key": "name", "label": "节点名称", "type": "text", "default": "Action"}, ], }, + { + "name": "update_state", + "displayName": "Update State", + "category": "execution_node", + "description": "原子更新已声明的动态变量，然后使用新状态继续路由。", + "icon": "Braces", + "accent": "mint", + "addable": True, + "constraints": {"minIncoming": 1, "minOutgoing": 0}, + "fields": [ + { + "key": "name", + "label": "节点名称", + "type": "text", + "default": "Update State", + }, + ], + }, { "name": "handoff", "displayName": "Handoff", @@ -160,6 +193,11 @@ def _normalize_agent_data(data: dict[str, Any]) -> None: if data.get("entryMode") not in AGENT_ENTRY_MODES: data["entryMode"] = "wait_user" data.pop("entrySpeech", None) + data["systemTools"] = list(normalize_system_tools(data.get("systemTools"))) + state_names = data.get("stateVariableNames") + data["stateVariableNames"] = list( + dict.fromkeys(str(name) for name in state_names or [] if str(name)) + ) if "inheritGlobalConfig" not in data: has_node_overrides = any( ( @@ -194,6 +232,12 @@ def _normalize_action_data(data: dict[str, Any]) -> None: data.pop("speech", None) +def _normalize_update_state_data(data: dict[str, Any]) -> None: + """Keep deterministic state updates as one explicit assignment object.""" + assignments = data.get("assignments") + data["assignments"] = dict(assignments) if isinstance(assignments, dict) else {} + + def _normalize_message_data(data: dict[str, Any]) -> None: """Fill the small built-in Message contract used by runtime and editor.""" data.setdefault("speech", "") @@ -246,6 +290,8 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: _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 [] @@ -262,6 +308,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: "agent": "agent", "message": "message", "action": "action", + "update_state": "update_state", "handoff": "handoff", "end": "end", } @@ -283,6 +330,8 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: _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) @@ -390,6 +439,16 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: entry_mode = data.get("entryMode", "wait_user") if entry_mode not in AGENT_ENTRY_MODES: errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}") + system_tools = data.get("systemTools", []) + if not isinstance(system_tools, list) or any( + tool not in SYSTEM_TOOL_KINDS for tool in system_tools + ): + errors.append(f"Agent 节点 {node_id} 的系统工具配置无效") + state_names = data.get("stateVariableNames", []) + if not isinstance(state_names, list) or any( + not isinstance(name, str) for name in state_names + ): + errors.append(f"Agent 节点 {node_id} 的状态变量授权必须是列表") elif node_type == "message": data = node.get("data") or {} speech = data.get("speech") @@ -447,6 +506,13 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: errors.append( f"Action 节点 {node_id} 的用户输入策略无效:{input_policy}" ) + elif node_type == "update_state": + data = node.get("data") or {} + assignments = data.get("assignments") + if not isinstance(assignments, dict) or not assignments: + errors.append( + f"Update State 节点 {node_id} 必须配置至少一个变量赋值" + ) if counts["start"] != 1: errors.append("工作流必须有且仅有一个 Start 节点") diff --git a/backend/services/system_tools.py b/backend/services/system_tools.py new file mode 100644 index 0000000..106e35c --- /dev/null +++ b/backend/services/system_tools.py @@ -0,0 +1,54 @@ +"""Shared contracts for platform-owned conversation system tools.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from typing import Any + + +SYSTEM_TOOL_KINDS = frozenset( + { + "end_conversation", + "update_state", + "skip_turn", + "request_human_handoff", + } +) + + +def normalize_system_tools(values: Iterable[Any] | None) -> tuple[str, ...]: + """Return known tool names once each while preserving editor order.""" + return tuple( + dict.fromkeys( + str(value) + for value in values or () + if str(value) in SYSTEM_TOOL_KINDS + ) + ) + + +def state_update_properties( + definitions: Mapping[str, Mapping[str, Any]] | None, + *, + allowed_names: Iterable[str] | None = None, +) -> dict[str, dict[str, Any]]: + """Build an explicit LLM schema for writable declared variables.""" + definitions = definitions or {} + names = ( + list(dict.fromkeys(str(name) for name in allowed_names)) + if allowed_names is not None + else list(definitions) + ) + properties: dict[str, dict[str, Any]] = {} + for name in names: + definition = definitions.get(name) + if not isinstance(definition, Mapping): + continue + variable_type = str(definition.get("type") or "string") + if variable_type not in {"string", "number", "boolean"}: + continue + properties[name] = { + "type": variable_type, + "description": f"更新动态变量 {name}。", + } + return properties diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 238b59f..f293d3b 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -11,6 +11,7 @@ from typing import Any from services.node_specs import normalize_graph from services.runtime_variables import DynamicVariableStore +from services.system_tools import normalize_system_tools @dataclass(frozen=True) @@ -24,6 +25,8 @@ class AgentStageConfig: vision_enabled: bool vision_model_resource_id: str | None tool_ids: tuple[str, ...] + system_tools: tuple[str, ...] + state_variable_names: tuple[str, ...] knowledge_base_id: str | None knowledge_mode: str knowledge_top_n: int @@ -145,6 +148,16 @@ class WorkflowEngine: str(source.get("visionModelResourceId") or "") or None ), tool_ids=tuple(str(tool_id) for tool_id in source.get("toolIds") or []), + # System tools and their writable state scope always belong to the + # Agent node. They are permissions, not inheritable model config. + system_tools=normalize_system_tools(data.get("systemTools")), + state_variable_names=tuple( + dict.fromkeys( + str(name) + for name in data.get("stateVariableNames") or [] + if str(name) + ) + ), knowledge_base_id=knowledge_base_id or None, knowledge_mode=( str(source.get("knowledgeMode") or "automatic") diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 0a36eff..f3d8e17 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -154,6 +154,46 @@ class BrainRegistryTests(unittest.TestCase): ) self.assertIn("user_name", assistant.dynamic_variable_definitions) + def test_system_tools_are_prompt_pipeline_only(self): + assistant = AssistantUpsert( + name="prompt", + type="prompt", + systemTools=["end_conversation", "skip_turn", "end_conversation"], + ) + self.assertEqual(assistant.system_tools, ["end_conversation", "skip_turn"]) + + workflow = AssistantUpsert( + name="workflow", + type="workflow", + systemTools=["update_state"], + graph={}, + ) + self.assertEqual(workflow.system_tools, []) + + with self.assertRaises(ValueError): + AssistantUpsert( + name="realtime prompt", + type="prompt", + runtimeMode="realtime", + systemTools=["skip_turn"], + ) + + def test_system_tools_reject_unknown_kind(self): + with self.assertRaises(ValueError): + AssistantUpsert( + name="prompt", + type="prompt", + systemTools=["update_state", "magic"], + ) + + def test_prompt_update_state_requires_a_declared_variable(self): + with self.assertRaisesRegex(ValueError, "必须声明至少一个动态变量"): + AssistantUpsert( + name="prompt", + type="prompt", + systemTools=["update_state"], + ) + def test_workflow_keeps_dynamic_variables_and_tool_bindings(self): assistant = AssistantUpsert( name="workflow", @@ -757,7 +797,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(prompts, ["面板状态：true"]) self.assertEqual(queued_frames, []) - async def test_end_call_tool_is_owned_by_prompt_brain(self): + async def test_end_conversation_system_tool_is_owned_by_prompt_brain(self): brain = build_brain( AssistantConfig( type="prompt", @@ -766,9 +806,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): id="end-call", name="结束通话", function_name="end_call", - type="end_call", + type="system", definition={ "config": { + "kind": "end_conversation", "message_type": "none", "capture_reason": True, } @@ -792,8 +833,13 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): id="end-call", name="结束通话", function_name="end_call", - type="end_call", - definition={"config": {"capture_reason": True}}, + type="system", + definition={ + "config": { + "kind": "end_conversation", + "capture_reason": True, + } + }, ) ], ), @@ -819,13 +865,18 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(call_end.response_started) self.assertEqual(params.result["action"], "ending_call") - async def test_end_call_waits_for_prompt_generated_closing_speech(self): + async def test_system_end_conversation_waits_for_generated_closing_speech(self): tool = RuntimeTool( id="end-call", name="结束通话", function_name="end_call", - type="end_call", - definition={"config": {"message_type": "none"}}, + type="system", + definition={ + "config": { + "kind": "end_conversation", + "message_type": "none", + } + }, ) cfg = AssistantConfig(type="prompt", tools=[tool]) brain = build_brain(cfg) @@ -855,6 +906,279 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(call_end.armed) self.assertTrue(call_end.waited_for_text) + async def test_system_tools_register_with_fixed_names(self): + cfg = AssistantConfig( + type="prompt", + dynamic_variable_definitions={ + "user_name": { + "type": "string", + "required": False, + "default": None, + } + }, + system_tools=[ + "end_conversation", + "update_state", + "skip_turn", + "request_human_handoff", + ], + ) + brain = build_brain(cfg) + llm = FakeLLM() + visible_tools = [] + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda tools: visible_tools.extend(tools or []), + call_end=FakeCallEnd(), + ), + ) + + self.assertEqual( + [tool.name for tool in visible_tools], + [ + "end_conversation", + "update_state", + "skip_turn", + "request_human_handoff", + ], + ) + update_schema = next( + tool for tool in visible_tools if tool.name == "update_state" + ) + self.assertEqual( + update_schema.properties, + { + "user_name": { + "type": "string", + "description": "更新动态变量 user_name。", + } + }, + ) + for name in ( + "end_conversation", + "update_state", + "skip_turn", + "request_human_handoff", + ): + self.assertIn(name, llm.functions) + + async def test_end_conversation_ends_call_after_generated_speech(self): + cfg = AssistantConfig( + type="prompt", + system_tools=["end_conversation"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + call_end = FakeCallEnd() + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + ), + ) + + await brain.on_assistant_text_start("turn-1") + params = FakeFunctionParams({"reason": "用户已完成咨询"}) + await llm.functions["end_conversation"](params) + self.assertEqual(call_end.reason, "用户已完成咨询") + self.assertTrue(call_end.ending) + self.assertFalse(call_end.finished) + self.assertEqual(params.result["action"], "ending_call") + self.assertFalse(params.properties.run_llm) + + await brain.on_assistant_text_end( + "turn-1", + "好的，再见。", + False, + ) + self.assertFalse(call_end.finished) + self.assertTrue(call_end.armed) + + async def test_end_conversation_finishes_when_no_speech(self): + cfg = AssistantConfig( + type="prompt", + system_tools=["end_conversation"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + call_end = FakeCallEnd() + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + ), + ) + + await brain.on_assistant_text_start("turn-1") + await llm.functions["end_conversation"](FakeFunctionParams({})) + await brain.on_assistant_text_end("turn-1", "", False) + self.assertTrue(call_end.finished) + + async def test_update_state_updates_declared_variables(self): + cfg = AssistantConfig( + type="prompt", + prompt="当前用户名: {{user_name}}", + dynamic_variable_definitions={ + "user_name": {"type": "string", "required": False, "default": None} + }, + system_tools=["update_state"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + prompts = [] + queued_frames = [] + + async def collect(frame): + queued_frames.append(frame) + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=collect, + set_system_prompt=prompts.append, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ), + ) + + params = FakeFunctionParams({"user_name": "王小明"}) + await llm.functions["update_state"](params) + + self.assertEqual(params.result["status"], "success") + self.assertEqual(params.result["changed"], ["user_name"]) + self.assertIsNone(params.properties) + self.assertEqual( + [frame.message for frame in queued_frames], + [ + { + "type": "session-variables", + "reason": "update_state", + "variables": {"user_name": "王小明"}, + "changed": ["user_name"], + } + ], + ) + self.assertTrue(prompts[-1].endswith("当前用户名: 王小明")) + + async def test_update_state_rejects_undeclared_variable(self): + cfg = AssistantConfig( + type="prompt", + system_tools=["update_state"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ), + ) + + params = FakeFunctionParams({"unknown_key": "x"}) + await llm.functions["update_state"](params) + self.assertEqual(params.result["status"], "error") + self.assertIn("未声明", params.result["message"]) + + async def test_skip_turn_suppresses_response(self): + cfg = AssistantConfig( + type="prompt", + system_tools=["skip_turn"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ), + ) + + params = FakeFunctionParams({"reason": "无需回复"}) + await llm.functions["skip_turn"](params) + self.assertEqual(params.result["action"], "skip_turn") + self.assertEqual(params.result["reason"], "无需回复") + self.assertFalse(params.properties.run_llm) + + async def test_request_human_handoff_keeps_call_available(self): + cfg = AssistantConfig( + type="prompt", + system_tools=["request_human_handoff"], + ) + brain = build_brain(cfg) + llm = FakeLLM() + call_end = FakeCallEnd() + queued_frames = [] + + async def collect(frame): + queued_frames.append(frame) + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=collect, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + ), + ) + + await brain.on_assistant_text_start("turn-1") + params = FakeFunctionParams({"reason": "用户要求人工客服"}) + await llm.functions["request_human_handoff"](params) + + self.assertEqual( + [frame.message for frame in queued_frames], + [ + { + "type": "handoff-requested", + "source": "prompt-system-tool", + "reason": "用户要求人工客服", + "message": "用户请求转接人工服务。", + } + ], + ) + self.assertEqual(call_end.reason, "") + self.assertFalse(call_end.ending) + self.assertEqual(params.result["action"], "human_handoff_requested") + self.assertIsNone(params.properties) + + await brain.on_assistant_text_end("turn-1", "", False) + self.assertFalse(call_end.finished) + async def test_http_tool_renders_secrets_and_updates_prompt_variable(self): requests = [] @@ -962,6 +1286,195 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_agent_system_tools_are_scoped_and_update_declared_state(self): + cfg = prepare_dynamic_config( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": { + "systemTools": [ + "update_state", + "skip_turn", + "request_human_handoff", + ], + "stateVariableNames": ["customer_name"], + }, + }, + ], + "edges": [], + }, + dynamic_variable_definitions={ + "customer_name": { + "type": "string", + "required": False, + "default": None, + }, + "internal_note": { + "type": "string", + "required": False, + "default": None, + }, + }, + ), + {}, + assistant_id="asst_workflow_system_tools", + ) + brain = WorkflowBrain(cfg) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + brain._runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + brain._agent_stage = SimpleNamespace( + node_config=lambda node_id, *, functions, leading_messages: { + "name": node_id, + "functions": functions, + "leading_messages": leading_messages, + } + ) + brain._emit_variables = AsyncMock() + brain._refresh_agent_prompt = AsyncMock() + + node_config = brain._agent_config("agent") + functions = {tool.name: tool for tool in node_config["functions"]} + self.assertEqual( + list(functions), + ["update_state", "skip_turn", "request_human_handoff"], + ) + self.assertEqual( + list(functions["update_state"].properties), + ["customer_name"], + ) + + result = await functions["update_state"].handler( + {"customer_name": "李白"}, + None, + ) + self.assertEqual(result["changed"], ["customer_name"]) + self.assertEqual(brain._store.public_values(), {"customer_name": "李白"}) + brain._emit_variables.assert_awaited_once_with( + reason="update_state", + node_id="agent", + changed=["customer_name"], + ) + brain._refresh_agent_prompt.assert_awaited_once_with("agent") + + unauthorized = await functions["update_state"].handler( + {"internal_note": "不可写"}, + None, + ) + self.assertEqual(unauthorized["status"], "error") + self.assertIn("未获当前节点授权", unauthorized["message"]) + + handoff = await functions["request_human_handoff"].handler( + {"reason": "用户要求人工"}, + None, + ) + self.assertEqual(handoff["status"], "requested") + self.assertFalse(brain._runtime.call_end.ending) + event = next( + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + ) + self.assertEqual(event["type"], "handoff-requested") + self.assertEqual(event["source"], "workflow-system-tool") + + async def test_update_state_node_applies_assignments_atomically(self): + cfg = prepare_dynamic_config( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "set_state", + "type": "update_state", + "data": { + "assignments": { + "confirmed": True, + "display_name": "{{source_name}}", + } + }, + }, + ], + "edges": [], + }, + dynamic_variable_definitions={ + "source_name": {"type": "string", "default": "李白"}, + "display_name": { + "type": "string", + "required": False, + "default": None, + }, + "confirmed": {"type": "boolean", "default": False}, + }, + ), + {}, + assistant_id="asst_workflow_update_state_node", + ) + brain = WorkflowBrain(cfg) + brain._emit_node_active = AsyncMock() + brain._emit_variables = AsyncMock() + + await brain._enter_update_state("set_state") + + self.assertEqual(brain._store.public_values()["display_name"], "李白") + self.assertTrue(brain._store.public_values()["confirmed"]) + brain._emit_variables.assert_awaited_once_with( + reason="update_state", + node_id="set_state", + changed=["confirmed", "display_name"], + ) + + async def test_workflow_end_conversation_waits_for_generated_speech(self): + brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [{"id": "start", "type": "start", "data": {}}], + "edges": [], + } + ) + call_end = FakeCallEnd() + 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, + ) + tool = brain._workflow_end_conversation_tool() + + await brain.on_assistant_text_start("turn-1") + result = await tool.handler({"reason": "用户告别"}, None) + self.assertEqual(result["action"], "ending_call") + self.assertTrue(call_end.ending) + self.assertEqual(call_end.reason, "用户告别") + self.assertTrue(getattr(tool.handler, "_suppress_followup_llm")) + + await brain.on_assistant_text_end("turn-1", "再见。", False) + self.assertTrue(call_end.armed) + self.assertTrue(call_end.waited_for_text) + async def test_flow_manager_dispatches_native_vision_without_auxiliary_handler(self): manager = object.__new__(ConfiguredFlowManager) fallback_transition = AsyncMock() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 4d1b077..bc306cc 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -360,6 +360,13 @@ class WorkflowGraphTests(unittest.TestCase): def test_agent_effective_config_inherits_then_switches_to_override(self): graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["id"] == "agent") + agent["data"].update( + { + "systemTools": ["update_state", "skip_turn"], + "stateVariableNames": ["customer", "order_status"], + } + ) graph["settings"].update( { "defaultLlmResourceId": "llm_global", @@ -389,6 +396,11 @@ class WorkflowGraphTests(unittest.TestCase): ) self.assertTrue(engine.uses_vision()) self.assertEqual(inherited.tool_ids, ("tool_global",)) + self.assertEqual(inherited.system_tools, ("update_state", "skip_turn")) + self.assertEqual( + inherited.state_variable_names, + ("customer", "order_status"), + ) self.assertEqual(inherited.knowledge_mode, "on_demand") self.assertFalse(inherited.enable_interrupt) self.assertEqual( @@ -417,6 +429,7 @@ class WorkflowGraphTests(unittest.TestCase): self.assertIsNone(custom.vision_model_resource_id) self.assertFalse(engine.uses_vision()) self.assertEqual(custom.tool_ids, ("tool_agent",)) + self.assertEqual(custom.system_tools, ("update_state", "skip_turn")) self.assertEqual(custom.knowledge_mode, "disabled") self.assertTrue(custom.enable_interrupt) self.assertEqual( @@ -424,6 +437,84 @@ class WorkflowGraphTests(unittest.TestCase): "smart_turn", ) + def test_update_state_node_and_agent_state_scope_are_validated(self): + graph = valid_graph() + graph["nodes"].insert( + 1, + { + "id": "set_state", + "type": "update_state", + "data": { + "name": "记录状态", + "assignments": {"confirmed": True}, + }, + }, + ) + graph["edges"][0]["target"] = "set_state" + graph["edges"].insert( + 1, + { + "id": "after_state", + "source": "set_state", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + }, + ) + agent = next(node for node in graph["nodes"] if node["id"] == "agent") + agent["data"].update( + { + "systemTools": ["update_state"], + "stateVariableNames": ["customer"], + } + ) + body = AssistantUpsert( + name="状态工作流", + type="workflow", + dynamicVariableDefinitions={ + "confirmed": { + "type": "boolean", + "required": False, + "default": False, + }, + "customer": { + "type": "string", + "required": False, + "default": None, + }, + }, + graph=graph, + ) + + _validate_workflow(body) + normalized_state = next( + node for node in body.graph["nodes"] if node["type"] == "update_state" + ) + self.assertEqual(normalized_state["data"]["assignments"], {"confirmed": True}) + + normalized_state["data"]["assignments"] = {"missing": "x"} + with self.assertRaisesRegex(HTTPException, "未声明变量:missing"): + _validate_workflow(body) + + def test_agent_update_state_tool_requires_an_authorized_variable(self): + graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["id"] == "agent") + agent["data"]["systemTools"] = ["update_state"] + body = AssistantUpsert( + name="缺少授权", + type="workflow", + dynamicVariableDefinitions={ + "customer": { + "type": "string", + "required": False, + "default": None, + } + }, + graph=graph, + ) + + with self.assertRaisesRegex(HTTPException, "必须授权至少一个变量"): + _validate_workflow(body) + def test_vision_resource_creates_isolated_runtime_config(self): base = AssistantConfig(type="workflow", model="text-only") resource = RuntimeModelResource( diff --git a/docs/features.mdx b/docs/features.mdx index f975c27..f73b360 100644 --- a/docs/features.mdx +++ b/docs/features.mdx @@ -26,6 +26,7 @@ icon: bot ## 工具与历史 -- 可复用工具支持结束通话和 HTTP 请求两种类型。 +- 可复用工具分为 System、HTTP、Client 和 MCP；结束对话属于 System 类别。 - HTTP 工具的 URL、请求头、参数和密钥独立管理。 +- Prompt 助手和 Workflow Agent 节点可开启内置系统工具：结束对话、更新状态、跳过本轮、请求人工接管；Workflow 还提供确定性的 Update State 节点。 - 历史记录保留会话、通道、运行模式及消息序列，便于排查对话。 diff --git a/frontend/src/components/assistant-editor/dynamic-variables.tsx b/frontend/src/components/assistant-editor/dynamic-variables.tsx index 3800397..a20fd01 100644 --- a/frontend/src/components/assistant-editor/dynamic-variables.tsx +++ b/frontend/src/components/assistant-editor/dynamic-variables.tsx @@ -3,7 +3,6 @@ import { Check, Copy } from "lucide-react"; import { useState } from "react"; -import { Badge } from "@/components/ui/badge"; import { Input } from "@/components/ui/input"; import { Select, @@ -252,12 +251,18 @@ function DynamicVariablesEditor({ {`{{${name}}}`} - - 已引用 - +
+ + 会话开始时必须提供 + + + updateDefinition(name, { required }) + } + /> +
- setDraftIds((current) => - checked - ? current.filter((id) => id !== tool.id) - : [...current, tool.id], - ) - } - className="size-4 accent-primary" - /> -
-
- - {tool.name} - - - {tool.type === "end_call" - ? "End Call" - : tool.type === "mcp" - ? "MCP" - : tool.type === "client" - ? "Client" - : "HTTP"} - -
-
- {tool.functionName} -
+ setActiveTab(value as Tool["type"])} + > + + {tabs.map((tab) => ( + + {tab.label} + + ))} + + + {tabs.map((tab) => { + const resources = tools.filter((tool) => tool.type === tab.value); + const hasBuiltIns = + tab.value === "system" && showBuiltInSystemTools; + const isEmpty = resources.length === 0 && !hasBuiltIns; + + return ( + + {isEmpty ? ( +
+ 暂无可用的 {tab.label} 工具
- - ); - })} -
- )} + ) : ( +
+ {hasBuiltIns && + SYSTEM_TOOL_OPTIONS.map((option) => { + const checked = draftSystemTools.includes(option.value); + return ( + + ); + })} + + {resources.map((tool) => { + const checked = draftIds.includes(tool.id); + return ( + + ); + })} +
+ )} + + ); + })} +
diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index c64900d..f285591 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -184,6 +184,7 @@ function blankPromptForm(name: string): AssistantForm { actions: [], openingMessage: null, }, + systemTools: [], visionEnabled: false, visionModelResourceId: "", toolIds: [], @@ -479,6 +480,7 @@ export function AssistantPage(props: AssistantPageProps) { actions: a.startup?.actions ?? [], openingMessage: a.startup?.openingMessage ?? null, }, + systemTools: a.systemTools ?? [], visionEnabled: a.visionEnabled, visionModelResourceId: a.visionModelResourceId ?? "", toolIds: a.toolIds ?? [], @@ -555,6 +557,7 @@ export function AssistantPage(props: AssistantPageProps) { actions: [], openingMessage: null, }, + systemTools: [], visionEnabled: false, visionModelResourceId: null, modelResourceIds: {}, @@ -614,6 +617,7 @@ export function AssistantPage(props: AssistantPageProps) { enableInterrupt: form.enableInterrupt, turnConfig: form.turnConfig, startup: form.startup, + systemTools: form.runtimeMode === "pipeline" ? form.systemTools : [], visionEnabled: form.visionEnabled, visionModelResourceId: form.visionModelResourceId || null, modelResourceIds: { @@ -1300,7 +1304,9 @@ export function AssistantPage(props: AssistantPageProps) { vision: visionModelOptionsFor(""), }} toolOptions={tools - .filter((tool) => tool.status === "active") + .filter( + (tool) => tool.status === "active" && tool.type !== "system", + ) .map((tool) => ({ value: tool.id, label: tool.name }))} knowledgeOptions={kbOptions} onBack={() => router.push("/assistants")} diff --git a/frontend/src/components/pages/ComponentsToolsPage.tsx b/frontend/src/components/pages/ComponentsToolsPage.tsx index 8d8bc91..2e921e7 100644 --- a/frontend/src/components/pages/ComponentsToolsPage.tsx +++ b/frontend/src/components/pages/ComponentsToolsPage.tsx @@ -64,9 +64,9 @@ import { type ToolUpsert, } from "@/lib/api"; -type ToolKind = "end_call" | "http" | "client"; +type ToolKind = "system" | "http" | "client"; type HttpMethod = HttpToolDefinition["config"]["method"]; -type ToolFilter = "全部" | "End Call" | "HTTP" | "Client" | "MCP"; +type ToolFilter = "全部" | "System" | "HTTP" | "Client" | "MCP"; type SortOrder = "newest" | "oldest"; type ToolResource = | { kind: "tool"; id: string; tool: Tool } @@ -74,7 +74,7 @@ type ToolResource = const toolFilters: readonly ToolFilter[] = [ "全部", - "End Call", + "System", "HTTP", "Client", "MCP", @@ -112,7 +112,7 @@ function blankForm(): ToolForm { return { name: "", functionName: "", - type: "end_call", + type: "system", description: "", status: "active", messageType: "none", @@ -147,7 +147,7 @@ function formFromTool(tool: Tool): ToolForm { base.type = tool.type; base.description = tool.description; base.status = tool.status; - if (tool.definition.type === "end_call") { + if (tool.definition.type === "system") { base.messageType = tool.definition.config.messageType; base.customMessage = tool.definition.config.customMessage; base.captureReason = tool.definition.config.captureReason; @@ -235,7 +235,7 @@ function payloadFromForm(form: ToolForm): ToolUpsert { throw new Error("函数名必须以小写字母开头，且只能包含小写字母、数字和下划线"); } - if (form.type === "end_call") { + if (form.type === "system") { return { name: form.name.trim(), functionName: form.functionName, @@ -244,8 +244,9 @@ function payloadFromForm(form: ToolForm): ToolUpsert { secrets: {}, definition: { schemaVersion: 1, - type: "end_call", + type: "system", config: { + kind: "end_conversation", messageType: form.messageType, customMessage: form.messageType === "custom" ? form.customMessage : "", captureReason: form.captureReason, @@ -418,7 +419,7 @@ export function ComponentsToolsPage() { const tool = resource.tool; return ( (filter === "全部" || - (filter === "End Call" && tool.type === "end_call") || + (filter === "System" && tool.type === "system") || (filter === "HTTP" && tool.type === "http") || (filter === "Client" && tool.type === "client")) && (!query || @@ -569,8 +570,8 @@ export function ComponentsToolsPage() { > {resource.kind === "mcp" ? "MCP" - : resource.tool.type === "end_call" - ? "End Call" + : resource.tool.type === "system" + ? "System" : resource.tool.type === "client" ? "Client" : "HTTP"} @@ -783,7 +784,7 @@ export function ComponentsToolsPage() { > - End Call + System HTTP Client Tool @@ -826,8 +827,8 @@ export function ComponentsToolsPage() { - {form.type === "end_call" ? ( - + {form.type === "system" ? ( + ) : form.type === "client" ? ( ) : ( @@ -860,7 +861,7 @@ export function ComponentsToolsPage() { ); } -function EndCallFields({ +function SystemToolFields({ form, setForm, }: { @@ -869,6 +870,16 @@ function EndCallFields({ }) { return (
+ + +