手写一个智能体循环

"智能体"（agent）这个词被说得很玄，好像是某种全新的技术。其实你在第 03 模块第 3 课已经写过它的核心了：那个"调用模型，有工具调用就执行，把结果放回去，再调用模型"的循环。

那一课的循环只查了一次 PyPI 就结束了。这一课把它写成一个真正的智能体：给它几个查文档的工具，让它自己决定先查什么、再查什么、什么时候可以回答。整个程序不用任何框架，一百多行。写完之后你再看 LangGraph、OpenAI Agents SDK 这类框架，会发现它们都是在这个循环外面加东西。

一个智能体由什么组成

            ┌──────────────────────────────┐
            │  消息列表：system、问题、        │
            │  模型的每一步、工具的每个结果      │
            └──────────────┬───────────────┘
                           ▼
        ┌──────▶  调用模型（带上工具说明书）
        │                  │
        │        有工具调用吗？ ──没有──▶ 这就是最终回答，结束
        │                  │有
        │                  ▼
        │        逐个执行工具，出错也变成一条结果
        │                  │
        └── 结果放回消息列表 ◀┘        （达到最大步数也结束）

五样东西：

1. 消息列表：整个过程的记录，也是模型每一步能看到的全部信息。
2. 工具：模型能调用的函数，以及给模型看的说明书。
3. 循环：调用模型、执行工具、放回结果，重复。
4. 停止条件：模型不再调用工具，或者达到最大步数。
5. 观察结果的处理：工具的返回值、报错信息怎么变成模型能读的文字，太长了怎么办。

下面一样一样写。

工具：函数加说明书

第 03 模块第 3 课手写了工具的 JSON 说明书，一个工具要写十几行。工具多了很麻烦，这次写一个装饰器，从函数本身自动生成说明书：

import inspect

TOOLS = {}


def tool(description, **params):
    """把一个函数注册成工具。params 是每个参数给模型看的说明。"""
    def register(fn):
        sig = inspect.signature(fn)
        properties = {name: {"type": "integer" if p.annotation is int else "string", "description": params[name]}
                      for name, p in sig.parameters.items()}
        required = [name for name, p in sig.parameters.items() if p.default is inspect.Parameter.empty]
        TOOLS[fn.__name__] = {
            "fn": fn,
            "schema": {"type": "function", "function": {
                "name": fn.__name__, "description": description,
                "parameters": {"type": "object", "properties": properties, "required": required}}},
        }
        return fn
    return register

它读取函数的参数列表：参数名变成 JSON Schema 的属性名，标注了 int 的参数类型是 integer，其他都当成 string，没有默认值的参数就是必填的。这是一个简化版，只支持字符串和整数，够这一课用了。

给智能体三个工具，都是操作 httpx 文档的：

DOCS = (Path(__file__).parent / "../../data/httpx-docs").resolve()


@tool("列出 httpx 文档的所有文件路径。")
def list_docs():
    return "\n".join(str(p.relative_to(DOCS)) for p in sorted(DOCS.rglob("*.md")) if p.name != "LICENSE.md")


@tool("在 httpx 文档里搜索一个英文关键词（不区分大小写），返回出现的文件和行号，最多 20 条。",
      keyword="要搜索的英文关键词，例如 timeout")
def grep_docs(keyword):
    hits = []
    for p in sorted(DOCS.rglob("*.md")):
        for n, line in enumerate(p.read_text().splitlines(), 1):
            if keyword.lower() in line.lower():
                hits.append(f"{p.relative_to(DOCS)}:{n}: {line.strip()[:100]}")
    return "\n".join(hits[:20]) or f"没有找到 {keyword}"


@tool("读取一个文档文件的指定行，返回带行号的内容。一次最多读 80 行。",
      path="文件路径，来自 list_docs 或 grep_docs 的结果", start="起始行号，从 1 开始", end="结束行号")
