Écrire une boucle d'agent à la main

On entoure le mot « agent » (agent) de beaucoup de mystère, comme s'il s'agissait d'une technologie entièrement nouvelle. En réalité, vous en avez déjà écrit le cœur à la leçon 3 du module 03 : la boucle « appeler le modèle, exécuter l'outil s'il y a un appel d'outil, renvoyer le résultat, rappeler le modèle ».

La boucle de cette leçon-là interrogeait PyPI une fois et s'arrêtait. Cette leçon en fait un vrai agent : on lui donne quelques outils pour consulter la documentation, et il décide lui-même quoi chercher d'abord, quoi ensuite, et quand il peut répondre. Le programme n'utilise aucun framework, un peu plus de cent lignes. Ensuite, en regardant des frameworks comme LangGraph ou l'OpenAI Agents SDK, vous verrez qu'ils ajoutent tous des choses autour de cette boucle.

De quoi se compose un agent

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

Cinq choses :

1. La liste de messages : le journal de tout le déroulé, et aussi tout ce que le modèle voit à chaque étape.
2. Les outils : les fonctions que le modèle peut appeler, avec leur description destinée au modèle.
3. La boucle : appeler le modèle, exécuter les outils, renvoyer les résultats, recommencer.
4. Les conditions d'arrêt : le modèle n'appelle plus d'outil, ou le nombre maximal d'étapes est atteint.
5. Le traitement des observations : comment les valeurs renvoyées par les outils et les messages d'erreur deviennent un texte que le modèle peut lire, et que faire si c'est trop long.

Écrivons-les une par une.

Les outils : une fonction plus une description

À la leçon 3 du module 03, nous avons écrit à la main la description JSON d'un outil, une douzaine de lignes par outil. Avec beaucoup d'outils, c'est fastidieux ; cette fois, on écrit un décorateur qui génère automatiquement la description à partir de la fonction :

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

Il lit la liste des paramètres de la fonction : les noms de paramètres deviennent les propriétés du JSON Schema, un paramètre annoté int est de type integer, tout le reste est traité comme string, et un paramètre sans valeur par défaut est obligatoire. C'est une version simplifiée qui ne gère que les chaînes et les entiers, suffisante pour cette leçon.

L'agent reçoit trois outils, tous pour manipuler la documentation de 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))

Remarquez que ces trois outils n'ont rien à voir avec le RAG du module 04 : ni vecteurs ni algorithme de recherche, juste le plus élémentaire « lister les fichiers, chercher un mot-clé, lire par lignes ». Toute l'intelligence de la recherche est confiée au modèle : il décide quel mot chercher, quelles lignes de quel fichier lire. C'est la façon dont un humain consulte une documentation, et celle dont des agents de programmation comme Claude Code ou Cursor parcourent du code.

Plusieurs détails sont voulus :

• grep_docs renvoie au plus 20 résultats, read_doc lit au plus 80 lignes à la fois. Des outils qui renvoient trop remplissent vite le contexte et font perdre l'essentiel au modèle.
• Chaque ligne renvoyée par read_doc porte son numéro, pour que le modèle puisse citer précisément la source dans sa réponse.
• read_doc vérifie le chemin et interdit de lire des fichiers hors du dossier de documentation. Si le modèle (ou un modèle manipulé) passe ../../../etc/passwd, le chemin après resolve() n'est pas dans DOCS, et c'est refusé d'emblée. La leçon 8 explique pourquoi c'est si important.
• Les messages d'erreur sont écrits pour le modèle : « ce fichier n'existe pas, utilise d'abord list_docs pour voir quels fichiers existent ». Ils ne disent pas seulement que c'est faux, mais aussi quoi faire ensuite.

La boucle

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

Par rapport à la boucle de la leçon 3 du module 03, voici ce qui s'ajoute :

• Deux conditions d'arrêt. Si le modèle n'appelle plus d'outil, il a donné sa réponse finale ; si max_steps est atteint sans fin, on arrête de force. La seconde est une assurance indispensable : le modèle peut tomber dans une boucle où il appelle sans cesse le même outil, et chaque étape coûte de l'argent.
• Toute erreur devient une observation. Nom d'outil inventé (KeyError), mauvais paramètres (TypeError), fichier illisible : rien ne fait planter le programme, tout devient un message « Erreur : … » renvoyé au modèle. En voyant l'erreur, le modèle se corrige généralement lui-même.
• Tronquer les résultats trop longs. Si un outil renvoie des dizaines de milliers de caractères et que tout va dans la liste de messages, chaque étape suivante paie pour eux. En tronquant à 3000 caractères et en disant au modèle « contenu trop long, tronqué », il sait qu'il doit chercher de façon plus précise.
• Compter les tokens. Pour une tâche, un agent appelle plusieurs fois le modèle, chaque entrée contenant toutes les étapes précédentes ; le coût grimpe plus vite que dans une conversation ordinaire, il faut le garder à l'œil.

model est un paramètre, et non un appel d'API écrit en dur. C'est pour l'étape suivante.

Tester d'abord avec un faux modèle

Le comportement d'un agent est décidé par le modèle et change à chaque fois, ce qui complique le test de la boucle elle-même : impossible de savoir si c'est la boucle qui est fausse ou le modèle qui a pris une décision étrange cette fois. Et chaque test coûte de l'argent.

La solution : écrire un « faux modèle ». Il n'appelle aucune API et se contente de renvoyer, dans l'ordre, des appels d'outils fixés d'avance selon un scénario écrit :

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 et Call sont deux petites classes de données de même structure que les objets renvoyés par le SDK d'OpenAI (avec des attributs comme content, tool_calls, call.function.name ; définition complète dans code/05-agents/agent_loop.py), si bien que la boucle ne peut pas savoir si elle a affaire à un vrai modèle ou à un faux.

