11.2 改造与自测

前置：11.1 · 约 60 分钟 · 本课练习均可在浏览器中完成；完成后建议在本地改造真实脚本（见文末） · 核对日期 2026-10-06

目标

• 给 Agent 增加第二个工具，理解“一轮内并行请求多个工具”。
• 为工具执行补上错误处理与参数校验——这是 examples/agent/README.md 中“限制”一节列出的缺口。
• 通过终点自测，确认已达成“能逐行读懂 minimal_agent.py”的目标。

核心概念

一轮内的多个工具调用

模型可以在一次回复中请求多个工具（多个 tool_use 块），也可能分多轮请求。minimal_agent.py 的内层 for block in resp.content 循环天然支持前者：逐个执行并把所有 tool_result 放进同一条 user 消息。下面的模拟客户端看到“几点”和“天气”会在同一轮请求两个工具：

Python · 可运行

from types import SimpleNamespace as NS

# ---------- 模拟的 anthropic 客户端（结构与真实 SDK 相同：client.messages.create → resp） ----------
class FakeMessages:
    TRIGGERS = {"get_weather": "天气", "get_time": "几点"}   # 问题中出现关键词才调用对应工具

    def create(self, *, model, max_tokens, tools, messages):
        last = messages[-1]["content"]
        if isinstance(last, str):                       # 用户提问 → 决定调用哪些工具
            city = next((c for c in ["上海", "北京", "深圳"] if c in last), "上海")
            calls = [
                NS(type="tool_use", id=f"toolu_{i:02d}", name=t["name"], input={"city": city})
                for i, t in enumerate(tools, 1)
                if self.TRIGGERS.get(t["name"], "\0") in last
            ]
            if calls:
                return NS(stop_reason="tool_use", content=[NS(type="text", text="我来查一下。"), *calls])
            return NS(stop_reason="end_turn", content=[NS(type="text", text="这个问题不需要工具。")])
        outputs = [r["content"] for r in last if r["type"] == "tool_result"]   # 拿到工具结果 → 最终回答
        return NS(stop_reason="end_turn", content=[NS(type="text", text="根据工具结果：" + "；".join(outputs))])


class FakeAnthropic:
    def __init__(self, api_key=None, base_url=None):
        self.messages = FakeMessages()


def run_agent(client, question, tools, functions, max_turns=5):
    """与 minimal_agent.py 主循环相同，封装为函数（11.1 练习 1 的答案）。"""
    messages = [{"role": "user", "content": question}]
    for turn in range(1, max_turns + 1):
        resp = client.messages.create(model="fake", max_tokens=1024, tools=tools, messages=messages)
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason != "tool_use":
            return "".join([b.text for b in resp.content if b.type == "text"]), messages
        results = []
        for block in resp.content:
            if block.type == "tool_use":
                output = functions[block.name](**block.input)
                print(f"[第 {turn} 轮] 调用工具 {block.name}({block.input}) -> {output}")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
    return None, messages


def get_weather(city):
    return f"{city}：晴，22°C"

def get_time(city):
    return f"{city}：10:00"