def read_doc(path, start: int = 1, end: int = 80):
    target = (DOCS / path).resolve()
    if DOCS not in target.parents or not target.exists():  # 不许读文档目录以外的文件
        return f"错误：没有这个文件 {path}，请先用 list_docs 查看有哪些文件"
    lines = target.read_text().splitlines()
    end = min(end, start + 79, len(lines))
    return "\n".join(f"{n}: {lines[n - 1]}" for n in range(start, end + 1))

注意这三个工具和第 04 模块的 RAG 完全不同：没有向量，没有检索算法，就是最朴素的"列文件、搜关键词、按行读"。检索的智能全部交给了模型：它决定搜什么词、读哪个文件的哪几行。这其实就是人查文档的方式，也是 Claude Code、Cursor 这类编程智能体翻代码的方式。

几个细节都是有意为之：

• grep_docs 最多返回 20 条，read_doc 一次最多读 80 行。工具返回太多内容会迅速撑满上下文，也会让模型抓不住重点。
• read_doc 返回的每一行都带着行号，模型回答时就能准确地注明出处。
• read_doc 会检查路径，不允许读文档目录以外的文件。如果模型（或者被人操纵的模型）传入 ../../../etc/passwd，resolve() 之后的路径不在 DOCS 里面，直接拒绝。第 8 课会专门讲为什么这很重要。
• 出错时返回的信息是写给模型看的："没有这个文件，请先用 list_docs 查看有哪些文件"。它不只说错了，还告诉模型下一步该怎么做。

循环

SYSTEM = """你是 httpx 的答疑助手，可以使用工具查阅 httpx 的官方文档。
先用工具找到依据，再回答；回答要注明依据的文件和行号。找不到依据就如实说明。"""
MAX_OBSERVATION = 3000  # 工具返回的内容太长时截断，免得把上下文撑爆


def run_agent(model, question, max_steps=8, verbose=True):
    """返回 (最终回答, 统计信息)。达到最大步数还没回答完，最终回答是 None。"""
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    schemas = [t["schema"] for t in TOOLS.values()]
    stats = {"steps": 0, "tool_calls": 0, "prompt_tokens": 0, "completion_tokens": 0}
    for step in range(1, max_steps + 1):
        message, usage = model(messages, schemas)
        stats["steps"] = step
        if usage:
            stats["prompt_tokens"] += usage.prompt_tokens
            stats["completion_tokens"] += usage.completion_tokens
        if not message.tool_calls:  # 没有要调用的工具，说明模型给出了最终回答
            if verbose:
                print(f"[第 {step} 步] 回答：\n{message.content}")
                print(f"\n共 {step} 步，输入 {stats['prompt_tokens']} 词元，输出 {stats['completion_tokens']} 词元")
            return message.content, stats
        messages.append(message.model_dump(exclude_none=True))
        for call in message.tool_calls:
            try:
                args = json.loads(call.function.arguments or "{}")
                result = TOOLS[call.function.name]["fn"](**args)
            except KeyError:
                result = f"错误：没有叫 {call.function.name} 的工具"
            except Exception as e:  # 参数不对、文件读不了……都变成一条观察结果交给模型，而不是让程序崩掉
                result = f"错误：{type(e).__name__}: {e}"
            if len(result) > MAX_OBSERVATION:
                result = result[:MAX_OBSERVATION] + f"\n……（内容太长，已截断，共 {len(result)} 字符）"
            preview = result.replace("\n", " | ")[:90]
            print(f"[第 {step} 步] {call.function.name}({call.function.arguments}) → {preview}")
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
    if verbose:
        print(f"达到最大步数 {max_steps}，停止。")
    return None, stats

和第 03 模块第 3 课的循环比，多了这几样：

• 两种停止条件。模型不再调用工具，就是给出了最终回答；达到 max_steps 还没结束，就强行停止。后者是必不可少的保险：模型可能陷入反复调用同一个工具的循环，每一步都在花钱。
• 所有错误都变成观察结果。模型编了一个不存在的工具名（KeyError）、参数传错了（TypeError）、文件读不了，都不会让程序崩溃，而是变成一条"错误：……"的消息交还给模型。模型看到错误，通常会自己改正。
• 截断过长的结果。一个工具如果返回了几万字，全部放进消息列表，后面每一步都要为这几万字付费。截断到 3000 字符，并告诉模型"内容太长，已截断"，它就知道要换个更精确的方式去查。
• 统计词元。智能体一次任务要调用好几次模型，每次的输入都包含之前所有的步骤，费用增长得比普通对话快，必须心里有数。

