10.2 工具设计模式:读写分离、幂等、可观测
10.1 解决了「工具从哪来」,但没回答「工具该怎么设计」。这两件事的差距,用一个例子就能说清:同一个 Agent,配 run_sql(sql) 能查库,配 query_metrics(metric, window) 也能查库,能力看起来一样——可前者的模型要自己写 SQL,后者的模型只填两个字段。工具的形状,直接决定了模型犯错的概率,也决定了系统能拦住多少错。
本节把 ResearchCopilot 的工具层推进到「可设计」:把工具设计收敛成五条可复用模式,并用假模型与 pydantic 两组对照实验,量化「好设计」相对「坏设计」的收益。
10.2.1 读写分离:只读随便调,写入要审批加幂等键
第一条模式是按「能不能撤销」把工具分档,这正是 4.3 已经立好的模型,这里把它当作设计起点复用:
| 档位 | 语义 | 例子 | 执行策略 |
|---|---|---|---|
readonly | 不改任何状态 | search_corpus / query_metrics | 直接执行,可并行 |
write | 改自己的数据,可回滚 | create_note / update_run | 执行,记审计,带幂等键 |
dangerous | 不可逆或影响外部 | delete_run / send_email | 挂起等人工审批 |
分档的价值在于它把「调用的自由度」和「后果的严重度」解耦:只读工具随便调,模型多试几次没关系;写入工具必须有幂等键(10.2.2),危险工具必须停下等人点头(4.3 的 ApprovalRequired + approved 集合)。一条工具被设计成什么档,比它的实现细节更重要——它决定了这行代码在系统里能造成多大破坏。
10.2.2 幂等:同一请求重复执行结果一致
Agent 会重试:网络抖动重试、模型重发、队列 redeliver(7.2 死信重投)。如果写入工具不幂等,一次重试就是一条重复记录。幂等的定义很朴素:同一个请求执行一次和执行 N 次,结果一致。
做法是给每个写入工具加一个幂等键,由调用方(Agent 服务)生成,而不是模型生成——模型可能每次编一个不同的键,那就不幂等了:
import hashlib, json
def idempotency_key(tool: str, arguments: dict) -> str:
"""由工具名 + 规范化参数推导幂等键;同参必同键,与调用次数无关。"""
canonical = json.dumps({"tool": tool, "args": arguments},
sort_keys=True, ensure_ascii=False)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
键的粒度是关键取舍:
| 键来源 | 幂等范围 | 适用 |
|---|---|---|
| 工具名 + 参数 | 全局:任何时刻同参都算一次 | 创建资源、发通知 |
| 会话 id + 工具名 + 参数 | 会话内去重 | 同一次对话内的重试 |
显式 idempotency_key 参数 | 调用方指定 | 跨重试、跨进程(7.2) |
sort_keys=True 不能省:{"a":1,"b":2} 和 {"b":2,"a":1} 是同一个请求,若键依赖字典顺序,两次调用会算出不同的键,幂等直接失效。执行侧则用「先查键、命中就返回上次结果,未命中才执行并记录」的经典套路——键要么是数据库唯一索引,要么是 Redis 的 SETNX(6.1)。真跑一遍验证「同参同键」与「重复调用只落一条」:
_store: dict[str, str] = {} # 生产环境里这里换成唯一索引或 Redis SETNX
def create_note(title, body):
key = idempotency_key("create_note", {"title": title, "body": body})
if key in _store:
return {"note_id": _store[key], "deduped": True}
nid = f"n{len(_store) + 1}"
_store[key] = nid
return {"note_id": nid, "deduped": False}
同参同键: True
异参异键: True
第一次: {"note_id": "n1", "deduped": false}
第二次: {"note_id": "n1", "deduped": true}
第二次调用拿到的是同一个 note_id,deduped: true 说明它命中了缓存、没有重复落库。这就是幂等的可观测形态——重试不再产生脏数据,而是明确地告诉你「这次是重复的」。
10.2.3 可观测:每次调用留入参、耗时、结果大小
第三条是每次工具调用都必须留痕。模型看不见工具内部,出了问题你只能靠记录复盘:这次调用传了什么、跑了多久、返回多大。这三样恰好对应三个诊断问题——参数错、慢、爆 token。
import time, json
def observed_call(registry, tc, sink: list) -> str:
t0 = time.perf_counter()
result = registry.call(tc)
record = {
"tool": tc.name,
"args": tc.arguments,
"ms": round((time.perf_counter() - t0) * 1000, 2),
"result_bytes": len(result.encode("utf-8")),
"error": json.loads(result).get("error") if result.startswith("{") else None,
}
sink.append(record)
return result
result_bytes 这个字段特别值得留:工具返回超大结果,是上下文爆掉的常见元凶。一个 search 工具返回 200KB 原文,模型下一轮就吃不下了。有了这个数,你可以给工具加输出上限(10.3 的沙箱就带 max_output),而不是等上下文溢出才去猜。这三条记录最终要进 OTel(13 章),这里先用一个内存 sink 演示形状:
观测记录: {"tool": "create_note", "args": {"title": "x", "body": "y"}, "ms": 0.02, "result_bytes": 34, "error": null}
一条记录里 ms 用来找慢工具、result_bytes 用来找「大胃王」工具、error 用来统计失败率——故障复盘时,先看这三个数,往往比翻模型日志更快定位问题。
10.2.4 粒度:一个工具做一件事
第四条模式是单一职责:一个工具只做一件事,别做「万能 exec」。反例是 run_sql(sql) 或 exec(command)——它把任意能力塞进一个工具,看起来省事,实则三重代价:
- 不可约束:参数是一个自由字符串,schema 无法表达「只能查、不能删」。
- 不可审计:日志里只有一坨 SQL,说不清这次调用属于哪个业务动作。
- 不可审批:4.3 的审批挂在工具粒度上,一个万能工具要么全审(烦)要么全放(险)。
好设计把能力拆开:query_metrics / list_metrics / search_papers,各自参数明确、可单独定档、可单独审计。粒度不是越细越好——拆太碎模型要多轮才凑齐一件事;但**「一个工具对应一个业务意图」是安全的起点**。两种设计的差别可以摊开成一张表:
| 维度 | 万能 run_sql(sql) | 拆分 query_metrics(metric, window) |
|---|---|---|
| schema 能否表达约束 | 否(自由字符串) | 是(枚举 + 类型) |
| 能否单独定档 | 否(只能整档) | 是(每个工具各定) |
| 审计可读性 | 低(一坨 SQL) | 高(字段即意图) |
| 模型填参难度 | 高(要会写 SQL) | 低(填两个字段) |
| 出错的爆炸半径 | 大(可删可改) | 小(只读查询) |
这张表也是 10.2.7 实验二的动机:不是「拆分更好用」,而是「拆分才能把约束写进 schema」——约束进不去 schema,服务端就失去了拦截的机会。
10.2.5 描述即提示词
第五条呼应卷一第 4 章:description 不是文档,它是提示词的一部分,直接进模型的上下文,决定模型「选不选它、填什么参数」。同名的三个工具,描述含糊与描述具体,模型的选中率能差出一大截——下面用实验量化。
写描述有三条实用规则:说清「什么时候用」而不只是「是什么」(「在本地论文语料里检索片段」比「搜索工具」有用得多);点名关键参数的含义(metric 如 cpu、latency);标出只读还是写入(模型据此判断要不要谨慎)。这三条都指向同一个目的:让模型仅凭 schema 文本,就能把用户意图映射到正确的工具上。
10.2.6 把五条模式收成一份工具契约
五条模式不是五个孤立的技巧,它们共同定义了一份工具契约:一个新工具进注册表之前,要能回答下面这张表里的每个问题。答不上来的,多半还没设计好。
| 模式 | 要回答的问题 | 落到哪里 |
|---|---|---|
| 读写分离 | 它能不能撤销?算哪一档? | PermissionedRegistry.add(tool, tier) |
| 幂等 | 重复调用会不会产生副作用?键是什么? | 写入工具的 idempotency_key |
| 可观测 | 出问题时能不能复盘? | 调用包装器记入参/耗时/字节数 |
| 粒度 | 它是不是只做一件事? | 一个工具对应一个业务意图 |
| 描述即提示词 | 模型知道什么时候用它吗? | Tool.description |
把契约写进代码,一个「合格的写入工具」长这样:
reg = PermissionedRegistry()
reg.add(
Tool(
name="create_note",
description="为当前用户创建一条研究笔记,写入后可回滚。title 是标题,body 是正文。",
parameters={"type": "object",
"properties": {"title": {"type": "string"},
"body": {"type": "string"}},
"required": ["title", "body"]},
func=create_note, # 内部用 idempotency_key 去重
),
tier=ToolTier.WRITE,
)
读这段定义,五条模式各就各位:描述说清了「何时用、参数是什么」,参数是结构化字段而非自由字符串,档位是 WRITE,函数体里带幂等键。这份契约的价值在于它可检查——评审一个新工具时,对着表格逐项过,缺哪项补哪项,比「感觉设计得还行」可靠得多。
10.2.7 真跑:好设计 vs 坏设计的选中率对照
实验一:描述质量 → 选中率。 用同一组工具名、同一组参数,唯一变量是 description。选一个「代理模型」来模拟模型的路由行为:它在候选工具里挑「问题用词」与「工具描述 + 参数名」词重叠最多的那个。这是真实模型路由行为的简化代理,趋势一致但绝对值不同。
import re
from minagent.tools import Tool
NAMES = ["query_metrics", "list_metrics", "search_papers"]
VAGUE = {"query_metrics": "查询数据。", "list_metrics": "列出数据。",
"search_papers": "搜索数据。"}
SPECIFIC = {
"query_metrics": "查询某个指标在时间窗口内的数值序列,只读。metric 如 cpu、latency;window 如 1h、24h。",
"list_metrics": "列出所有可查询的指标名称,只读,无参数。",
"search_papers": "在本地论文语料里按关键词检索相关片段,只读。query 是检索词。",
}
CASES = [("过去一小时的 cpu 指标是多少", "query_metrics"),
("有哪些指标可以查", "list_metrics"),
("帮我找关于 mcp 的论文", "search_papers"),
("查一下最近 24 小时的延迟", "query_metrics"),
("列出全部指标名称", "list_metrics"),
("检索向量缓存相关的文献", "search_papers")]
def tokens(s): return set(re.findall(r"[a-z0-9]+|[一-鿿]", s.lower()))
def route(intent, tools):
return max(tools, key=lambda t: len(tokens(intent) &
tokens(t.description + " " + " ".join(t.parameters["properties"])))).name
实测输出(Python 3.14.6):
[描述含糊] 正确选中率 = 4/6 = 67%
MISS 期望 list_metrics 实际 query_metrics | 有哪些指标可以查
MISS 期望 search_papers 实际 query_metrics | 帮我找关于 mcp 的论文
[描述具体] 正确选中率 = 6/6 = 100%
含糊描述下,query_metrics 和 list_metrics 的边界被「查询数据 / 列出数据」这两句糊掉了,模型(这里是代理)把两个意图都路由给了 query_metrics。描述写具体后,每个意图都命中了正确的工具。同一份代码、同一组工具,只改文案,选中率从 67% 涨到 100%——这就是「描述即提示词」的量化形态。
实验二:粒度 + 约束 → 危险调用拦截率。 同一个意图「查指标」,坏设计是万能 run_sql(sql: str),好设计是 query_metrics,字段用 Literal 枚举、加 extra="forbid"。给两边各喂 6 个粗心/越权模型可能产出的调用,看 schema 能拦下几个:
from typing import Literal
from pydantic import BaseModel, ConfigDict, ValidationError
class RunSqlArgs(BaseModel): # 坏设计:自由字符串,零约束
sql: str
class QueryMetricsArgs(BaseModel): # 好设计:枚举字段 + 禁止多余参数
model_config = ConfigDict(extra="forbid")
metric: Literal["cpu", "latency", "qps"]
window: Literal["1h", "24h"]
实测输出:
run_sql(sql: str) 拦截危险/非法调用 0/6
query_metrics(枚举+forbid) 拦截危险/非法调用 5/6
坏设计的 6 个调用里包含 DROP TABLE runs、DELETE FROM papers 这类破坏性语句,sql: str 一个都拦不住——因为从 schema 的角度它们都是「合法的字符串」。好设计靠枚举 + forbid 拦下 5 个(唯一放行的是那个合法调用),越权、拼错、缺参、多参全被挡在门外。约束写在 schema 里,模型就没有绕过它的机会。
10.2.8 真实环境 / 验收清单
真实模型下,两个实验的趋势一致、数值不同:真实 Claude 面对含糊描述时不会像代理那样「抓最像的那个」,它更可能反问、或随机挑一个,选中率通常比 67% 高;但「描述具体 → 选得更准」这条规律在真实模型上只会更强,因为真实模型读的是语义而非字面重叠。第二个实验的数值则与模型无关:约束是 pydantic 在服务端执行的,模型怎么写都不影响拦截率——这正是「把安全写进 schema」的价值。验收清单:
- 每个工具已定档(readonly / write / dangerous),危险档接 4.3 审批
- 写入工具有幂等键,键由服务端生成、
sort_keys=True - 每次调用记录入参、耗时、结果字节数
- 没有
run_sql/exec这类万能工具,能力已按意图拆开 - 工具描述写清了「什么时候用」与「参数含义」,不是一句「查询数据」
阅读导航:上一节:10.1 MCP 客户端接入:minagent 作为 client 连本地 Server · 下一节:10.3 工具沙箱:子进程/容器隔离高危工具 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。