Modul 05 · Lektion 2

Eine Agentenschleife von Hand schreiben

Ohne Framework einen Agenten in gut hundert Zeilen schreiben – Tool-Registrierung, Aufrufschleife, Stoppbedingungen, Fehlerbehandlung. Erst offline mit einem Scheinmodell testen, das ein Drehbuch abspielt, dann ein echtes Modell einsetzen und zusehen, wie es selbst entscheidet, was es nachschlägt.

Etwa 50 Minuten
Niveau: Fortgeschritten
Getestet: 2026-09-14 deepseek-flash

Code und Programmausgaben stehen genau so da, wie sie gelaufen sind – Kommentare und Ausgaben sind daher auf Chinesisch.

Um das Wort „Agent“ wird viel Geheimnis gemacht, als wäre es eine völlig neue Technik. Den Kern hast du aber schon in Modul 03, Lektion 3 geschrieben: die Schleife „Modell aufrufen, bei einem Tool-Aufruf das Tool ausführen, das Ergebnis zurückgeben, Modell erneut aufrufen“.

Die Schleife jener Lektion fragte PyPI einmal ab und war fertig. Diese Lektion macht daraus einen echten Agenten: Er bekommt einige Tools zum Nachschlagen in der Dokumentation und entscheidet selbst, was er zuerst und was danach nachschlägt und wann er antworten kann. Das ganze Programm kommt ohne Framework aus, gut hundert Zeilen. Wenn du danach Frameworks wie LangGraph oder das OpenAI Agents SDK ansiehst, wirst du merken, dass sie alle Dinge um diese Schleife herum ergänzen.

Woraus ein Agent besteht

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

Fünf Dinge:

Nachrichtenliste: das Protokoll des gesamten Ablaufs und zugleich alles, was das Modell bei jedem Schritt sieht.
Tools: Funktionen, die das Modell aufrufen kann, samt der Beschreibung für das Modell.
Schleife: Modell aufrufen, Tools ausführen, Ergebnisse zurückgeben, wiederholen.
Stoppbedingungen: Das Modell ruft keine Tools mehr auf, oder die Höchstzahl an Schritten ist erreicht.
Umgang mit Beobachtungen: wie Rückgabewerte und Fehlermeldungen der Tools zu Text werden, den das Modell lesen kann, und was bei zu langen Ergebnissen passiert.

Wir schreiben sie der Reihe nach.

Tools: Funktion plus Beschreibung

In Modul 03, Lektion 3 haben wir die JSON-Beschreibung eines Tools von Hand geschrieben, gut zehn Zeilen pro Tool. Bei vielen Tools wird das mühsam; diesmal schreiben wir einen Dekorator, der die Beschreibung automatisch aus der Funktion erzeugt:

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

Er liest die Parameterliste der Funktion: Parameternamen werden zu Eigenschaften des JSON Schema, mit int annotierte Parameter bekommen den Typ integer, alles andere gilt als string, und Parameter ohne Standardwert sind Pflicht. Das ist eine vereinfachte Version, die nur Strings und Ganzzahlen unterstützt; für diese Lektion reicht es.

Der Agent bekommt drei Tools, alle für die httpx-Dokumentation:

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

Beachte, dass diese drei Tools nichts mit dem RAG aus Modul 04 zu tun haben: keine Vektoren, kein Suchalgorithmus, nur das Schlichteste – „Dateien auflisten, Stichwort suchen, zeilenweise lesen“. Die Intelligenz der Suche liegt ganz beim Modell: Es entscheidet, nach welchem Wort es sucht und welche Zeilen welcher Datei es liest. So schlägt ein Mensch in Dokumentation nach, und so gehen auch Programmieragenten wie Claude Code oder Cursor durch Code.

Einige Details sind Absicht:

grep_docs liefert höchstens 20 Treffer, read_doc liest höchstens 80 Zeilen auf einmal. Zu viel Rückgabe füllt den Kontext rasch und lässt das Modell das Wesentliche verlieren.
Jede Zeile, die read_doc zurückgibt, trägt ihre Zeilennummer, damit das Modell in der Antwort die Quelle genau angeben kann.
read_doc prüft den Pfad und erlaubt keine Dateien außerhalb des Dokumentationsverzeichnisses. Übergibt das Modell (oder ein manipuliertes Modell) ../../../etc/passwd, liegt der Pfad nach resolve() nicht in DOCS und wird sofort abgelehnt. Lektion 8 erklärt, warum das so wichtig ist.
Fehlermeldungen sind für das Modell geschrieben: „Diese Datei gibt es nicht, bitte erst mit list_docs nachsehen, welche Dateien es gibt.“ Sie sagen nicht nur, dass etwas falsch ist, sondern auch, was als Nächstes zu tun ist.