tools = [
    {"name": "get_weather", "description": "查询城市天气",
     "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}},
    {"name": "get_time", "description": "查询城市当前时间",
     "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}},
]
TOOL_FUNCTIONS = {"get_weather": get_weather, "get_time": get_time}

text, messages = run_agent(FakeAnthropic(), "上海现在几点、天气如何？", tools, TOOL_FUNCTIONS)
print("最终回答：", text)
print("第 1 轮工具结果条数：", len(messages[2]["content"]))

tool_result 的错误标记

工具失败时，不应让整个程序崩溃，而应把错误作为结果告诉模型，让它决定下一步（重试、换参数或向用户说明）。Messages API 的 tool_result 支持 "is_error": true 字段来标记这种结果：

{"type": "tool_result", "tool_use_id": block.id, "content": "未知工具: get_time", "is_error": True}

易错点

• 只改 tools 不改 TOOL_FUNCTIONS（或反之）：模型请求了工具，本地却找不到函数 → KeyError。两处必须同步，或用 10.2 的注册型装饰器从函数生成两者。
• 多个 tool_result 分开发送：同一轮的所有结果应放在同一条 user 消息的 content 列表中。
• 吞掉工具异常后不告诉模型：模型会以为工具成功，基于错误信息继续推理。
• 相信模型生成的参数：参数可能缺失、多余或类型错误，执行前应校验。

对照

做法	本课（手写循环）	Agent 框架常见做法
注册工具	tools 列表 + TOOL_FUNCTIONS 字典	装饰器 @tool 根据函数签名与文档字符串自动生成 schema
错误处理	自己 try/except，返回 is_error	框架内置
参数校验	自己对照 input_schema 检查	由 Pydantic 等库按类型注解校验

手写一遍之后，再看任何 Agent 框架，都能认出它在替你做哪一步。

读代码

对照改造前后的第 86–96 行：

# 改造前：任何异常都会终止程序
output = TOOL_FUNCTIONS[block.name](**block.input)
results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

# 改造后：把执行细节封装进函数，错误作为结果回传
results.append(execute_tool(block, TOOL_FUNCTIONS, schemas))

练习

练习 1：增加 get_time 工具

在起始代码中增加工具 get_time(city)（返回 f"{city}：10:00"）：同时修改 tools（参照 get_weather 的 schema，description 写“查询城市当前时间”）与 TOOL_FUNCTIONS。检查会用问题“上海现在几点、天气如何？”运行 run_agent。

练习

from types import SimpleNamespace as NS

# ---------- 模拟的 anthropic 客户端（结构与真实 SDK 相同：client.messages.create → resp） ----------
class FakeMessages:
    TRIGGERS = {"get_weather": "天气", "get_time": "几点"}   # 问题中出现关键词才调用对应工具

    def create(self, *, model, max_tokens, tools, messages):
        last = messages[-1]["content"]
        if isinstance(last, str):                       # 用户提问 → 决定调用哪些工具
            city = next((c for c in ["上海", "北京", "深圳"] if c in last), "上海")
            calls = [
                NS(type="tool_use", id=f"toolu_{i:02d}", name=t["name"], input={"city": city})
                for i, t in enumerate(tools, 1)
                if self.TRIGGERS.get(t["name"], "\0") in last
            ]
            if calls:
                return NS(stop_reason="tool_use", content=[NS(type="text", text="我来查一下。"), *calls])
            return NS(stop_reason="end_turn", content=[NS(type="text", text="这个问题不需要工具。")])
        outputs = [r["content"] for r in last if r["type"] == "tool_result"]   # 拿到工具结果 → 最终回答
        return NS(stop_reason="end_turn", content=[NS(type="text", text="根据工具结果：" + "；".join(outputs))])


class FakeAnthropic:
    def __init__(self, api_key=None, base_url=None):
        self.messages = FakeMessages()


def run_agent(client, question, tools, functions, max_turns=5):
    """与 minimal_agent.py 主循环相同，封装为函数（11.1 练习 1 的答案）。"""
    messages = [{"role": "user", "content": question}]
    for turn in range(1, max_turns + 1):
        resp = client.messages.create(model="fake", max_tokens=1024, tools=tools, messages=messages)
        messages.append({"role": "assistant", "content": resp.content})
        if resp.stop_reason != "tool_use":
            return "".join([b.text for b in resp.content if b.type == "text"]), messages
        results = []
        for block in resp.content:
            if block.type == "tool_use":
                output = functions[block.name](**block.input)
                print(f"[第 {turn} 轮] 调用工具 {block.name}({block.input}) -> {output}")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
    return None, messages


def get_weather(city):
    return f"{city}：晴，22°C"

tools = [
    {"name": "get_weather", "description": "查询城市天气",
     "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}},
]
TOOL_FUNCTIONS = {"get_weather": get_weather}

快捷键：⌘/Ctrl + Enter 提交检查

练习 2：安全执行工具

实现 execute_tool(block, functions)，返回一个 tool_result 字典：

• 工具名不在 functions 中 → content 为 f"未知工具: {block.name}"，并带 "is_error": True；
• 执行时抛出任何 Exception → content 为 f"工具执行失败: {类型名}: {异常信息}"，带 "is_error": True；
• 成功 → {"type": "tool_result", "tool_use_id": block.id, "content": 输出}（不带 is_error）。

练习

def execute_tool(block, functions):
    output = functions[block.name](**block.input)
    return {"type": "tool_result", "tool_use_id": block.id, "content": output}

快捷键：⌘/Ctrl + Enter 提交检查

练习 3：按 schema 校验参数

实现 validate_input(schema, tool_input)，返回错误信息列表（无错误返回 []）：

1. schema["required"] 中缺失的键，每个生成 f"缺少参数: {键}"（按 required 中的顺序）；
2. tool_input 中不在 schema["properties"] 里的键，每个生成 f"未知参数: {键}"（按 tool_input 中的顺序）；
3. 属性声明 "type": "string" 但值不是 str 的，生成 f"类型错误: {键} 应为 string"。

练习

def validate_input(schema, tool_input):
    return []

快捷键：⌘/Ctrl + Enter 提交检查

练习 4：终点自测

不运行 minimal_agent.py，回答下列问题，把选项字母填入 ANSWERS（检查会核对答案）。

1. range(1, MAX_TURNS + 1) 在 MAX_TURNS = 5 时产生几个值？ A. 4 B. 5 C. 6
2. .env 中写了 LLM_PROVIDER=（空值），provider 的值是？ A. "" B. None C. "deepseek"
3. TOOL_FUNCTIONS[block.name](**block.input) 在 block.input == {"city": "上海"} 时等价于？ A. get_weather({"city": "上海"}) B. get_weather(city="上海") C. get_weather("city", "上海")
4. 第 97 行的 else 什么时候执行？ A. if 条件为假时 B. 循环被 break 时 C. 5 轮都没有 break 时
5. 工具结果以什么角色追加到 messages？ A. "user" B. "assistant" C. "tool"
6. 第 81 行的推导式中，b 是什么？ A. dict B. SDK 的内容块对象 C. 字符串
7. 模型在第 1 轮直接回答（不调用工具）时，循环结束后 messages 有几条？ A. 1 B. 2 C. 3
8. config = PROVIDERS[provider] 在 provider == "openai" 时会？ A. 返回 None B. 抛出 KeyError C. 使用 deepseek 配置

练习

ANSWERS = {1: "", 2: "", 3: "", 4: "", 5: "", 6: "", 7: "", 8: ""}

快捷键：⌘/Ctrl + Enter 提交检查

小结

• 一轮可以请求多个工具；同一轮的所有 tool_result 放进同一条 user 消息。
• 工具失败用 is_error 告诉模型，而不是让程序崩溃；执行前按 input_schema 校验参数。
• 通过自测，即达成本课程终点：能逐行读懂并改造 minimal_agent.py。

在本地改造真实脚本

浏览器练习通过后，在项目中实践（建议复制到 practice/ 目录再改，保留原示例）：

mkdir -p practice && cp examples/agent/minimal_agent.py practice/agent_v2.py
# 修改 practice/agent_v2.py：增加 get_time、execute_tool、validate_input，把问题改为“上海现在几点、天气如何？”
uv run practice/agent_v2.py

观察真实模型是在同一轮并行请求两个工具，还是分两轮请求（模型行为可能每次不同）。

延伸阅读

• 错误和异常
• 更多控制流工具
• 数据结构
• types.SimpleNamespace