Le scénario prévoit volontairement plusieurs situations : une recherche sans résultat, une lecture de 200 lignes (au-delà de la limite de 80 de l'outil), un nom de fichier erroné. Exécution :

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

Chaque situation est traitée comme prévu : la recherche sans résultat renvoie « rien trouvé », le mauvais nom de fichier un message d'erreur explicatif ; le programme ne plante pas et se termine normalement. Ce test ne coûte pas un centime, donne exactement le même résultat à chaque exécution et peut entrer dans les tests automatisés. Quand vous modifierez la boucle (ajout de journaux, nouvelle règle de troncature), faites d'abord passer le scénario pour vérifier que rien n'est cassé.

Passer à un vrai modèle

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

La question : « Quels types de délais d'expiration httpx distingue-t-il ? À quoi sert chacun ? Indique la source dans la documentation. » Mon résultat (vos étapes et votre formulation seront différentes) :

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

Le processus de décision du modèle : à l'étape 1, il cherche le mot-clé « timeout » et liste tous les fichiers en même temps (deux appels d'outils indépendants, envoyés ensemble) ; voyant un fichier nommé advanced/timeouts.md, il en lit directement les 80 premières lignes à l'étape 2 ; à l'étape 3, il en sait assez et répond.

J'ai ouvert timeouts.md pour vérifier les numéros de ligne de la réponse : les définitions des quatre délais sont bien aux lignes 45 à 61, connect aux lignes 48 à 50, pool aux lignes 57 à 61, tout concorde. C'est grâce aux numéros de ligne dans le texte renvoyé par read_doc.

L'ensemble a demandé 3 appels de modèle et 3742 tokens d'entrée. Par comparaison, le RAG du module 04 n'appelle le modèle qu'une fois par question, avec environ 1000 tokens d'entrée. L'agent est plus souple, mais aussi plus cher ; la leçon précédente les a comparés.

La trace d'exécution de l'agent

Remarquez dans la sortie le journal de chaque étape : quel outil a été appelé, avec quels paramètres, et ce qui a été renvoyé. C'est la trace d'exécution (trace).

Quand un agent a un problème, le moyen le plus efficace de le diagnostiquer est de regarder la trace : a-t-il cherché le mauvais mot-clé dès le début ? Lu le mauvais fichier ? Continué à chercher alors qu'il en savait déjà assez ? Sans trace, vous ne voyez qu'une réponse finale fausse, sans aucune idée du chemin qui y a mené. Ici, on l'affiche simplement ; la leçon 3 du module 06 l'enregistre sous forme de journal structuré.

Problèmes courants

Le modèle appelle des outils sans jamais s'arrêter : vérifiez que le prompt system dit clairement quand il peut répondre. max_steps est l'ultime garde-fou, mais s'il se déclenche, c'est que le prompt ou les outils posent problème.

Le modèle invente un nom d'outil inexistant : la boucle le gère déjà et renvoie « aucun outil nommé … ». Si cela arrive souvent, les descriptions des outils sont peu claires et le modèle ne sait pas lequel utiliser.

Le contexte s'allonge et coûte de plus en plus cher : chaque étape de l'agent emporte toutes les précédentes. Limiter la longueur des retours d'outils est le moyen le plus efficace. La leçon 5 en présente d'autres.

Exercices

1. Ajoutez une étape au scénario : l'appel d'un outil inexistant search_web, et vérifiez que la boucle le gère correctement. Ajoutez ensuite un appel avec un mauvais type de paramètre (par exemple la chaîne "abc" pour le start de read_doc) et voyez ce qui est renvoyé.
2. Passez max_steps à 2, lancez avec --real et voyez ce qui se passe.
3. Écrivez un nouvel outil count_lines(path) qui renvoie le nombre de lignes d'un fichier de documentation, enregistré avec le décorateur. Demandez au vrai modèle « quel fichier de la documentation de httpx est le plus long ? » et voyez s'il utilise ce nouvel outil.
4. Avec --real, posez une question dont la réponse n'est pas dans la documentation (par exemple « httpx prend-il en charge HTTP/3 ? »), et voyez en combien d'étapes il cherche et ce qu'il répond au final.

Auto-test

1. Quelles sont les deux conditions d'arrêt de la boucle d'agent ? Pourquoi faut-il les deux ?

L'une : le modèle n'appelle plus d'outil, il a donc donné sa réponse finale. L'autre : le nombre maximal d'étapes est atteint. La seconde est une assurance : le modèle peut tomber dans une boucle d'appels d'outils répétés, et sans limite, il continuerait à dépenser de l'argent sans que le programme se termine.

2. Quand l'exécution d'un outil échoue, pourquoi transmettre le message d'erreur au modèle plutôt que lever directement une exception ?

Lever une exception ferait échouer toute la tâche. En transmettant le message d'erreur comme observation, le modèle peut généralement se corriger lui-même, par exemple essayer un autre nom de fichier ou appeler d'abord list_docs pour voir les fichiers existants. Le message d'erreur doit être écrit pour le modèle : pas seulement dire que c'est faux, mais aussi quoi faire ensuite.

3. Quels avantages a le test d'un agent avec un « modèle scénarisé » ? Que détecte-t-il, que ne détecte-t-il pas ?

Il ne coûte rien, donne toujours le même résultat et peut entrer dans les tests automatisés. Il détecte les problèmes de la boucle elle-même : exécution des outils, gestion des erreurs, troncature, conditions d'arrêt. Il ne détecte pas si le modèle prend les bonnes décisions ; pour cela, il faut un vrai modèle et un jeu d'évaluation.