wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add workflow action runtime policies

Browse Source
This commit is contained in:
Xin Wang
2026-07-31 23:30:04 +08:00
parent c2f0f5eb04
commit f155f98e6e
10 changed files with 445 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
backend/services/brains/workflow_brain.py
View File

@@ -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)

43
backend/services/node_specs.py
View File

@@ -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 节点")

185
backend/tests/test_brains.py
View File

@@ -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 = []

54
backend/tests/test_workflow_v3.py
View File

@@ -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(

11
frontend/src/components/assistant-editor/dynamic-variables.tsx
View File

@@ -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);
}

8
frontend/src/components/pages/ComponentsToolsPage.tsx
View File

@@ -955,7 +955,8 @@ function ClientToolFields({
成功后更新动态变量
</div>
<div className="mt-0.5 text-xs text-muted-foreground">
只在客户端返回 status=ok 时应用结果映射
只在客户端返回 status=ok 时应用；Action
节点默认继承，也可单独覆盖或禁用
</div>
</div>
<Switch
@@ -1065,7 +1066,8 @@ function ToolRuntimeFields({
<div>
<div className="text-sm font-medium text-foreground">运行策略</div>
<div className="mt-1 text-xs leading-5 text-muted-foreground">
执行模式决定 Agent 是否等待结果；中断设置决定执行期间及随后播报是否接收用户插话。
仅用于 Agent 主动调用工具时的对话行为；Workflow Action
节点使用自己的用户输入策略。
</div>
</div>
<Field label="执行模式">
@@ -1088,7 +1090,7 @@ function ToolRuntimeFields({
<div>
<div className="text-sm font-medium text-foreground">允许用户打断</div>
<div className="mt-0.5 text-xs text-muted-foreground">
关闭后，工具运行和紧随其后的 Agent 播报期间会忽略用户输入
关闭后，Agent 工具运行和紧随其后的播报期间会忽略用户输入
</div>
</div>
<Switch

6
frontend/src/components/workflow/WorkflowCanvas.tsx
View File

@@ -70,6 +70,12 @@ function defaultNodeData(spec: RuntimeNodeSpec): WorkflowNodeData {
entryMode: "wait_user",
entrySpeech: "",
}
: spec.type === "action"
? {
arguments: {},
resultAssignmentMode: "inherit",
userInputPolicy: "queue",
}
: {}),
};
for (const field of spec.fields) {

83
frontend/src/components/workflow/panels/ActionNodePanel.tsx
View File

@@ -6,7 +6,11 @@ import { SectionCard } from "@/components/editor/section-card";
import { Textarea } from "@/components/ui/textarea";
import { NodeSelect } from "./controls";
import type { WorkflowNodeData } from "../specs";
import {
actionResultAssignmentMode,
actionUserInputPolicy,
type WorkflowNodeData,
} from "../specs";
import type { ModelOption } from "../types";
type ActionNodePanelProps = {
@@ -32,11 +36,14 @@ export function ActionNodePanel({
setAssignmentsJson,
commitActionJson,
}: ActionNodePanelProps) {
const resultAssignmentMode = actionResultAssignmentMode(draft);
const userInputPolicy = actionUserInputPolicy(draft);
return (
<SectionCard
icon={<Zap size={15} />}
title="工具执行"
description="确定性调用工具，并将响应字段写入会话动态变量"
description="确定性调用工具，并控制结果写入及执行期间的用户输入"
>
<NodeSelect
label="执行工具"
@@ -63,24 +70,60 @@ export function ActionNodePanel({
字符串值可使用 {"{{variable}}"} 动态变量。
</span>
</label>
<label className="block">
<div className="mb-1.5 text-sm font-medium text-foreground">
结果变量映射 JSON
</div>
<Textarea
rows={4}
value={assignmentsJson}
onChange={(event) => {
const value = event.target.value;
setAssignmentsJson(value);
commitActionJson(argumentsJson, value);
}}
className="field-sizing-fixed min-h-28 resize-y border-hairline-strong bg-background font-mono text-xs text-foreground placeholder:text-muted-soft"
/>
<span className="mt-1.5 block text-xs text-muted-foreground">
格式：变量名 → 响应 JSON Path。
</span>
</label>
<NodeSelect
label="结果变量更新"
value={resultAssignmentMode}
options={[
{ value: "inherit", label: "继承工具高级设置" },
{ value: "override", label: "使用节点自定义映射" },
{ value: "none", label: "不更新动态变量" },
]}
onChange={(value) =>
set("resultAssignmentMode", value || "inherit")
}
allowNone={false}
/>
{resultAssignmentMode === "override" ? (
<label className="block">
<div className="mb-1.5 text-sm font-medium text-foreground">
结果变量映射 JSON
</div>
<Textarea
rows={4}
value={assignmentsJson}
onChange={(event) => {
const value = event.target.value;
setAssignmentsJson(value);
commitActionJson(argumentsJson, value);
}}
className="field-sizing-fixed min-h-28 resize-y border-hairline-strong bg-background font-mono text-xs text-foreground placeholder:text-muted-soft"
/>
<span className="mt-1.5 block text-xs text-muted-foreground">
格式：变量名 → 响应 JSON Path；仅覆盖当前 Action。
</span>
</label>
) : (
<p className="-mt-1 text-xs leading-5 text-muted-foreground">
{resultAssignmentMode === "inherit"
? "使用所选工具高级设置中的动态变量赋值配置。"
: "工具执行结果不会写入会话动态变量。"}
</p>
)}
<NodeSelect
label="执行期间用户输入"
value={userInputPolicy}
options={[
{ value: "queue", label: "允许输入并排队（默认）" },
{ value: "block", label: "暂时禁止输入" },
]}
onChange={(value) => set("userInputPolicy", value || "queue")}
allowNone={false}
/>
<p className="-mt-1 text-xs leading-5 text-muted-foreground">
{userInputPolicy === "queue"
? "用户输入会保留，Action 完成并进入后续节点后再处理。"
: "Action 执行期间忽略新的语音、文本和图片输入。"}
</p>
{jsonError && (
<p role="alert" className="text-xs text-destructive">
{jsonError}

55
frontend/src/components/workflow/panels/NodeSettingsPanel.tsx
View File

@@ -52,8 +52,12 @@ export function NodeSettingsPanel({
setDraft(next);
onChange(next);
};
const set = (key: string, val: unknown) =>
const set = (key: string, val: unknown) => {
if (key === "resultAssignmentMode" && val !== "override") {
setJsonError("");
}
commit({ ...draft, [key]: val });
};
const setPatch = (patch: Partial<WorkflowNodeData>) =>
commit({ ...draft, ...patch });
const commitActionJson = (
@@ -230,44 +234,17 @@ export function NodeSettingsPanel({
)}
{spec.type === "action" && (
<>
<NodeSelect
label="确定性执行工具"
value={(draft.toolId as string) || ""}
options={toolOptions}
onChange={(value) => set("toolId", value || "")}
noneLabel="请选择工具"
/>
<div className="flex flex-col gap-2">
<label className="text-sm font-medium text-foreground">工具参数 JSON</label>
<Textarea
rows={5}
value={argumentsJson}
onChange={(event) => {
const value = event.target.value;
setArgumentsJson(value);
commitActionJson(value, assignmentsJson);
}}
className="field-sizing-fixed min-h-32 resize-y border-hairline-strong bg-background text-foreground placeholder:text-muted-soft"
/>
<span className="text-xs text-muted-soft">字符串值可使用 {"{{variable}}"} 动态变量。</span>
</div>
<div className="flex flex-col gap-2">
<label className="text-sm font-medium text-foreground">结果变量映射 JSON</label>
<Textarea
rows={4}
value={assignmentsJson}
onChange={(event) => {
const value = event.target.value;
setAssignmentsJson(value);
commitActionJson(argumentsJson, value);
}}
className="field-sizing-fixed min-h-28 resize-y border-hairline-strong bg-background text-foreground placeholder:text-muted-soft"
/>
<span className="text-xs text-muted-soft">格式：变量名 → 响应 JSON Path。</span>
</div>
{jsonError && <span className="text-xs text-destructive">{jsonError}</span>}
</>
<ActionNodePanel
draft={draft}
set={set}
toolOptions={toolOptions}
argumentsJson={argumentsJson}
assignmentsJson={assignmentsJson}
jsonError={jsonError}
setArgumentsJson={setArgumentsJson}
setAssignmentsJson={setAssignmentsJson}
commitActionJson={commitActionJson}
/>
)}
{spec.type === "handoff" && (

31
frontend/src/components/workflow/specs.ts
View File

@@ -10,6 +10,8 @@ export type WorkflowNodeType = "start" | "agent" | "action" | "handoff" | "end";
export type ContextPolicy = "inherit" | "fresh";
export type KnowledgeMode = "automatic" | "on_demand" | "disabled";
export type AgentEntryMode = "wait_user" | "generate" | "fixed_speech";
export type ActionResultAssignmentMode = "inherit" | "override" | "none";
export type ActionUserInputPolicy = "queue" | "block";
export type EdgeMode = "llm" | "expression" | "always";
export type ExpressionOperator =
| "eq"
@@ -44,7 +46,9 @@ export type WorkflowNodeData = {
turnConfig?: TurnConfig;
toolId?: string;
arguments?: Record<string, unknown>;
resultAssignmentMode?: ActionResultAssignmentMode;
resultAssignments?: Record<string, string>;
userInputPolicy?: ActionUserInputPolicy;
targetType?: "ai" | "human" | "queue" | "phone";
target?: string;
message?: string;
@@ -52,6 +56,33 @@ export type WorkflowNodeData = {
[key: string]: unknown;
};
/**
* Resolve how an Action writes tool results while preserving workflows saved
* before resultAssignmentMode existed. Those nodes explicitly passed an empty
* mapping at runtime, so an absent/empty mapping means "none", not "inherit".
*/
export function actionResultAssignmentMode(
data: WorkflowNodeData,
): ActionResultAssignmentMode {
if (
data.resultAssignmentMode === "inherit" ||
data.resultAssignmentMode === "override" ||
data.resultAssignmentMode === "none"
) {
return data.resultAssignmentMode;
}
return Object.keys(data.resultAssignments ?? {}).length > 0
? "override"
: "none";
}
/** 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;