diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 27f3233..5d86086 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -724,19 +724,29 @@ class WorkflowBrain(BaseBrain): self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION) await self._emit_node_active(node_id) data = self._engine.data(node_id) + runtime = self._require_runtime() + block_user_input = data.get("userInputPolicy") == "block" + if block_user_input and runtime.set_input_enabled: + # Blocking only suppresses new audio/text input while the Action + # runs. It deliberately does not cancel the tool. The default + # queue policy leaves input enabled; the turn lock serializes any + # completed user turn until this automatic path has finished. + runtime.set_input_enabled(False) tool_id = str(data.get("toolId") or "") tool = self._tool_by_id.get(tool_id) - if not tool: - self._store.values["system__last_action_status"] = "error" - self._store.values["system__last_action_error"] = f"工具不存在:{tool_id}" - return try: + if not tool: + raise ToolExecutionError(f"工具不存在:{tool_id}") arguments = self._store.render_data(data.get("arguments") or {}) result = await self._tools.execute( tool, arguments, - result_assignments=data.get("resultAssignments") or {}, + result_assignments=self._action_result_assignments(data), ) + if result.get("status") != "ok": + raise ToolExecutionError( + str(result.get("message") or "工具返回执行失败状态") + ) updated_variables = list(result.get("updated_variables") or []) if updated_variables: await self._emit_variables( @@ -749,6 +759,26 @@ class WorkflowBrain(BaseBrain): except (ToolExecutionError, ValueError) as exc: self._store.values["system__last_action_status"] = "error" self._store.values["system__last_action_error"] = str(exc)[:2048] + finally: + if block_user_input and runtime.set_input_enabled: + runtime.set_input_enabled(True) + + @staticmethod + def _action_result_assignments( + data: dict[str, Any], + ) -> dict[str, str] | None: + """Resolve node mapping semantics for ToolExecutor. + + ``None`` means inherit the reusable tool's mapping, while an empty + dictionary explicitly disables all result assignments. + """ + mode = str(data.get("resultAssignmentMode") or "none") + if mode == "inherit": + return None + if mode == "override": + assignments = data.get("resultAssignments") + return dict(assignments) if isinstance(assignments, dict) else {} + return {} async def _enter_handoff(self, node_id: str) -> None: self._state.enter(node_id, WorkflowStatus.HANDOFF) diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 4c6798c..72ee34c 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -11,6 +11,8 @@ SPEC_VERSION = "3" NODE_TYPES = {"start", "agent", "action", "handoff", "end"} EDGE_MODES = {"llm", "expression", "always"} AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"} +ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"} +ACTION_USER_INPUT_POLICIES = {"queue", "block"} AUTOMATIC_NODE_TYPES = {"start", "action", "handoff"} EXPRESSION_OPERATORS = { "eq", @@ -154,6 +156,24 @@ def _normalize_agent_data(data: dict[str, Any]) -> None: data["inheritGlobalConfig"] = not has_node_overrides +def _normalize_action_data(data: dict[str, Any]) -> None: + """Add Action defaults while preserving the behavior of saved graphs. + + Older Action nodes always passed an empty mapping when no node-level + assignments were configured, so they did *not* inherit the reusable + tool's assignments. Infer ``none`` for that shape and ``override`` for + an existing non-empty mapping. Newly created nodes should persist their + intended mode explicitly. + """ + if "resultAssignmentMode" not in data: + assignments = data.get("resultAssignments") + data["resultAssignmentMode"] = ( + "override" if isinstance(assignments, dict) and assignments else "none" + ) + data.setdefault("resultAssignments", {}) + data.setdefault("userInputPolicy", "queue") + + def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: settings.setdefault("globalPrompt", global_prompt) settings.setdefault("defaultLlmResourceId", "") @@ -179,10 +199,11 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: source.setdefault("nodes", []) source.setdefault("edges", []) for node in source["nodes"]: - if node.get("type") != "agent": - continue data = node.setdefault("data", {}) - _normalize_agent_data(data) + if node.get("type") == "agent": + _normalize_agent_data(data) + elif node.get("type") == "action": + _normalize_action_data(data) return source nodes = source.get("nodes") or [] @@ -215,6 +236,8 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: data.setdefault("scope", "session") elif new_type == "agent": _normalize_agent_data(data) + elif new_type == "action": + _normalize_action_data(data) elif new_type == "start": prompt = str(data.pop("prompt", "") or "").strip() if prompt: @@ -326,6 +349,20 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: data.get("entrySpeech") or "" ).strip(): errors.append(f"Agent 节点 {node_id} 的固定进入语不能为空") + elif node_type == "action": + data = node.get("data") or {} + assignment_mode = data.get("resultAssignmentMode") + if assignment_mode not in ACTION_RESULT_ASSIGNMENT_MODES: + errors.append( + f"Action 节点 {node_id} 的结果变量模式无效:{assignment_mode}" + ) + if not isinstance(data.get("resultAssignments"), dict): + errors.append(f"Action 节点 {node_id} 的结果变量映射必须是对象") + input_policy = data.get("userInputPolicy") + if input_policy not in ACTION_USER_INPUT_POLICIES: + errors.append( + f"Action 节点 {node_id} 的用户输入策略无效:{input_policy}" + ) if counts["start"] != 1: errors.append("工作流必须有且仅有一个 Start 节点") diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 1418e18..cd8975c 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -92,6 +92,10 @@ class FakeFunctionParams: self.properties = properties +async def noop_queue_frame(_frame): + return None + + class BrainRegistryTests(unittest.TestCase): def test_capability_matrix(self): self.assertEqual( @@ -806,6 +810,187 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(variable_events[-1]["changed"], ["order_status"]) self.assertEqual(variable_events[-1]["variables"], {"order_status": "paid"}) + async def test_action_result_assignment_modes_reach_tool_executor(self): + tool = RuntimeTool( + id="client_action", + name="客户端操作", + function_name="show_message", + type="client", + ) + brain = WorkflowBrain( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "inherit_action", + "type": "action", + "data": { + "toolId": "client_action", + "resultAssignmentMode": "inherit", + }, + }, + { + "id": "override_action", + "type": "action", + "data": { + "toolId": "client_action", + "resultAssignmentMode": "override", + "resultAssignments": {"choice": "action"}, + }, + }, + { + "id": "none_action", + "type": "action", + "data": { + "toolId": "client_action", + "resultAssignmentMode": "none", + }, + }, + ], + "edges": [], + }, + tools=[tool], + ) + ) + captured = [] + + async def execute(_tool, _arguments, *, result_assignments=None): + captured.append(result_assignments) + return {"status": "ok", "updated_variables": []} + + 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=FakeCallEnd(), + ) + brain._tools.execute = execute + + await brain._enter_action("inherit_action") + await brain._enter_action("override_action") + await brain._enter_action("none_action") + + self.assertEqual(captured, [None, {"choice": "action"}, {}]) + + async def test_action_client_error_sets_error_status(self): + tool = RuntimeTool( + id="client_action", + name="客户端操作", + function_name="show_message", + type="client", + ) + brain = WorkflowBrain( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "action", + "type": "action", + "data": {"toolId": "client_action"}, + }, + ], + "edges": [], + }, + tools=[tool], + ) + ) + + async def execute(_tool, _arguments, *, result_assignments=None): + return { + "status": "error", + "message": "用户关闭了确认弹窗", + "updated_variables": [], + } + + 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=FakeCallEnd(), + ) + brain._tools.execute = execute + + await brain._enter_action("action") + + self.assertEqual(brain._store.values["system__last_action_status"], "error") + self.assertEqual( + brain._store.values["system__last_action_error"], + "用户关闭了确认弹窗", + ) + + async def test_action_block_policy_only_suppresses_input_while_running(self): + tool = RuntimeTool( + id="client_action", + name="客户端操作", + function_name="show_message", + type="client", + ) + brain = WorkflowBrain( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "block_action", + "type": "action", + "data": { + "toolId": "client_action", + "userInputPolicy": "block", + }, + }, + { + "id": "queue_action", + "type": "action", + "data": { + "toolId": "client_action", + "userInputPolicy": "queue", + }, + }, + ], + "edges": [], + }, + tools=[tool], + ) + ) + input_states = [] + + async def execute(_tool, _arguments, *, result_assignments=None): + input_states.append("executing") + return {"status": "ok", "updated_variables": []} + + 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=FakeCallEnd(), + set_input_enabled=input_states.append, + ) + brain._tools.execute = execute + + await brain._enter_action("block_action") + self.assertEqual(input_states, [False, "executing", True]) + + input_states.clear() + await brain._enter_action("queue_action") + self.assertEqual(input_states, ["executing"]) + async def test_nodes_without_outgoing_edges_remain_active(self): queued = [] diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 5013a1f..6d1e263 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -92,6 +92,60 @@ class WorkflowGraphTests(unittest.TestCase): agent["data"]["entrySpeech"] = "您好,{{customer}}" self.assertEqual(validate_graph(normalized), []) + def test_action_defaults_preserve_legacy_result_assignment_behavior(self): + graph = valid_graph() + graph["nodes"].extend( + [ + { + "id": "legacy_override", + "type": "action", + "data": {"resultAssignments": {"status": "order.status"}}, + }, + {"id": "legacy_none", "type": "action", "data": {}}, + { + "id": "explicit_inherit", + "type": "action", + "data": {"resultAssignmentMode": "inherit"}, + }, + ] + ) + + normalized = normalize_graph(graph) + actions = { + node["id"]: node["data"] + for node in normalized["nodes"] + if node["type"] == "action" + } + + self.assertEqual( + actions["legacy_override"]["resultAssignmentMode"], "override" + ) + self.assertEqual(actions["legacy_none"]["resultAssignmentMode"], "none") + self.assertEqual( + actions["explicit_inherit"]["resultAssignmentMode"], "inherit" + ) + self.assertEqual(actions["legacy_none"]["resultAssignments"], {}) + self.assertEqual(actions["legacy_none"]["userInputPolicy"], "queue") + + def test_action_advanced_settings_are_validated(self): + graph = valid_graph() + graph["nodes"].append( + { + "id": "action", + "type": "action", + "data": { + "resultAssignmentMode": "unexpected", + "resultAssignments": [], + "userInputPolicy": "cancel", + }, + } + ) + errors = validate_graph(graph) + + self.assertTrue(any("结果变量模式无效" in error for error in errors)) + self.assertTrue(any("结果变量映射必须是对象" in error for error in errors)) + self.assertTrue(any("用户输入策略无效" in error for error in errors)) + def test_voice_resource_creates_isolated_runtime_config(self): base = AssistantConfig(type="workflow", asr="default", voice="default") asr = RuntimeModelResource( diff --git a/frontend/src/components/assistant-editor/dynamic-variables.tsx b/frontend/src/components/assistant-editor/dynamic-variables.tsx index 3f78bdd..5314f68 100644 --- a/frontend/src/components/assistant-editor/dynamic-variables.tsx +++ b/frontend/src/components/assistant-editor/dynamic-variables.tsx @@ -26,7 +26,10 @@ import { TabsList, TabsTrigger, } from "@/components/ui/tabs"; -import type { WorkflowGraph } from "@/components/workflow/specs"; +import { + actionResultAssignmentMode, + type WorkflowGraph, +} from "@/components/workflow/specs"; import type { DynamicVariableDefinition } from "@/lib/api"; const DYNAMIC_VARIABLE_PATTERN = /{{\s*([A-Za-z][A-Za-z0-9_]{0,63})\s*}}/g; @@ -93,6 +96,12 @@ export function activeWorkflowDynamicVariableDefinitions( } } for (const node of graph.nodes) { + if ( + node.type === "action" && + actionResultAssignmentMode(node.data) !== "override" + ) { + continue; + } for (const name of Object.keys(node.data.resultAssignments ?? {})) { if (isPublicDynamicVariableName(name)) names.add(name); } diff --git a/frontend/src/components/pages/ComponentsToolsPage.tsx b/frontend/src/components/pages/ComponentsToolsPage.tsx index af93c3b..d5c40fa 100644 --- a/frontend/src/components/pages/ComponentsToolsPage.tsx +++ b/frontend/src/components/pages/ComponentsToolsPage.tsx @@ -955,7 +955,8 @@ function ClientToolFields({ 成功后更新动态变量
- 只在客户端返回 status=ok 时应用结果映射 + 只在客户端返回 status=ok 时应用;Action + 节点默认继承,也可单独覆盖或禁用
运行策略
- 执行模式决定 Agent 是否等待结果;中断设置决定执行期间及随后播报是否接收用户插话。 + 仅用于 Agent 主动调用工具时的对话行为;Workflow Action + 节点使用自己的用户输入策略。
@@ -1088,7 +1090,7 @@ function ToolRuntimeFields({
允许用户打断
- 关闭后,工具运行和紧随其后的 Agent 播报期间会忽略用户输入 + 关闭后,Agent 工具运行和紧随其后的播报期间会忽略用户输入
} title="工具执行" - description="确定性调用工具,并将响应字段写入会话动态变量" + description="确定性调用工具,并控制结果写入及执行期间的用户输入" > -