Die Schleife

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

Verglichen mit der Schleife aus Modul 03, Lektion 3 kommen diese Dinge hinzu:

Zwei Stoppbedingungen. Ruft das Modell keine Tools mehr auf, hat es die endgültige Antwort gegeben; ist max_steps erreicht und es noch nicht fertig, wird zwangsweise gestoppt. Letzteres ist eine unverzichtbare Absicherung: Das Modell kann in eine Schleife geraten, in der es immer wieder dasselbe Tool aufruft, und jeder Schritt kostet Geld.
Jeder Fehler wird zur Beobachtung. Erfindet das Modell einen nicht existierenden Tool-Namen (KeyError), übergibt falsche Argumente (TypeError) oder kann eine Datei nicht gelesen werden, stürzt das Programm nicht ab; stattdessen geht eine Nachricht „Fehler: …“ an das Modell zurück. Sieht das Modell den Fehler, korrigiert es sich meist selbst.
Zu lange Ergebnisse kürzen. Liefert ein Tool Zehntausende Zeichen und landet das alles in der Nachrichtenliste, bezahlt man sie bei jedem weiteren Schritt mit. Auf 3000 Zeichen kürzen und dem Modell sagen „Inhalt zu lang, gekürzt“, dann weiß es, dass es genauer nachschlagen muss.
Tokens zählen. Ein Agent ruft für eine Aufgabe das Modell mehrmals auf, und jede Eingabe enthält alle bisherigen Schritte; die Kosten wachsen schneller als bei einem normalen Gespräch, das muss man im Blick haben.

model ist ein Parameter, kein fest verdrahteter API-Aufruf. Das ist für den nächsten Schritt.

Erst mit einem Scheinmodell testen

Das Verhalten eines Agenten bestimmt das Modell, und es ist jedes Mal anders. Das macht das Testen der Schleife selbst mühsam: Man weiß nicht, ob die Schleife falsch ist oder das Modell diesmal eine seltsame Entscheidung getroffen hat. Außerdem kostet jeder Test Geld.

Die Lösung ist ein „Scheinmodell“: Es ruft keine API auf, sondern liefert nach einem festen Drehbuch der Reihe nach vorab festgelegte Tool-Aufrufe:

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 und Call sind zwei kleine Datenklassen mit derselben Struktur wie die Objekte, die das OpenAI SDK liefert (mit Attributen wie content, tool_calls, call.function.name; vollständige Definition in code/05-agents/agent_loop.py). Die Schleife kann also nicht unterscheiden, ob sie mit einem echten oder einem Scheinmodell arbeitet.

Im Drehbuch sind absichtlich einige Fälle eingebaut: eine Suche ohne Treffer, ein Lesevorgang über 200 Zeilen (mehr als die 80-Zeilen-Grenze des Tools) und ein falsch geschriebener Dateiname. Ausführen:

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

Jeder Fall wurde wie erwartet behandelt: Die Suche ohne Treffer liefert „nichts gefunden“, der falsche Dateiname eine Fehlererklärung, das Programm stürzt nicht ab und endet normal. Dieser Test kostet keinen Cent, liefert bei jedem Lauf genau dasselbe Ergebnis und kann in die automatisierten Tests. Wenn du später die Schleife änderst (etwa Logging hinzufügst oder die Kürzungsregel änderst), lass zuerst das Drehbuch laufen, um sicherzugehen, dass nichts kaputt ist.

Ein echtes Modell einsetzen

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

Die Frage lautet „Welche Arten von Timeouts hat httpx? Wofür ist jede zuständig? Bitte die Quelle in der Dokumentation angeben.“ Mein Ergebnis (deine Schritte und Formulierungen werden anders sein):

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

Die Entscheidungen des Modells: In Schritt 1 sucht es nach dem Stichwort „timeout“ und listet zugleich alle Dateien auf (zwei voneinander unabhängige Tool-Aufrufe, gemeinsam abgeschickt); es sieht eine Datei namens advanced/timeouts.md und liest in Schritt 2 direkt deren erste 80 Zeilen; in Schritt 3 reichen die Informationen, und es antwortet.

Ich habe timeouts.md geöffnet und die Zeilennummern der Antwort nachgeprüft: Die Definitionen der vier Timeouts stehen tatsächlich in den Zeilen 45 bis 61, connect in 48 bis 50, pool in 57 bis 61, alles stimmt. Das verdankt sich den Zeilennummern im Rückgabetext von read_doc.

