Skip to content

26|工具接入:注册与排序

上一篇搭好了项目骨架。这一篇把工具接进来——第 18 篇设计的 Tool 接口,怎么在项目里组织成可用的工具池。

工具不是写完函数就完事。agent 运行时要拿到当前可用的工具列表,这个列表是动态组装的,顺序还有讲究。本篇讲工具注册中心和工具池的组装。

工具注册中心

所有工具集中注册,agent 执行工具前只从注册中心取。这样工具信息和工具实现绑定在一起,不会出现"函数能调用但权限信息找不到"的情况。

AIOps 概念图:工具注册中心

python
from dataclasses import dataclass
from typing import Callable

@dataclass
class ToolSpec:
    name: str
    permission: str            # 权限信息跟着工具走
    handler: Callable
    description: str
    parameters: dict
    is_concurrency_safe: bool = False
    is_read_only: bool = False
    is_destructive: bool = False

REGISTRY: dict[str, ToolSpec] = {}

def register_tool(spec: ToolSpec) -> None:
    REGISTRY[spec.name] = spec

def get_tool(name: str) -> ToolSpec | None:
    return REGISTRY.get(name)

def list_tools(readonly_only: bool = False) -> list[ToolSpec]:
    tools = list(REGISTRY.values())
    if readonly_only:
        tools = [t for t in tools if t.is_read_only]
    return tools

ToolSpec 把第 18 篇的能力声明和第 19 篇的权限信息打包在一起。注册时一并声明,调用时一并检查。list_tools(readonly_only=True) 支持只拿只读工具——某些降级场景只允许只读操作。

动态组装工具池

工具池不是启动时一次性建好不动。它是运行时根据环境动态组装的:哪些内建工具可用、哪些 MCP 工具在线、feature gate 开了哪些。

python
def assemble_tool_pool(session, mode) -> list[ToolSpec]:
    pool = []
    # 内建工具
    pool.extend(list_tools(readonly_only=(mode == "plan")))
    # MCP 工具(第 24 篇,动态发现)
    pool.extend(mcp_manager.get_tools())
    # 按 feature gate 过滤
    pool = [t for t in pool if feature_gate.allows(t.name)]
    # 排序(影响 prompt cache)
    pool = sort_tools_for_cache(pool)
    return pool

plan 模式只要只读工具(第 19 篇)。MCP 工具动态发现,可能运行时变化。feature gate 控制某些工具的启用。

工具注册不是启动时一次性完成。MCP 服务器可能新上线工具,feature gate 可能切换。工具池支持热更新,但热更新要小心——正在执行的轮次继续用旧工具列表,新工具从下一轮开始生效,不打断当前轮次。

工具排序和 prompt cache

工具列表的顺序影响 prompt cache 的缓存断点。这个在第 4 篇提过,这里展开。

AIOps 概念图:工具排序和缓存

工具定义是请求体前缀的一部分。前缀稳定,缓存命中;工具列表顺序变了,前缀变了,缓存失效。内建工具和 MCP 工具不随意混排——混排会让每次 MCP 工具发现都打乱内建工具的缓存前缀。

python
def sort_tools_for_cache(tools: list[ToolSpec]) -> list[ToolSpec]:
    # 内建工具在前(稳定),MCP 工具在后(可能变)
    builtin = [t for t in tools if not t.name.startswith("mcp__")]
    mcp = [t for t in tools if t.name.startswith("mcp__")]
    # 内建工具按稳定性排序:最常用的在前
    builtin.sort(key=lambda t: USAGE_FREQUENCY.get(t.name, 0), reverse=True)
    return builtin + mcp

排序原则:稳定的内建工具在前,可能变化的 MCP 工具在后。内建工具里最常用的放最前(第 7 篇讲过,模型对列表前面的工具关注度更高)。这样内建工具的前缀稳定,缓存命中率高;MCP 工具的变化只影响后缀,不破坏前缀缓存。

返回格式统一

所有工具返回统一结构,agent 循环处理结果时不用为每个工具写不同解析逻辑:

python
def run_tool_call(tool_call, context) -> dict:
    spec = get_tool(tool_call.name)
    if not spec:
        return {"ok": False, "error": f"未知工具: {tool_call.name}"}

    args = json.loads(tool_call.arguments)
    result = spec.handler(**args, context=context)
    return {
        "ok": result.get("ok", True),
        "data": result.get("data"),
        "error": result.get("error"),
        "truncated": result.get("truncated", False),
        "meta": {"tool": spec.name, "permission": spec.permission},
    }

ok 表示成功失败,data 是结果,truncated 标记是否截断(第 21 篇),meta 用于审计不发给模型。失败结果也要回填给模型(第 7 篇),防止编造。

工具描述走变更管理

工具描述(description)的修改要走变更管理,不能随手改。

修改工具描述看似简单,但会直接影响模型的选择行为。一次描述修改可能让模型从"查日志"变成"查指标",导致排查路径完全改变。把工具描述纳入版本控制,修改后跑回归测试验证选择准确率没退化(第 5 篇的回归测试)。

工具描述放在工具函数的 docstring 附近,代码评审时一起检查描述是否仍然准确。每次修改工具实现时,同步审查描述。

工具接入之后

工具池组装好了,agent 能调到正确的工具。但多轮对话的状态还没存——进程重启后 SessionState 就丢了,多 Worker 时每个进程有自己的状态。

下一篇讲会话持久化。第 8 篇的 SessionState 怎么落盘、怎么在多 Worker 间共享、竞态怎么处理。