手寫一個智慧體迴圈

"智慧體"（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. 用"指令碼模型"測試智慧體有什麼好處？它能測出什麼、測不出什麼？

好處是不花錢、結果每次都一樣，可以放進自動化測試。它能測出迴圈本身的問題：工具執行、錯誤處理、截斷、停止條件。它測不出模型會不會做出正確的決策，那需要用真模型和評估集來測。