Writing an agent loop by hand

The word "agent" gets talked about as if it were something mysterious, some entirely new technology. In fact you already wrote its core in Module 03, Lesson 3: the loop that "calls the model, runs any tool calls, puts the results back, and calls the model again".

That lesson's loop looked up PyPI once and was done. This lesson turns it into a real agent: give it a few tools for searching the docs and let it decide for itself what to look up first, what next, and when it can answer. The whole program uses no framework and is a little over a hundred lines. Once it's written, look at frameworks like LangGraph or the OpenAI Agents SDK and you'll see they all add things around this loop.

What an agent is made of

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

Five things:

1. The message list: the record of the whole process, and everything the model can see at each step.
2. Tools: functions the model can call, plus the descriptions the model reads.
3. The loop: call the model, run tools, put results back, repeat.
4. Stop conditions: the model stops calling tools, or the step limit is reached.
5. Handling observations: how tool return values and error messages become text the model can read, and what to do when they're too long.

Let's write them one at a time.

Tools: a function plus a description

In Module 03, Lesson 3 we wrote the JSON description for a tool by hand, a dozen lines per tool. That gets tedious with many tools, so this time we write a decorator that generates the description from the function itself:

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

It reads the function's parameter list: parameter names become JSON Schema property names, parameters annotated int have type integer, everything else is treated as string, and parameters without defaults are required. This is a simplified version that supports only strings and integers, which is enough for this lesson.

The agent gets three tools, all for working with the httpx docs:

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))

Notice how different these three tools are from Module 04's RAG: no vectors, no retrieval algorithm, just the plainest "list files, search keywords, read by line". All the intelligence of retrieval is handed to the model: it decides what words to search for and which lines of which file to read. This is really how people look things up in docs, and also how coding agents like Claude Code and Cursor navigate code.

Several details are deliberate:

• grep_docs returns at most 20 matches and read_doc reads at most 80 lines at a time. Tools that return too much quickly fill the context and make it hard for the model to find what matters.
• Every line read_doc returns carries its line number, so the model can cite its source precisely when answering.
• read_doc checks the path and refuses to read files outside the docs directory. If the model (or a model someone is manipulating) passes ../../../etc/passwd, the path after resolve() isn't inside DOCS, and it is refused outright. Lesson 8 covers why this matters.
• Error messages are written for the model: "no such file; use list_docs first to see which files exist". It doesn't just say something went wrong; it tells the model what to do next.

The loop

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

Compared with the loop in Module 03, Lesson 3, there are a few additions:

• Two stop conditions. When the model stops calling tools, it has given its final answer; if it reaches max_steps without finishing, it's stopped forcibly. The latter is an essential safeguard: the model can get stuck calling the same tool over and over, and every step costs money.
• Every error becomes an observation. If the model invents a tool name that doesn't exist (KeyError), passes the wrong arguments (TypeError), or a file can't be read, the program doesn't crash; the error becomes an "Error: …" message handed back to the model. On seeing the error, the model usually corrects itself.
• Overlong results are truncated. If a tool returns tens of thousands of characters and all of it goes into the message list, every later step pays for them. Truncate to 3,000 characters and tell the model "content too long, truncated", and it knows to look things up in a more precise way.
• Token counting. One agent task calls the model several times, and each call's input includes all the earlier steps, so cost grows faster than in an ordinary conversation. You need to keep an eye on it.

model is a parameter rather than a hard-coded API call. That's for the next step.

Testing with a fake model first

The agent's behaviour is decided by the model and differs every time, which makes testing the loop itself awkward: you can't tell whether the loop is wrong or the model just made an odd decision this time. And every test costs money.

The fix is a "fake model": it calls no API and simply follows a script, returning predetermined tool calls one after another:

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 and Call are two small data classes with the same shape as the objects the OpenAI SDK returns (they have content, tool_calls, call.function.name and so on; full definitions in code/05-agents/agent_loop.py), so the loop can't tell whether it's talking to a real model or a fake one.

The script deliberately includes several cases: a search that finds nothing, a read of 200 lines (over the tool's 80-line limit), and a misspelled file name. Running it:

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 词元

Every case was handled as expected: the empty search returned "not found", the wrong file name returned an error explanation, the program didn't crash, and it finished normally. This test costs nothing, gives exactly the same result every run, and can go into automated tests. Whenever you change the loop later (adding logging, changing the truncation rules), run the script first to make sure nothing broke.

Switching to a real model

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

The question is "What kinds of timeout does httpx have, and what does each control? Please cite the docs." My run (your steps and wording will differ):

[第 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 词元

The model's decisions went like this: in step 1 it searched for the keyword "timeout" and listed all files at the same time (two independent tool calls sent together); seeing a file called advanced/timeouts.md, in step 2 it read its first 80 lines directly; in step 3 it had enough information and answered.

I opened timeouts.md and checked the line numbers in the answer: the definitions of the four timeouts really are on lines 45 to 61, connect on 48 to 50 and pool on 57 to 61, all matching. That's thanks to read_doc returning content with line numbers.

The whole process took 3 model calls and 3,742 input tokens. For comparison, Module 04's RAG answers a question with 1 model call and about 1,000 input tokens. The agent is more flexible, but also more expensive; the next lesson compares the two directly.

The agent's execution trace

Notice the record of each step in the output: which tool was called, with what arguments, and what came back. This is called the execution trace.

When an agent goes wrong, the most effective way to investigate is to read the trace: did it search for the wrong keyword from the start? Read the wrong file? Keep searching after it already had enough? Without a trace you see only a wrong final answer and have no idea how it got there. Here it's simply printed; Module 06, Lesson 3 records it as structured logs.

Common problems

The model keeps calling tools and won't stop: check whether the system prompt makes clear when it may answer. max_steps is the last safeguard, but hitting it means something is wrong with the prompt or the tools.

The model invents a tool name that doesn't exist: the loop already handles this and returns "no tool called such-and-such". If it happens often, the tool descriptions aren't clear and the model doesn't know which to use.

The context keeps growing, and so does the cost: every step of an agent carries all the earlier steps. Limiting the length of tool results is the most effective remedy. Lesson 5 covers more methods.

Exercises

1. Add a step to the script that calls a nonexistent tool, search_web, and confirm the loop handles it correctly. Add another call with an argument of the wrong type (such as passing the string "abc" as read_doc's start) and see what comes back.
2. Set max_steps to 2, run with --real, and see what happens.
3. Write a new tool count_lines(path) that returns how many lines a doc file has, and register it with the decorator. Ask the real model "which httpx doc file is the longest?" and see whether it uses the new tool.
4. Use --real to ask a question the docs don't answer (such as "does httpx support HTTP/3?"), and see how many steps it takes and how it finally answers.

Self-check

1. What are the two stop conditions of an agent loop, and why are both needed?

One is the model no longer calling tools, which means it has given its final answer. The other is reaching the maximum number of steps. The second is a safeguard: the model can get stuck calling tools repeatedly, and without a limit it would keep spending money and the program would never end.

2. When a tool fails, why hand the error message to the model instead of raising an exception?

Raising an exception fails the whole task. Given the error as an observation, the model can usually correct itself, for example by trying another file name or calling list_docs first to see which files exist. Error messages are best written for the model, saying not just that something went wrong but what to do next.

3. What are the benefits of testing an agent with a "scripted model"? What can and can't it catch?

It costs nothing and gives the same result every time, so it can go into automated tests. It catches problems in the loop itself: running tools, error handling, truncation, stop conditions. It can't tell whether the model will make the right decisions; that needs a real model and an evaluation set.