model 是一个参数，而不是写死的 API 调用。这是为了下一步。

先用假模型测试

智能体的行为由模型决定，每次都不一样，这让测试循环本身变得很麻烦：你没法确定是循环写错了，还是模型这次做了奇怪的决定。而且每测一次都要花钱。

解决办法是写一个"假模型"：它不调用任何 API，只按照一个写好的剧本，依次返回预先定好的工具调用：

class ScriptedModel:
    """按预先写好的剧本依次返回。用来在不调用 API 的情况下测试循环本身。"""

    def __init__(self, script):
        self.script = list(script)

    def __call__(self, messages, tools):
        return self.script.pop(0), None


SCRIPT = [
    Message(tool_calls=[Call("c1", "grep_docs", '{"keyword": "pool timeout"}')]),
    Message(tool_calls=[Call("c2", "read_doc", '{"path": "advanced/timeouts.md", "start": 1, "end": 200}')]),
    Message(tool_calls=[Call("c3", "read_doc", '{"path": "advanced/timeout.md"}')]),  # 故意写错文件名
    Message(content="httpx 有四种超时：connect、read、write、pool（见 advanced/timeouts.md 第 43～62 行）。"),
]

Message 和 Call 是两个小的数据类，结构和 OpenAI SDK 返回的对象一样（有 content、tool_calls、call.function.name 这些属性，完整定义见 code/05-agents/agent_loop.py），所以循环分不出它面对的是真模型还是假模型。

剧本里故意安排了几种情况：一次搜不到结果的搜索、一次要读 200 行（超过工具的 80 行上限）、一次写错的文件名。运行：

python agent_loop.py

[第 1 步] grep_docs({"keyword": "pool timeout"}) → 没有找到 pool timeout
[第 2 步] read_doc({"path": "advanced/timeouts.md", "start": 1, "end": 200}) → 1: HTTPX is careful to enforce timeouts everywhere by default. | 2:  | 3: The default beha
[第 3 步] read_doc({"path": "advanced/timeout.md"}) → 错误：没有这个文件 advanced/timeout.md，请先用 list_docs 查看有哪些文件
[第 4 步] 回答：
httpx 有四种超时：connect、read、write、pool（见 advanced/timeouts.md 第 43～62 行）。

共 4 步，输入 0 词元，输出 0 词元

每种情况都按预期处理了：搜不到返回"没有找到"，写错文件名返回一条错误说明，程序没有崩溃，最后正常结束。这个测试不花一分钱，每次运行结果完全一样，可以放进自动化测试里。以后你修改了循环（比如加上日志、改变截断规则），先用剧本跑一遍，确认没改坏。

换成真模型

class RealModel:
    def __init__(self):
        from openai import OpenAI
        self.client = OpenAI(api_key=os.environ["LLM_API_KEY"],
                             base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"))
        self.model = os.environ.get("LLM_MODEL", "deepseek-flash")

    def __call__(self, messages, tools):
        response = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=tools,
            extra_body={"thinking": {"type": "disabled"}})
        return response.choices[0].message, response.usage

python agent_loop.py --real

问题是"httpx 的超时分成哪几种？每一种管什么？请注明文档出处。"我运行的结果（你的步骤和措辞都会不一样）：

[第 1 步] grep_docs({"keyword": "timeout"}) → advanced/extensions.md:10: # Request timeouts actually implemented as an extension on | ad
[第 1 步] list_docs({}) → advanced/authentication.md | advanced/clients.md | advanced/event-hooks.md | advanced/exte
[第 2 步] read_doc({"path": "advanced/timeouts.md", "start": 1, "end": 80}) → 1: HTTPX is careful to enforce timeouts everywhere by default. | 2:  | 3: The default beha
[第 3 步] 回答：
httpx 的超时一共分为**四种**，文档依据在 `advanced/timeouts.md`（第 45–61 行）。

