エージェントループを手書きする

「エージェント」（agent）という言葉は、何かまったく新しい技術であるかのように神秘的に語られがちです。実は、その核心はモジュール 03 第 3 課ですでに書いています。「モデルを呼び出し、ツール呼び出しがあれば実行し、結果を戻し、またモデルを呼び出す」というあのループです。

あの課のループは PyPI を 1 回調べて終わりでした。この課ではそれを本物のエージェントにします。ドキュメントを調べるツールをいくつか与え、まず何を調べ、次に何を調べ、いつ答えてよいかを自分で決めさせるのです。プログラム全体はフレームワークを使わず、百数十行です。書き終えてから 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 文字に切り詰め、「内容が長すぎるので切り詰めた」とモデルに伝えれば、もっと的を絞ったやり方で調べ直すべきだとわかります。
• トークンを集計する。エージェントの 1 回のタスクではモデルを何度も呼び、毎回の入力にそれまでのすべてのステップが含まれるので、費用は普通の会話より速く増えます。把握しておく必要があります。

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 問に答えるのにモデルを 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. 「台本どおりのモデル」でエージェントをテストする利点は何ですか？何がわかり、何がわからないのですか？

お金がかからず、結果が毎回同じなので、自動テストに入れられます。ループ自体の問題はわかります。ツールの実行、エラー処理、切り詰め、停止条件です。モデルが正しい判断をするかどうかはわからず、それには本物のモデルと評価セットでテストする必要があります。