Der ganze Ablauf brauchte 3 Modellaufrufe und 3742 Eingabe-Tokens. Zum Vergleich: Das RAG aus Modul 04 ruft das Modell pro Frage nur 1-mal auf, mit etwa 1000 Eingabe-Tokens. Der Agent ist flexibler, aber auch teurer; die vorige Lektion hat das direkt verglichen.

Die Ausführungsspur des Agenten

Achte in der Ausgabe auf das Protokoll jedes Schritts: welches Tool aufgerufen wurde, mit welchen Argumenten, was zurückkam. Das nennt man Ausführungsspur (Trace).

Hat ein Agent ein Problem, ist die wirksamste Diagnose, die Spur anzusehen: Hat er gleich zu Beginn nach dem falschen Stichwort gesucht? Die falsche Datei gelesen? Weiter gesucht, obwohl die Informationen schon reichten? Ohne Spur siehst du nur eine falsche Endantwort und weißt überhaupt nicht, wie er dorthin gekommen ist. Hier wird sie nur einfach ausgegeben; Modul 06, Lektion 3 zeichnet sie als strukturiertes Log auf.

Häufige Probleme

Das Modell ruft immer weiter Tools auf und hört nicht auf: Prüf, ob der System-Prompt klar sagt, wann geantwortet werden darf. max_steps ist die letzte Absicherung; wird sie ausgelöst, stimmt etwas mit dem Prompt oder den Tools nicht.

Das Modell erfindet einen nicht existierenden Tool-Namen: Die Schleife behandelt das bereits und antwortet „kein Tool namens …“. Passiert es häufig, sind die Tool-Beschreibungen unklar, und das Modell weiß nicht, welches es nehmen soll.

Der Kontext wird immer länger und teurer: Jeder Schritt des Agenten trägt alle vorherigen Schritte mit. Die Rückgabelänge der Tools zu begrenzen ist das wirksamste Mittel. Lektion 5 behandelt weitere.

Übungen

Füg dem Drehbuch einen Schritt hinzu: den Aufruf eines nicht existierenden Tools search_web, und prüfe, dass die Schleife das richtig behandelt. Füg dann einen Aufruf mit falschem Argumenttyp hinzu (etwa für start von read_doc den String "abc") und sieh nach, was zurückkommt.
Setz max_steps auf 2, starte mit --real und sieh, was passiert.
Schreib ein neues Tool count_lines(path), das zurückgibt, wie viele Zeilen eine Dokumentationsdatei hat, und registriere es mit dem Dekorator. Frag das echte Modell „Welche Dokumentationsdatei von httpx ist die längste?“ und sieh, ob es das neue Tool nutzt.
Stell mit --real eine Frage, deren Antwort nicht in der Dokumentation steht (etwa „Unterstützt httpx HTTP/3?“), und sieh, wie viele Schritte es sucht und was es am Ende antwortet.

Selbsttest

1. Welche zwei Stoppbedingungen hat die Agentenschleife? Warum braucht man beide?

Die eine: Das Modell ruft keine Tools mehr auf, hat also seine endgültige Antwort gegeben. Die andere: Die Höchstzahl an Schritten ist erreicht. Die zweite ist eine Absicherung: Das Modell kann in eine Schleife wiederholter Tool-Aufrufe geraten; ohne Obergrenze würde es immer weiter Geld kosten, und das Programm würde nicht enden.

2. Warum gibt man bei einem Fehler in der Tool-Ausführung die Fehlermeldung an das Modell weiter, statt direkt eine Ausnahme zu werfen?

Eine Ausnahme ließe die ganze Aufgabe scheitern. Gibt man die Fehlermeldung als Beobachtung an das Modell, kann es sich anhand des Fehlers meist selbst korrigieren, etwa einen anderen Dateinamen versuchen oder zuerst list_docs aufrufen, um die vorhandenen Dateien zu sehen. Die Fehlermeldung sollte für das Modell geschrieben sein: nicht nur sagen, dass etwas falsch ist, sondern auch, was als Nächstes zu tun ist.

3. Welche Vorteile hat das Testen eines Agenten mit einem „Skript-Modell“? Was deckt es auf, was nicht?

Es kostet nichts, liefert jedes Mal dasselbe Ergebnis und passt in automatisierte Tests. Es deckt Probleme der Schleife selbst auf: Tool-Ausführung, Fehlerbehandlung, Kürzung, Stoppbedingungen. Ob das Modell richtige Entscheidungen trifft, deckt es nicht auf; das braucht ein echtes Modell und einen Evaluationsdatensatz.