## 四种超时类型

| 类型 | 管什么 | 超时抛出的异常 |
|---|---|---|
| **connect** | 与目标主机建立 socket 连接的最长等待时间 | `ConnectTimeout` |
| **read** | 等待收到一块数据（例如响应体的一块）的最长时长 | `ReadTimeout` |
| **write** | 等待发送一块数据（例如请求体的一块）的最长时长 | `WriteTimeout` |
| **pool** | 从连接池中获取一个连接的最长等待时长 | `PoolTimeout` |
（后面还有逐条引用的原文和补充说明，省略）

共 3 步，输入 3742 词元，输出 666 词元

模型的决策过程是这样的：第 1 步，同时搜索关键词 "timeout"、列出所有文件（两个互不依赖的工具调用，一起发出）；看到有一个文件就叫 advanced/timeouts.md，第 2 步直接读它的前 80 行；第 3 步，信息够了，给出回答。

我打开 timeouts.md 核对了回答里的行号：四种超时的定义确实在第 45 到 61 行，connect 在第 48 到 50 行，pool 在第 57 到 61 行，都对得上。这得益于 read_doc 返回的内容带着行号。

整个过程用了 3 次模型调用、3742 个输入词元。对比一下：第 04 模块的 RAG 回答一个问题只调用 1 次模型，输入 1000 词元左右。智能体更灵活，但也更贵，下一课会专门比较。

智能体的执行轨迹

注意输出里每一步的记录：调用了什么工具、参数是什么、返回了什么。这叫执行轨迹（trace）。

智能体出了问题，最有效的排查办法就是看轨迹：它是不是一开始就搜错了关键词？是不是读了错误的文件？是不是信息已经够了还在继续查？没有轨迹，你只能看到一个错误的最终回答，完全不知道它是怎么走到那一步的。这里只是简单地打印出来，06 模块第 3 课会把它记录成结构化的日志。

常见问题

模型一直调用工具停不下来：检查 system 提示词有没有说清楚什么时候可以回答。max_steps 是最后的保险，但触发它说明提示词或者工具有问题。

模型编造了一个不存在的工具名：循环里已经处理了，会返回"没有叫某某的工具"。如果频繁发生，说明工具的说明书写得不清楚，模型不知道该用哪个。

上下文越来越长，越来越贵：智能体的每一步都要带上之前所有的步骤。控制工具返回的长度是最有效的办法。第 5 课会讲更多方法。

练习

1. 给剧本加一步：调用一个不存在的工具 search_web，确认循环能正确处理。再加一步参数类型错误的调用（比如 read_doc 的 start 传一个字符串 "abc"），看看会返回什么。
2. 把 max_steps 改成 2，用 --real 运行，看看会发生什么。
3. 写一个新工具 count_lines(path)，返回一个文档文件有多少行，用装饰器注册。问真模型"httpx 的哪个文档文件最长"，看它会不会用这个新工具。
4. 用 --real 问一个文档里没有答案的问题（比如"httpx 支持 HTTP/3 吗？"），看它查了几步、最后怎么回答。

自测

1. 智能体循环有哪两种停止条件？为什么两种都需要？

一种是模型不再调用工具，说明它给出了最终回答。另一种是达到最大步数。第二种是保险：模型可能陷入反复调用工具的循环，没有上限的话会一直花钱，程序也不会结束。

2. 工具执行出错时，为什么要把错误信息交给模型，而不是直接抛出异常？

抛出异常会让整个任务失败。把错误信息作为观察结果交给模型，模型通常能根据错误自己改正，比如换个文件名、先调用 list_docs 看看有哪些文件。错误信息最好写给模型看，不只说错了，还说明下一步该怎么做。

3. 用"脚本模型"测试智能体有什么好处？它能测出什么、测不出什么？

好处是不花钱、结果每次都一样，可以放进自动化测试。它能测出循环本身的问题：工具执行、错误处理、截断、停止条件。它测不出模型会不会做出正确的决策，那需要用真模型和评估集来测。