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

refactor: unify system tools as resources

Browse Source
This commit is contained in:
Xin Wang
2026-08-04 17:05:26 +08:00
parent d1b05f16c7
commit 74a8be2357
26 changed files with 788 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/db/models.py
View File

@@ -154,8 +154,6 @@ 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),

36
backend/db/session.py
View File

@@ -10,6 +10,7 @@ import json
import settings
from services.interface_catalog import INTERFACE_DEFINITIONS
from services.system_tools import SYSTEM_TOOL_SPECS
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
AsyncSession,
@@ -49,24 +50,31 @@ async def sync_interface_definitions() -> None:
async def sync_default_tools() -> None:
"""Ensure system-provided reusable tools exist without overwriting edits."""
default_tools = [
{
"id": "tool_end_call_default",
"name": "结束对话",
"function_name": "end_call",
"type": "system",
"description": "当用户明确要求结束对话，或任务已完成时调用。",
"definition": {
"schema_version": 1,
"type": "system",
"config": {
"kind": "end_conversation",
default_system_tools = []
for kind, spec in SYSTEM_TOOL_SPECS.items():
config: dict[str, object] = {"kind": kind}
if kind == "end_conversation":
config.update(
{
"message_type": "none",
"custom_message": "",
"capture_reason": True,
}
)
default_system_tools.append(
{
**spec,
"type": "system",
"definition": {
"schema_version": 1,
"type": "system",
"config": config,
},
},
},
}
)
default_tools = [
*default_system_tools,
{
"id": "tool_show_message_default",
"name": "显示确认消息",

235
backend/migrations/versions/20260804_0012_unify_system_tool_resources.py Normal file
View File

@@ -0,0 +1,235 @@
"""unify system tools as reusable tool resources
Revision ID: 20260804_0012
Revises: 20260804_0011
"""
from __future__ import annotations
from collections.abc import Sequence
import json
from alembic import op
import sqlalchemy as sa
revision: str = "20260804_0012"
down_revision: str | Sequence[str] | None = "20260804_0011"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
SYSTEM_TOOLS = {
"end_conversation": {
"id": "tool_end_call_default",
"name": "结束对话",
"function_name": "end_conversation",
"description": (
"礼貌地结束本次对话。当用户明确告别、表示任务已完成"
"或要求挂断时调用。"
),
},
"update_state": {
"id": "tool_update_state_default",
"name": "更新状态",
"function_name": "update_state",
"description": (
"静默更新本次对话中已经声明并明确授权的动态变量。"
"只提交本轮获得或确认的信息，更新后继续当前回答。"
),
},
"skip_turn": {
"id": "tool_skip_turn_default",
"name": "跳过本轮",
"function_name": "skip_turn",
"description": (
"跳过当前轮次，不生成任何语音回复。"
"仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
),
},
"request_human_handoff": {
"id": "tool_request_human_handoff_default",
"name": "转接人工",
"function_name": "request_human_handoff",
"description": (
"提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
"解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
"回复用户并说明正在等待人工响应。"
),
},
}
def _definition(kind: str) -> dict:
config: dict[str, object] = {"kind": kind}
if kind == "end_conversation":
config.update(
{
"message_type": "none",
"custom_message": "",
"capture_reason": True,
}
)
return {"schema_version": 1, "type": "system", "config": config}
def _copy_global_agent_config(data: dict, settings: dict) -> None:
"""Freeze an inherited Agent before moving its node-local System tools."""
key_pairs = {
"defaultLlmResourceId": "llmResourceId",
"defaultAsrResourceId": "asrResourceId",
"defaultTtsResourceId": "ttsResourceId",
}
for source, target in key_pairs.items():
data[target] = settings.get(source)
for key in (
"visionEnabled",
"visionModelResourceId",
"knowledgeBaseId",
"knowledgeMode",
"knowledgeTopN",
"knowledgeScoreThreshold",
"enableInterrupt",
"turnConfig",
):
if key in settings:
data[key] = settings[key]
data["toolIds"] = list(settings.get("toolIds") or [])
data["inheritGlobalConfig"] = False
def upgrade() -> None:
connection = op.get_bind()
for kind, spec in SYSTEM_TOOLS.items():
if kind == "end_conversation":
connection.execute(
sa.text(
"""
UPDATE tools
SET name = :name,
function_name = :function_name,
type = 'system',
description = :description,
definition = CAST(:definition AS jsonb),
status = 'active',
updated_at = now()
WHERE id = :id
"""
),
{**spec, "definition": json.dumps(_definition(kind))},
)
connection.execute(
sa.text(
"""
INSERT INTO tools
(id, name, function_name, type, description, definition, secrets, status)
VALUES
(:id, :name, :function_name, 'system', :description,
CAST(:definition AS jsonb), '{}'::jsonb, 'active')
ON CONFLICT (function_name) DO NOTHING
"""
),
{**spec, "definition": json.dumps(_definition(kind))},
)
assistants = connection.execute(
sa.text("SELECT id, system_tools, graph FROM assistants")
).mappings().all()
for assistant in assistants:
assistant_id = str(assistant["id"])
for kind in assistant["system_tools"] or []:
spec = SYSTEM_TOOLS.get(str(kind))
if not spec:
continue
connection.execute(
sa.text(
"""
INSERT INTO assistant_tool_bindings (assistant_id, tool_id)
VALUES (:assistant_id, :tool_id)
ON CONFLICT DO NOTHING
"""
),
{"assistant_id": assistant_id, "tool_id": spec["id"]},
)
graph = assistant["graph"] or {}
if not isinstance(graph, dict):
continue
settings = graph.get("settings") or {}
changed = False
for node in graph.get("nodes") or []:
if node.get("type") != "agent":
continue
data = node.get("data") or {}
had_legacy_selection = "systemTools" in data
kinds = list(dict.fromkeys(data.pop("systemTools", []) or []))
if had_legacy_selection:
node["data"] = data
changed = True
if not kinds:
continue
if data.get("inheritGlobalConfig", True):
_copy_global_agent_config(data, settings)
tool_ids = list(data.get("toolIds") or [])
for kind in kinds:
spec = SYSTEM_TOOLS.get(str(kind))
if spec and spec["id"] not in tool_ids:
tool_ids.append(spec["id"])
connection.execute(
sa.text(
"""
INSERT INTO assistant_tool_bindings (assistant_id, tool_id)
VALUES (:assistant_id, :tool_id)
ON CONFLICT DO NOTHING
"""
),
{"assistant_id": assistant_id, "tool_id": spec["id"]},
)
data["toolIds"] = tool_ids
node["data"] = data
if changed:
connection.execute(
sa.text("UPDATE assistants SET graph = CAST(:graph AS json) WHERE id = :id"),
{"id": assistant_id, "graph": json.dumps(graph)},
)
op.drop_column("assistants", "system_tools")
def downgrade() -> None:
op.add_column(
"assistants",
sa.Column(
"system_tools",
sa.JSON(),
server_default=sa.text("'[]'"),
nullable=False,
),
)
op.execute(
"""
DELETE FROM assistant_tool_bindings
WHERE tool_id IN (
'tool_update_state_default',
'tool_skip_turn_default',
'tool_request_human_handoff_default'
)
"""
)
op.execute(
"""
DELETE FROM tools
WHERE id IN (
'tool_update_state_default',
'tool_skip_turn_default',
'tool_request_human_handoff_default'
)
"""
)
op.execute(
"""
UPDATE tools
SET function_name = 'end_call'
WHERE id = 'tool_end_call_default'
"""
)

2
backend/models.py
View File

@@ -114,8 +114,6 @@ 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 = ""

66
backend/routes/assistants.py
View File

@@ -17,6 +17,7 @@ from schemas import AssistantOut, AssistantUpsert
from services.auth import require_admin
from services.masking import mask, resolve_incoming_key
from services.node_specs import graph_references, normalize_graph, validate_graph
from services.system_tools import system_tool_kind
from services.tool_policy import policy_for_tool
from services.workflow_engine import WorkflowEngine
from sqlalchemy import select
@@ -58,13 +59,6 @@ def _validate_workflow(body: AssistantUpsert) -> None:
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
@@ -141,6 +135,60 @@ async def _validate_workflow_references(
raise HTTPException(400, f"Workflow 引用了无效知识库:{knowledge_id}")
async def _validate_system_tool_selection(
session: AsyncSession,
body: AssistantUpsert,
) -> None:
"""Validate System resources in the same selection path as every other tool."""
selected_ids = list(dict.fromkeys(body.tool_ids))
if not selected_ids or body.type not in {"prompt", "workflow"}:
return
rows = (
await session.execute(select(Tool).where(Tool.id.in_(selected_ids)))
).scalars().all()
tools_by_id = {tool.id: tool for tool in rows if tool.status == "active"}
system_kinds = {
tool.id: system_tool_kind(tool.definition or {})
for tool in rows
if tool.status == "active" and tool.type == "system"
}
invalid = [tool_id for tool_id, kind in system_kinds.items() if kind is None]
if invalid:
raise HTTPException(400, "系统工具配置无效: " + ", ".join(invalid))
if body.type == "prompt":
if body.runtime_mode == "realtime" and system_kinds:
raise HTTPException(400, "Prompt Realtime 模式暂不支持系统工具")
if (
"update_state" in system_kinds.values()
and not body.dynamic_variable_definitions
):
raise HTTPException(400, "启用更新状态工具前必须声明至少一个动态变量")
return
engine = WorkflowEngine(body.graph)
for node_id, node in engine.nodes.items():
node_type = node.get("type")
data = node.get("data") or {}
if node_type == "agent":
stage = engine.agent_stage_config(node_id)
kinds = {
system_kinds[tool_id]
for tool_id in stage.tool_ids
if tool_id in system_kinds
}
if "update_state" in kinds and not stage.state_variable_names:
raise HTTPException(
400,
f"Agent 节点 {node_id} 启用更新状态工具时必须授权至少一个变量",
)
elif node_type == "action":
tool_id = str(data.get("toolId") or "")
tool = tools_by_id.get(tool_id)
if tool and tool.type == "system":
raise HTTPException(400, f"Action 节点 {node_id} 不能调用系统工具")
async def _validate_startup_actions(
session: AsyncSession,
body: AssistantUpsert,
@@ -312,7 +360,6 @@ 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),
@@ -347,6 +394,7 @@ async def create_assistant(
):
_validate_workflow(body)
await _validate_workflow_references(session, body)
await _validate_system_tool_selection(session, body)
await _validate_startup_actions(session, body)
await _validate_vision_model(session, body)
await _validate_knowledge_base(session, body)
@@ -389,7 +437,6 @@ 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,
@@ -423,6 +470,7 @@ async def update_assistant(
raise HTTPException(404, "助手不存在")
_validate_workflow(body)
await _validate_workflow_references(session, body)
await _validate_system_tool_selection(session, body)
await _validate_startup_actions(session, body)
await _validate_vision_model(session, body)
await _validate_knowledge_base(session, body)

16
backend/schemas.py
View File

@@ -159,9 +159,6 @@ 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
knowledge_retrieval_config: KnowledgeRetrievalConfig = Field(
@@ -199,8 +196,6 @@ 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":
@@ -220,15 +215,6 @@ 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
@@ -270,7 +256,7 @@ class ToolParameter(CamelModel):
class SystemToolConfig(CamelModel):
kind: Literal["end_conversation"] = "end_conversation"
kind: SystemToolKind = "end_conversation"
message_type: Literal["none", "custom"] = "none"
custom_message: str = ""
capture_reason: bool = True

85
backend/services/brains/prompt_brain.py
View File

@@ -44,7 +44,7 @@ from services.message_stage import (
MessageStageSpec,
)
from services.runtime_variables import DynamicVariableError, DynamicVariableStore
from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties
from services.system_tools import state_update_properties, system_tool_kind
from services.tool_executor import ToolExecutionError, ToolExecutor
from services.tool_policy import policy_for_tool
@@ -128,11 +128,16 @@ class PromptBrain(BaseBrain):
if llm_tool_ids is not None and tool.id not in llm_tool_ids:
continue
if tool.type == "system":
schema, handler = self._make_end_call_tool(tool, runtime)
schema, handler = self._make_system_tool(tool, runtime)
elif tool.type in {"http", "mcp", "client"}:
schema, handler = self._make_remote_tool(tool, runtime)
else:
continue
if schema.name in registered_names:
logger.warning(
f"跳过工具 {tool.id}: 函数名 {schema.name} 已被占用"
)
continue
schemas.append(schema)
registered_names.add(schema.name)
policy = policy_for_tool(tool)
@@ -141,19 +146,6 @@ class PromptBrain(BaseBrain):
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:
@@ -598,50 +590,23 @@ class PromptBrain(BaseBrain):
)
return schema, end_call
# ---------- 内置系统工具(助手配置 system_tools 开关) ----------
# ---------- System 工具资源 ----------
def _make_system_tool(self, kind: str, runtime: BrainRuntime):
def _make_system_tool(self, tool, runtime: BrainRuntime):
kind = system_tool_kind(tool.definition or {})
if not kind:
raise ValueError(f"系统工具 {tool.id} 缺少有效 kind")
if kind == "end_conversation":
return self._make_end_conversation_tool(runtime)
return self._make_end_call_tool(tool, runtime)
if kind == "update_state":
return self._make_update_state_tool(runtime)
return self._make_update_state_tool(tool, runtime)
if kind == "skip_turn":
return self._make_skip_turn_tool()
return self._make_skip_turn_tool(tool)
if kind == "request_human_handoff":
return self._make_handoff_tool(runtime)
return self._make_handoff_tool(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):
def _make_update_state_tool(self, tool, runtime: BrainRuntime):
"""更新已声明的动态变量(会话状态),并让模型继续当前回答。"""
writable = state_update_properties(self._cfg.dynamic_variable_definitions)
@@ -676,8 +641,8 @@ class PromptBrain(BaseBrain):
)
schema = FunctionSchema(
name="update_state",
description=(
name=tool.function_name,
description=tool.description or (
"静默更新本次对话中已经声明并明确列出的动态变量。"
"只提交本轮获得或确认的信息，更新后继续当前回答。"
),
@@ -686,7 +651,7 @@ class PromptBrain(BaseBrain):
)
return schema, update_state
def _make_skip_turn_tool(self):
def _make_skip_turn_tool(self, tool):
"""跳过当前轮次,不生成任何语音回复。"""
async def skip_turn(params: FunctionCallParams) -> None:
@@ -700,8 +665,8 @@ class PromptBrain(BaseBrain):
)
schema = FunctionSchema(
name="skip_turn",
description=(
name=tool.function_name,
description=tool.description or (
"跳过当前轮次,不生成任何语音回复。"
"仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
),
@@ -715,7 +680,7 @@ class PromptBrain(BaseBrain):
)
return schema, skip_turn
def _make_handoff_tool(self, runtime: BrainRuntime):
def _make_handoff_tool(self, tool, runtime: BrainRuntime):
"""提交人工接管请求；请求完成前保持当前 AI 会话可用。"""
async def request_human_handoff(params: FunctionCallParams) -> None:
@@ -741,8 +706,8 @@ class PromptBrain(BaseBrain):
)
schema = FunctionSchema(
name="request_human_handoff",
description=(
name=tool.function_name,
description=tool.description or (
"提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
"解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
"回复用户并说明正在等待人工响应。"

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

@@ -57,7 +57,7 @@ from services.message_stage import (
MessageStageSpec,
)
from services.runtime_variables import DynamicVariableError, DynamicVariableStore
from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties
from services.system_tools import state_update_properties, system_tool_kind
from services.tool_executor import ToolExecutionError, ToolExecutor
from services.tool_policy import policy_for_tool
from services.workflow.agent import WorkflowAgentStage
@@ -564,22 +564,21 @@ class WorkflowBrain(BaseBrain):
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"}:
if not tool:
continue
if tool.type == "system":
append_function(
self._workflow_system_tool(
tool,
node_id=node_id,
state_variable_names=stage.state_variable_names,
)
)
elif tool.type in {"http", "mcp", "client"}:
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:
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,
@@ -738,27 +737,30 @@ class WorkflowBrain(BaseBrain):
def _workflow_system_tool(
self,
kind: str,
tool: RuntimeTool,
*,
node_id: str,
state_variable_names: tuple[str, ...],
) -> FlowsFunctionSchema:
"""Build one platform-owned tool scoped to the active Agent node."""
kind = system_tool_kind(tool.definition or {})
if kind == "update_state":
return self._workflow_update_state_tool(
tool,
node_id,
state_variable_names=state_variable_names,
)
if kind == "skip_turn":
return self._workflow_skip_turn_tool()
return self._workflow_skip_turn_tool(tool)
if kind == "request_human_handoff":
return self._workflow_handoff_tool(node_id)
return self._workflow_handoff_tool(tool, node_id)
if kind == "end_conversation":
return self._workflow_end_conversation_tool()
raise ValueError(f"未知系统工具: {kind}")
return self._workflow_end_conversation_tool(tool, node_id)
raise ValueError(f"系统工具 {tool.id} 缺少有效 kind")
def _workflow_update_state_tool(
self,
tool: RuntimeTool,
node_id: str,
*,
state_variable_names: tuple[str, ...],
@@ -791,8 +793,8 @@ class WorkflowBrain(BaseBrain):
}
return FlowsFunctionSchema(
name="update_state",
description=(
name=tool.function_name,
description=tool.description or (
"静默更新当前阶段明确授权的动态变量。"
"只提交本轮获得或确认的信息，更新后继续当前回答。"
),
@@ -805,7 +807,7 @@ class WorkflowBrain(BaseBrain):
)
@staticmethod
def _workflow_skip_turn_tool() -> FlowsFunctionSchema:
def _workflow_skip_turn_tool(tool: RuntimeTool) -> FlowsFunctionSchema:
async def handler(args, _flow_manager):
reason = str((args or {}).get("reason") or "").strip()
result = {"status": "success", "action": "skip_turn"}
@@ -815,8 +817,8 @@ class WorkflowBrain(BaseBrain):
setattr(handler, "_suppress_followup_llm", True)
return FlowsFunctionSchema(
name="skip_turn",
description=(
name=tool.function_name,
description=tool.description or (
"跳过当前轮次，不生成任何语音回复。"
"仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
),
@@ -830,7 +832,9 @@ class WorkflowBrain(BaseBrain):
handler=handler,
)
def _workflow_handoff_tool(self, node_id: str) -> FlowsFunctionSchema:
def _workflow_handoff_tool(
self, tool: RuntimeTool, 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(
@@ -852,8 +856,8 @@ class WorkflowBrain(BaseBrain):
}
return FlowsFunctionSchema(
name="request_human_handoff",
description=(
name=tool.function_name,
description=tool.description or (
"提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
"解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
"回复用户并说明正在等待人工响应。"
@@ -865,27 +869,55 @@ class WorkflowBrain(BaseBrain):
handler=handler,
)
def _workflow_end_conversation_tool(self) -> FlowsFunctionSchema:
def _workflow_end_conversation_tool(
self, tool: RuntimeTool, node_id: str
) -> FlowsFunctionSchema:
config = (tool.definition or {}).get("config") or {}
message_type = str(config.get("message_type") or "none")
custom_message = str(config.get("custom_message") or "").strip()
capture_reason = bool(config.get("capture_reason", True))
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)
uses_custom_message = message_type == "custom" and bool(custom_message)
self._waiting_for_generated_end_speech = not uses_custom_message
runtime = self._require_runtime()
runtime.call_end.begin(reason)
if uses_custom_message:
await self._queue_visible_speech(
custom_message,
source="workflow-system-tool",
node_id=node_id,
)
arm_tracked = getattr(
runtime.call_end,
"arm_after_tracked_speech",
None,
)
if callable(arm_tracked):
await arm_tracked()
else:
runtime.call_end.arm_after_speech()
return {"status": "success", "action": "ending_call"}
setattr(handler, "_suppress_followup_llm", True)
return FlowsFunctionSchema(
name="end_conversation",
description=(
name=tool.function_name,
description=tool.description or (
"礼貌地结束本次对话。当用户明确告别、表示任务已完成"
"或要求挂断时调用。"
),
properties={
"reason": {
"type": "string",
"description": "结束对话的简短原因。",
properties=(
{
"reason": {
"type": "string",
"description": "结束对话的简短原因。",
}
}
},
required=[],
if capture_reason
else {}
),
required=["reason"] if capture_reason else [],
handler=handler,
)

1
backend/services/config_resolver.py
View File

@@ -266,7 +266,6 @@ 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,

8
backend/services/node_specs.py
View File

@@ -11,7 +11,6 @@ from services.message_policy import (
MESSAGE_CONFIRMATION,
MESSAGE_PLAYBACK,
)
from services.system_tools import SYSTEM_TOOL_KINDS, normalize_system_tools
SPEC_VERSION = "3"
@@ -193,7 +192,7 @@ 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")))
data.pop("systemTools", None)
state_names = data.get("stateVariableNames")
data["stateVariableNames"] = list(
dict.fromkeys(str(name) for name in state_names or [] if str(name))
@@ -439,11 +438,6 @@ 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

65
backend/services/system_tools.py
View File

@@ -6,25 +6,56 @@ from collections.abc import Iterable, Mapping
from typing import Any
SYSTEM_TOOL_KINDS = frozenset(
{
"end_conversation",
"update_state",
"skip_turn",
"request_human_handoff",
}
)
SYSTEM_TOOL_SPECS: dict[str, dict[str, str]] = {
"end_conversation": {
"id": "tool_end_call_default",
"name": "结束对话",
"function_name": "end_conversation",
"description": (
"礼貌地结束本次对话。当用户明确告别、表示任务已完成"
"或要求挂断时调用。"
),
},
"update_state": {
"id": "tool_update_state_default",
"name": "更新状态",
"function_name": "update_state",
"description": (
"静默更新本次对话中已经声明并明确授权的动态变量。"
"只提交本轮获得或确认的信息，更新后继续当前回答。"
),
},
"skip_turn": {
"id": "tool_skip_turn_default",
"name": "跳过本轮",
"function_name": "skip_turn",
"description": (
"跳过当前轮次，不生成任何语音回复。"
"仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
),
},
"request_human_handoff": {
"id": "tool_request_human_handoff_default",
"name": "转接人工",
"function_name": "request_human_handoff",
"description": (
"提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
"解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
"回复用户并说明正在等待人工响应。"
),
},
}
SYSTEM_TOOL_KINDS = frozenset(SYSTEM_TOOL_SPECS)
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 system_tool_kind(definition: Mapping[str, Any] | None) -> str | None:
"""Return the supported platform behavior declared by a System resource."""
config = (definition or {}).get("config")
if not isinstance(config, Mapping):
return None
kind = str(config.get("kind") or "")
return kind if kind in SYSTEM_TOOL_KINDS else None
def state_update_properties(

7
backend/services/workflow_engine.py
View File

@@ -11,7 +11,6 @@ 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)
@@ -25,7 +24,6 @@ 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
@@ -148,9 +146,8 @@ 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")),
# Variable write scope always belongs to the Agent node, even when
# its selected tool resources come from Workflow defaults.
state_variable_names=tuple(
dict.fromkeys(
str(name)

101
backend/tests/test_brains.py
View File

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from schemas import AssistantUpsert, REALTIME_CAPABLE_TYPES
from schemas import AssistantUpsert, REALTIME_CAPABLE_TYPES, SystemToolConfig
from services.brains import BrainRuntime, SPECS, build_brain
from services.brains.base import GREETING_CONTEXT_MARKER
from services.brains.dify_llm import (
@@ -104,6 +104,17 @@ async def noop_queue_frame(_frame):
return None
def system_runtime_tool(kind: str) -> RuntimeTool:
return RuntimeTool(
id=f"tool_{kind}",
name=kind,
function_name=kind,
type="system",
description=f"{kind} system tool",
definition={"type": "system", "config": {"kind": kind}},
)
class BrainRegistryTests(unittest.TestCase):
def test_capability_matrix(self):
self.assertEqual(
@@ -154,45 +165,18 @@ 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, [])
def test_system_tool_config_accepts_all_platform_actions(self):
for kind in (
"end_conversation",
"update_state",
"skip_turn",
"request_human_handoff",
):
self.assertEqual(SystemToolConfig(kind=kind).kind, kind)
def test_system_tool_config_rejects_unknown_kind(self):
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"],
)
SystemToolConfig(kind="magic")
def test_workflow_keeps_dynamic_variables_and_tool_bindings(self):
assistant = AssistantUpsert(
@@ -916,11 +900,11 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
"default": None,
}
},
system_tools=[
"end_conversation",
"update_state",
"skip_turn",
"request_human_handoff",
tools=[
system_runtime_tool("end_conversation"),
system_runtime_tool("update_state"),
system_runtime_tool("skip_turn"),
system_runtime_tool("request_human_handoff"),
],
)
brain = build_brain(cfg)
@@ -971,7 +955,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_end_conversation_ends_call_after_generated_speech(self):
cfg = AssistantConfig(
type="prompt",
system_tools=["end_conversation"],
tools=[system_runtime_tool("end_conversation")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1009,7 +993,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_end_conversation_finishes_when_no_speech(self):
cfg = AssistantConfig(
type="prompt",
system_tools=["end_conversation"],
tools=[system_runtime_tool("end_conversation")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1039,7 +1023,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
dynamic_variable_definitions={
"user_name": {"type": "string", "required": False, "default": None}
},
system_tools=["update_state"],
tools=[system_runtime_tool("update_state")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1083,7 +1067,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_update_state_rejects_undeclared_variable(self):
cfg = AssistantConfig(
type="prompt",
system_tools=["update_state"],
tools=[system_runtime_tool("update_state")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1108,7 +1092,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_skip_turn_suppresses_response(self):
cfg = AssistantConfig(
type="prompt",
system_tools=["skip_turn"],
tools=[system_runtime_tool("skip_turn")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1134,7 +1118,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
async def test_request_human_handoff_keeps_call_available(self):
cfg = AssistantConfig(
type="prompt",
system_tools=["request_human_handoff"],
tools=[system_runtime_tool("request_human_handoff")],
)
brain = build_brain(cfg)
llm = FakeLLM()
@@ -1299,10 +1283,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
"id": "agent",
"type": "agent",
"data": {
"systemTools": [
"update_state",
"skip_turn",
"request_human_handoff",
"inheritGlobalConfig": False,
"toolIds": [
"tool_update_state",
"tool_skip_turn",
"tool_request_human_handoff",
],
"stateVariableNames": ["customer_name"],
},
@@ -1322,6 +1307,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
"default": None,
},
},
tools=[
system_runtime_tool("update_state"),
system_runtime_tool("skip_turn"),
system_runtime_tool("request_human_handoff"),
],
),
{},
assistant_id="asst_workflow_system_tools",
@@ -1462,7 +1452,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
set_tools=lambda _tools: None,
call_end=call_end,
)
tool = brain._workflow_end_conversation_tool()
tool = brain._workflow_end_conversation_tool(
system_runtime_tool("end_conversation"),
"agent",
)
await brain.on_assistant_text_start("turn-1")
result = await tool.handler({"reason": "用户告别"}, None)

41
backend/tests/test_workflow_v3.py
View File

@@ -1,12 +1,18 @@
from __future__ import annotations
import asyncio
import unittest
from copy import deepcopy
from types import SimpleNamespace
from unittest.mock import AsyncMock
from models import AssistantConfig, RuntimeModelResource
from fastapi import HTTPException
from routes.assistants import _validate_workflow, _validate_workflow_references
from routes.assistants import (
_validate_system_tool_selection,
_validate_workflow,
_validate_workflow_references,
)
from schemas import AssistantUpsert
from services.pipecat.service_factory import config_with_resource
from services.node_specs import graph_references, normalize_graph, validate_graph
@@ -363,7 +369,6 @@ class WorkflowGraphTests(unittest.TestCase):
agent = next(node for node in graph["nodes"] if node["id"] == "agent")
agent["data"].update(
{
"systemTools": ["update_state", "skip_turn"],
"stateVariableNames": ["customer", "order_status"],
}
)
@@ -374,7 +379,7 @@ class WorkflowGraphTests(unittest.TestCase):
"defaultTtsResourceId": "tts_global",
"visionEnabled": True,
"visionModelResourceId": "vision_global",
"toolIds": ["tool_global"],
"toolIds": ["tool_global", "update_state", "skip_turn"],
"knowledgeBaseId": "kb_global",
"knowledgeMode": "on_demand",
"knowledgeTopN": 8,
@@ -395,8 +400,10 @@ class WorkflowGraphTests(unittest.TestCase):
"vision_global",
)
self.assertTrue(engine.uses_vision())
self.assertEqual(inherited.tool_ids, ("tool_global",))
self.assertEqual(inherited.system_tools, ("update_state", "skip_turn"))
self.assertEqual(
inherited.tool_ids,
("tool_global", "update_state", "skip_turn"),
)
self.assertEqual(
inherited.state_variable_names,
("customer", "order_status"),
@@ -412,7 +419,7 @@ class WorkflowGraphTests(unittest.TestCase):
{
"inheritGlobalConfig": False,
"llmResourceId": "llm_agent",
"toolIds": ["tool_agent"],
"toolIds": ["tool_agent", "update_state", "skip_turn"],
"knowledgeBaseId": "",
"visionEnabled": False,
"visionModelResourceId": "",
@@ -428,8 +435,10 @@ class WorkflowGraphTests(unittest.TestCase):
self.assertFalse(custom.vision_enabled)
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.tool_ids,
("tool_agent", "update_state", "skip_turn"),
)
self.assertEqual(custom.knowledge_mode, "disabled")
self.assertTrue(custom.enable_interrupt)
self.assertEqual(
@@ -463,7 +472,6 @@ class WorkflowGraphTests(unittest.TestCase):
agent = next(node for node in graph["nodes"] if node["id"] == "agent")
agent["data"].update(
{
"systemTools": ["update_state"],
"stateVariableNames": ["customer"],
}
)
@@ -497,8 +505,8 @@ class WorkflowGraphTests(unittest.TestCase):
def test_agent_update_state_tool_requires_an_authorized_variable(self):
graph = valid_graph()
graph["settings"]["toolIds"] = ["tool_update_state"]
agent = next(node for node in graph["nodes"] if node["id"] == "agent")
agent["data"]["systemTools"] = ["update_state"]
body = AssistantUpsert(
name="缺少授权",
type="workflow",
@@ -512,8 +520,19 @@ class WorkflowGraphTests(unittest.TestCase):
graph=graph,
)
_validate_workflow(body)
system_tool = SimpleNamespace(
id="tool_update_state",
status="active",
type="system",
definition={"config": {"kind": "update_state"}},
)
result = SimpleNamespace(
scalars=lambda: SimpleNamespace(all=lambda: [system_tool])
)
session = SimpleNamespace(execute=AsyncMock(return_value=result))
with self.assertRaisesRegex(HTTPException, "必须授权至少一个变量"):
_validate_workflow(body)
asyncio.run(_validate_system_tool_selection(session, body))
def test_vision_resource_creates_isolated_runtime_config(self):
base = AssistantConfig(type="workflow", model="text-only")