本文目录7 个章节
普通函数进入 Agent 后会多出哪些风险
Fake Model 和行为测试已经固定了“搜索—读取—结束”这条基线。现在可以在不改变行为的前提下，把 dispatch_tool() 换成统一的 Tool Registry。
search_local_sources() 单独调用很简单。进入 Agent Loop 后，模型可能重复请求、传入极长参数，工具也可能超时或返回巨量内容。写工具还会出现更麻烦的情况：客户端认为超时，服务器已经完成写入，下一次重试造成重复副作用。
Tool Registry 给所有工具套上同一条执行边界。
模型请求
│
▼
Schema 校验 ─▶ 权限检查 ─▶ 超时与并发 ─▶ 执行函数
│
▼
模型结果 ◀─ 稳定错误 ◀─ 截断与脱敏 ◀─ 原始结果先给工具标出副作用
| 级别 | 示例 | 自动重试 |
|---|---|---|
read | 搜索、读取文档 | 暂时故障可有限重试 |
reversible_write | 保存可覆盖草稿 | 有幂等键并能查状态时重试 |
irreversible_write | 发送消息、付款 | 默认不自动重试，进入审批 |
超时只表示调用方不知道结果，无法证明操作失败。读取操作可以再试，写入操作要先查询执行状态。
ToolSpec 把描述与策略放在一起
from dataclasses import dataclass
from typing import Any, Callable
@dataclass(frozen=True)
class ToolSpec:
name: str
description: str
effect: str
timeout_seconds: float
max_attempts: int
max_output_chars: int
handler: Callable[..., Any]
class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, ToolSpec] = {}
def register(self, spec: ToolSpec) -> None:
if spec.name in self._tools:
raise ValueError("duplicate_tool")
self._tools[spec.name] = spec
def get(self, name: str) -> ToolSpec | None:
return self._tools.get(name)模型可见的描述、运行时策略和 handler 来自同一注册项，避免修改了 Schema 却忘记更新执行器。
统一调用结果
def invoke_tool(registry: ToolRegistry, name: str, arguments: dict) -> dict:
spec = registry.get(name)
if spec is None:
return {"ok": False, "error": "unknown_tool", "retryable": False}
try:
value = spec.handler(**arguments)
return {"ok": True, "data": str(value)[:spec.max_output_chars]}
except PermissionError:
return {"ok": False, "error": "forbidden", "retryable": False}
except (ConnectionError, TimeoutError):
return {"ok": False, "error": "transient_failure", "retryable": True}
except TypeError:
return {"ok": False, "error": "invalid_arguments", "retryable": False}
except Exception:
return {"ok": False, "error": "tool_failure", "retryable": False}模型只看到稳定错误码和 retryable。异常堆栈留给服务日志。retryable=True 也要受 max_attempts 和任务总预算限制。
写工具需要业务幂等键
保存简报时，幂等键可以由 task_id + path + content_hash 计算。相同业务意图重复提交，返回首次结果，不再写第二次。
class IdempotencyStore:
def __init__(self) -> None:
self._results: dict[str, dict] = {}
def run_once(self, key: str, operation: Callable[[], dict]) -> dict:
if key in self._results:
return self._results[key]
result = operation()
if result.get("ok"):
self._results[key] = result
return result网络 request ID 只能标识一次传输，不代表业务意图。进程重启后仍要去重时，幂等结果应放进数据库，并与输出文件状态一起提交。
用故障矩阵写测试
| 故障 | 读工具 | 写工具 | 预期结果 |
|---|---|---|---|
| 参数缺失 | 不执行 | 不执行 | invalid_arguments |
| 暂时断连 | 最多重试 2 次 | 先查状态 | 计入总预算 |
| 权限拒绝 | 不重试 | 不重试 | forbidden |
| 超时且结果未知 | 可以重查 | 禁止盲目重写 | outcome_unknown |
| 输出过大 | 截断 | 截断回执 | 记录原始长度 |
测试 handler 的实际调用次数。只断言最终错误码，无法发现执行器已经重复写了两次。
可靠工具层完成后的变化
Agent Loop 不再直接调用任意函数。所有行动经过注册表，错误和预算在同一位置可见。第 7 篇沿用这套 Tool Registry，把演示数据换成可检索资料库，并让 search_local_sources 与 get_document 进入同一条 RAG 管线。
REFERENCES
参考链接
所属系列
AI AGENT 入门教程