4.2 参数校验与执行隔离
4.1 的分发层只做转发，它假设模型填的参数是对的。但模型是概率性的：它会把整数填成字符串、漏掉必填字段、甚至多编一个不存在的参数。这些参数一旦直接展开成 func(**arguments)，轻则抛 TypeError，重则执行了语义完全错误的操作。
本节把 minagent 推进到
minagent/tools.py的第二层：给每个Tool配一个 pydantic 参数模型，校验失败时把结构化错误回灌给模型，并用线程池超时、异常捕获、统一序列化把工具执行隔离在沙箱里。
4.2.1 用 pydantic 给每个工具配一个参数模型
第 4.1 节的极简推导器有两个硬伤：处理不了默认值，也处理不了嵌套类型。工业界的标准答案是 pydantic——它的 BaseModel 既描述 schema，又执行校验，还能量产 JSON Schema，一份定义三处复用。
需要说明：pydantic 是 minagent 核心引擎之外的一个可选依赖。第 4.1 节的纯标准库推导器已经够跑通流程，pydantic 带来的是工程上的严谨。不想引入第三方库的读者，可以用 inspect.signature 自己补默认值处理，代价是要重造一遍 pydantic 已经解决的问题。本节的示例按工程化口径写，把它装进 venv 即可。
本机实测 pydantic 版本 2.14.0（BRIEF §五 口径）。下面给两个工具各配一个参数模型：
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable
from pydantic import BaseModel, ConfigDict
class AddArgs(BaseModel):
a: int
b: int
class SearchArgs(BaseModel):
model_config = ConfigDict(extra="forbid")
keyword: str
limit: int = 5
AddArgs 两个必填整数；SearchArgs 的 limit 有默认值 5，extra="forbid" 表示多传一个参数就报错——这是默认关闭的开关，pydantic 默认会静默忽略多余字段，而"模型多编一个参数"恰恰是最该拦住的情况之一。
为什么把校验模型独立成类、而不是写一堆 if 判断？因为 BaseModel 同时承担了三件事：描述（产出 JSON Schema 给模型看）、校验（运行时检查并强制转换）、文档（字段定义即接口说明）。手写 if 判断只能做第二件，且极易漏掉分支。一份定义复用三处，是这个设计最实际的收益。
代价是多了一层间接：Tool 上多挂一个 model 字段，注册时要同时给函数和模型。这点样板代码换来的是类型安全与 schema 一致，值得。
把参数模型挂到 Tool 上，schema() 直接调用 model_json_schema()：
@dataclass
class Tool:
name: str
description: str
parameters: dict[str, Any]
func: Callable[..., Any]
model: type[BaseModel] # 参数校验模型
def schema(self) -> dict[str, Any]:
return {"name": self.name, "description": self.description,
"input_schema": self.model.model_json_schema()}
SearchArgs.model_json_schema() 的实测输出：
{
"additionalProperties": false,
"properties": {
"keyword": { "title": "Keyword", "type": "string" },
"limit": { "default": 5, "title": "Limit", "type": "integer" }
},
"required": ["keyword"],
"title": "SearchArgs",
"type": "object"
}
对比 4.1 的手写推导：limit 正确进了 properties 却不在 required 里，default: 5 也被如实带出。这正是 pydantic 相对极简推导器的价值——默认值、可选性、additionalProperties 全都对。
如果不想为每个工具单独定义一个类，也可以用 TypeAdapter 包一个 TypedDict。它同样能产出 JSON Schema、同样能校验：
from typing import TypedDict
from pydantic import TypeAdapter, ValidationError
class SearchParams(TypedDict):
keyword: str
limit: int
adapter = TypeAdapter(SearchParams)
print(adapter.validate_python({"keyword": "rag", "limit": 3}))
实测输出：
{'keyword': 'rag', 'limit': 3}
TypeAdapter 的 json_schema() 同样产出标准 schema，校验失败同样抛 ValidationError。两者的取舍很清楚：
| 维度 | BaseModel | TypeAdapter(TypedDict) |
|---|---|---|
| 复用粒度 | 一个类可挂多个方法、可嵌套 | 纯结构，无方法 |
| 默认值与约束 | 字段级 Field(...) 直接写 | 需在 Annotated 上挂元数据 |
| 适合场景 | 工具有明确参数对象 | 快速原型、一次性校验 |
本卷的 Tool 用 BaseModel，因为它要长期挂在注册表里、被反复复用。TypeAdapter 更适合临时校验一段外部 JSON。
4.2.2 校验失败：结构化错误回灌，不中断循环
校验失败时不能抛异常，否则循环就断了。正确做法是把 ValidationError 的 .errors() 序列化成 JSON，当成工具结果回灌。下面是改造后的 call()：
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from pydantic import ValidationError
class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, Tool] = {}
def add(self, tool: Tool) -> None:
self._tools[tool.name] = tool
def call(self, tc: ToolCall, timeout: float = 2.0) -> str:
tool = self._tools.get(tc.name)
if tool is None:
return json.dumps({"error": f"unknown tool: {tc.name}"}, ensure_ascii=False)
# 1) 参数校验：失败回灌结构化错误，不抛
try:
args = tool.model.model_validate(tc.arguments)
except ValidationError as e:
return json.dumps({"error": "invalid arguments",
"details": e.errors(include_url=False)},
ensure_ascii=False)
# 2) 执行隔离：超时 + 异常捕获
try:
with ThreadPoolExecutor(max_workers=1) as pool:
result = pool.submit(tool.func, **args.model_dump()).result(timeout=timeout)
except FutureTimeout:
return json.dumps({"error": f"timeout after {timeout}s"}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": f"{type(e).__name__}: {e}"}, ensure_ascii=False)
# 3) 结果序列化
if isinstance(result, str):
return result
return json.dumps(result, ensure_ascii=False)
e.errors(include_url=False) 是关键：不加这个参数，每条错误会带上一个指向 pydantic 文档的 URL，白白浪费 token。实测三种典型错误：
# a 传字符串 "two"
{"error": "invalid arguments", "details": [{"type": "int_parsing", "loc": ["a"], "msg": "Input should be a valid integer, unable to parse string as an integer", "input": "two"}]}
# 缺参数 b
{"error": "invalid arguments", "details": [{"type": "missing", "loc": ["b"], "msg": "Field required", "input": {"a": 2}}]}
# extra=forbid 下多传 limitt
{"error": "invalid arguments", "details": [{"type": "extra_forbidden", "loc": ["limitt"], "msg": "Extra inputs are not permitted", "input": 3}]}
每条错误都精确指出了 loc（哪个字段）、type（哪类问题）、input（模型给的值）。这比一句"参数错误"信息量大得多——模型能据此精确修正。
错误信息的详细程度要权衡 token。三类典型错误里，missing 和 extra_forbidden 都是一句话，只有 int_parsing 带了一长串解释。如果工具参数很多、错误很密，可以把 details 精简成只保留 loc 和 type：
slim = [{"loc": e["loc"], "type": e["type"]} for e in errs]
这在评测或高频重试场景下能省下可观的 token，代价是模型少了"为什么错"的解释、修正得更慢。卷一的默认选择是保留完整 details，因为教学场景下可读性优先。
4.2.3 把错误包成 tool 消息回灌
校验错误最终要变成一条 role="tool" 的消息，挂在原 tool_call_id 上，追加进 messages。第 3 章的循环看到的是"工具返回了一段 JSON"，而不是异常：
from dataclasses import dataclass, field
@dataclass
class Message:
role: str
content: str = ""
tool_calls: list[Any] = field(default_factory=list)
tool_call_id: str | None = None
err = reg.call(ToolCall("c2", "add", {"a": "two", "b": 3}))
messages: list[Message] = [Message(role="user", content="算一下 two + 3")]
messages.append(Message(role="tool", content=err, tool_call_id="c2"))
实测这条消息的形状：
role=tool tool_call_id=c2
content: {"error": "invalid arguments", "details": [{"type": "int_parsing", "loc": ["a"], "msg": "Input should be a valid integer", "input": "two"}]}
循环未中断，messages 长度 = 2
messages 长度变成 2，循环继续。模型下一轮看到"a 不是合法整数"，通常会重发一个 {"a": 2, "b": 3}。一次坏参数只消耗一轮，而不是终止会话——这就是"不中断循环"的价值。
tool_call_id 不能省。Anthropic 与 OpenAI 都要求每条 tool 消息带一个 id，指向它回应的那个 tool_call。当一轮里有多个工具调用时，这个 id 就是模型把"结果"和"请求"对上的唯一凭据——4.3 节并行调用时，它的作用会更明显。忘了带 id，API 会直接报 400。
4.2.4 执行隔离的三道闸
参数校验保证"输入合法"，执行隔离保证"过程可控"。工具是我们自己写的，但不等于可信：它可能访问网络、读文件、调外部服务，任何一个环节都可能抛异常或长时间不返回。Agent 循环是长跑，一个卡住的工具会拖垮整个会话，所以执行层必须自带围栏。
工具执行层要挡住三类问题，各自对应一段代码：
| 风险 | 挡法 | 失败时的返回 |
|---|---|---|
| 工具内部抛异常 | except Exception | {"error": "TypeError: ..."} |
| 工具卡死不返回 | ThreadPoolExecutor(...).result(timeout=) | {"error": "timeout after 2.0s"} |
| 返回不可序列化对象 | json.dumps(result, ensure_ascii=False) | 直接是 JSON 字符串 |
第三点常被忽略：工具可能返回一个 datetime、一个自定义对象、甚至一个打开的文件句柄。json.dumps 遇到这些会抛 TypeError——所以序列化本身也要在 try 的保护范围内，或者先用 default=str 兜底。ensure_ascii=False 则是为了让中文正常输出，而不是被转成 \uXXXX。
隔离是分层的，本节的线程池只是第一层：
| 层级 | 手段 | 挡得住 | 挡不住 |
|---|---|---|---|
| 进程内 | try/except | 普通异常 | os._exit、段错误 |
| 线程 | ThreadPoolExecutor + timeout | 死循环、IO 卡死 | GIL 抢占、CPU 密集饿死 |
| 进程 | subprocess + 超时 | 崩溃、内存爆炸 | 启动开销、IPC 复杂度 |
卷一停在线程层，因为它零依赖、够用且能讲清楚。真正的"不可信代码沙箱"（用户上传的脚本、任意命令执行）要下沉到进程甚至容器层，那是卷二生产化加固的内容。这里的超时尤其重要：ThreadPoolExecutor 的 result(timeout=) 超时后不会杀死线程（Python 无法强制终止线程），它只是让 call() 先返回一个错误，那个卡住的线程仍在后台跑——所以对不可信工具不能只靠这一层。
另一个细节是 args.model_dump()：它把校验后的模型还原成普通 dict 再展开给 func。用 model_dump() 而不是原始 tc.arguments，是为了让工具拿到的是已经过校验和强制转换的值——比如字符串 "3" 已经被转成整数 3，工具本身不必再做类型转换。
还有一点容易被忽略：ThreadPoolExecutor 最好每次调用都新建，或者用模块级的共享池。前者简单但有创建开销，后者高效但要在退出时 shutdown。卷一的教学版图省事用了 with 局部池，每次调用建一次；真实高并发场景应换成共享池，否则工具调用频繁时线程创建本身就是瓶颈。
到这一步，minagent 的工具层已经有了完整的"接—验—跑—返"闭环：Tool 描述形状，pydantic 校验参数，线程池隔离执行，json.dumps 统一出口。它唯一的假设是"一次只跑一个工具"。
真实模型下会怎样：假模型不会主动填错参数，本节所有错误都是我们手工构造来演示的。真实 Claude 在 schema 写得清楚时，参数类型基本正确，但会稳定踩两类坑：一是把可选字段的默认值显式填成 null，二是把数字填成字符串（尤其从上下文里抄来的数字）。有了 4.2 的校验层，这些都不再是崩溃点，而是一条能自我纠正的反馈。另一个真实差异是超时——真实工具（网络请求、数据库查询）耗时可从几十毫秒到几十秒不等，timeout 的具体取值要按工具类型分别设定，而不是全局一个数。
阅读导航：上一节：4.1 工具即函数：schema 生成、注册表与分发 · 下一节：4.3 并行工具调用与结果合并 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。