Construis ton premier agent IA (safe, avec code)

Le problème

Ton premier agent va essayer de tout faire.

Il va naviguer sur le web, écrire dans des bases, “optimiser” des configs… bref, se comporter comme un stagiaire avec un accès root.

Ne commence pas comme ça.

Commence avec un seul tool, read-only, avec des budgets et des logs.

Pourquoi ça compte dans la vraie vie

La première fois que tu shippes du tool calling, tu vas découvrir :

• des timeouts manquants
• une idempotence inexistante
• des logs “présents”, mais inutiles

Autant le découvrir avec un tool plutôt qu’avec douze.

Le playbook (20 minutes, sans héroïsme)

1. Choisis un seul tool read-only (web.search ou kb.search).
2. Ajoute des budgets : steps + time (+ $ si tu peux).
3. Logge chaque tool call (nom, hash des args, durée, statut).
4. Détecte les boucles via les args identiques.
5. N’ajoute pas encore de tools “send” ou “write”.

Ce que “un seul tool” veut vraiment dire

“Un seul tool”, ce n’est pas “un tool + une porte dérobée”.

Ça veut dire :

• tu peux pointer l’allowlist et la compter sur une main
• pas de tool générique http.request avec internet complet
• pas de run_shell “pour débug”

Si ton premier agent peut fetch n’importe quelle URL, tu as déjà sauté la phase “safe”.

Étape 0 : choisis une tâche qui ne va pas te ruiner

Bonnes premières tâches :

• “trouve 5 docs pertinents dans notre KB”
• “résume les incidents récents depuis le dossier de postmortems”
• “rédige une réponse en utilisant des templates connus”

Mauvaises premières tâches :

• “browse le web jusqu’à être confiant”
• “corrige la config prod”
• “exécute des commandes sur des serveurs”

Ton premier agent doit être boring et pas cher.

Étape 1 : budgets (ce qui transforme une boucle en système)

Les budgets ne sont pas un “nice to have”. C’est la différence entre :

• une requête qui s’arrête
• une requête qui continue de te faire payer

On met au minimum :

• max steps
• max seconds

Si tu peux estimer le coût en $, ajoute-le tôt. Ça se rentabilise.

Étape 2 : logge la trace d’actions

Tu n’as pas besoin d’un système de tracing parfait le premier jour. Tu as besoin d’assez pour répondre :

• quels tools a-t-il appelés ?
• avec quels arguments ?
• combien de temps chaque call a pris ?
• pourquoi il s’est arrêté ?

Si tu ne peux pas répondre à ça, tu ne peux pas le shipper.

Étape 2.5 : rends les tools déterministes (sinon tu ne peux pas débug)

Un agent, c’est une boucle. Les boucles sont déjà assez pénibles. Ajoute des tools non déterministes (timeouts, APIs flaky, pages qui changent) et tu as gagné. Bravo : tu as construit un système non déterministe et tu vas le débugger à 03:00.

Le truc qu’on utilise tôt : record/replay. Pour un run, enregistre :

• le nom du tool
• le hash des args
• la réponse (ou l’erreur)

Ensuite tu peux rejouer le même “monde” pour tester :

• changements de prompt
• changements de modèle
• réglages du loop guard

Code conceptuel minimal :

class TapeTools(SafeTools):
  def __init__(self, allow: set[str], impl: dict, tape: list[dict] | None = None):
      super().__init__(allow=allow, impl=impl)
      self.tape = tape if tape is not None else []

  def call(self, name: str, *, args: dict[str, Any], request_id: str) -> Any:
      try:
          out = super().call(name, args=args, request_id=request_id)
          self.tape.append({"tool": name, "args": args, "ok": True, "out": out})
          return out
      except Exception as e:
          self.tape.append({"tool": name, "args": args, "ok": False, "err": str(e)})
          raise


def replay(tape: list[dict]):
  # Replace real tools with deterministic playback.
  i = 0
  def impl(name: str, **args):
      nonlocal i
      item = tape[i]
      i += 1
      assert item["tool"] == name
      if not item["ok"]:
          raise RuntimeError(item["err"])
      return item["out"]
  return impl

export class TapeTools extends SafeTools {
constructor({ allow, impl, tape }) {
  super({ allow, impl });
  this.tape = tape || [];
}

async call(name, { args, requestId }) {
  try {
    const out = await super.call(name, { args, requestId });
    this.tape.push({ tool: name, args, ok: true, out });
    return out;
  } catch (e) {
    this.tape.push({
      tool: name,
      args,
      ok: false,
      err: String(e && e.message ? e.message : e),
    });
    throw e;
  }
}
}

export function replay(tape) {
let i = 0;
return async function call(name, args) {
  const item = tape[i];
  i += 1;
  if (!item || item.tool !== name) throw new Error("tape mismatch");
  if (!item.ok) throw new Error(item.err);
  return item.out;
};
}

C’est glamour ? Non. Est-ce que ça te permet de lancer une “golden suite” de 20 tâches avant de shipper ? Oui. C’est comme ça que tu évites d’apprendre les régressions via de vrais clients.

Code (minimal, mais “prod‑shaped”)

from dataclasses import dataclass
import time
from typing import Any


@dataclass(frozen=True)
class Budget:
  max_steps: int = 12
  max_seconds: int = 20


class Tools:
  def __init__(self):
      self.allow = {"web.search"}

  def call(self, name: str, *, args: dict[str, Any]) -> Any:
      if name not in self.allow:
          raise RuntimeError(f"tool not allowed: {name}")
      return tool_impl(name, args=args)  # (pseudo)


def first_agent(question: str, *, tools: Tools, budget: Budget) -> str:
  started = time.time()
  last = None

  for step in range(budget.max_steps):
      if time.time() - started > budget.max_seconds:
          return "stopped: time budget exceeded"

      action = llm_pick_action(question)  # (pseudo): either {"tool":..., "args":...} or {"finish":...}

      if action.get("finish"):
          return action["finish"]

      key = f"{action['tool']}:{action['args']}"
      if key == last:
          return "stopped: loop guard (same call twice)"
      last = key

      obs = tools.call(action["tool"], args=action["args"])
      question = update_state(question, action, obs)  # (pseudo)

  return "stopped: step budget exceeded"

export class Tools {
constructor() {
  this.allow = new Set(["web.search"]);
}

async call(name, { args }) {
  if (!this.allow.has(name)) throw new Error("tool not allowed: " + name);
  return toolImpl(name, { args }); // (pseudo)
}
}

export async function firstAgent(question, { tools, budget }) {
const started = Date.now();
let last = null;

for (let step = 0; step < budget.max_steps; step++) {
  if ((Date.now() - started) / 1000 > budget.max_seconds) {
    return "stopped: time budget exceeded";
  }

  const action = await llmPickAction(question); // (pseudo): { tool, args } or { finish }
  if (action.finish) return action.finish;

  const key = String(action.tool) + ":" + JSON.stringify(action.args);
  if (key === last) return "stopped: loop guard (same call twice)";
  last = key;

  const obs = await tools.call(action.tool, { args: action.args });
  question = updateState(question, action, obs); // (pseudo)
}

return "stopped: step budget exceeded";
}

Rends ça réel (un petit skeleton exécutable)

L’exemple minimal au-dessus est volontairement court. Voici un skeleton “single file” un peu plus complet, qui ressemble à quelque chose que tu peux vraiment mettre derrière une API :

from dataclasses import dataclass
import hashlib
import json
import time
import uuid
from typing import Any, Callable


@dataclass(frozen=True)
class Budget:
  max_steps: int = 12
  max_seconds: int = 20


def args_hash(args: dict[str, Any]) -> str:
  raw = json.dumps(args, sort_keys=True, default=str).encode("utf-8")
  return hashlib.sha256(raw).hexdigest()[:12]


class LoopGuard(RuntimeError):
  pass


class Monitor:
  def __init__(self, *, max_repeat: int = 2):
      self.max_repeat = max_repeat
      self.counts: dict[str, int] = {}

  def mark(self, tool: str, args: dict[str, Any]) -> None:
      key = f"{tool}:{args_hash(args)}"
      self.counts[key] = self.counts.get(key, 0) + 1
      if self.counts[key] >= self.max_repeat:
          raise LoopGuard(f"loop guard: {key} repeated {self.counts[key]}x")


class SafeTools:
  def __init__(self, allow: set[str], impl: dict[str, Callable[..., Any]]):
      self.allow = allow
      self.impl = impl

  def call(self, name: str, *, args: dict[str, Any], request_id: str) -> Any:
      if name not in self.allow:
          raise RuntimeError(f"[{request_id}] tool not allowed: {name}")
      started = time.time()
      try:
          return self.impl[name](**args)
      finally:
          ms = int((time.time() - started) * 1000)
          print(f"[{request_id}] tool={name} ms={ms} args_hash={args_hash(args)}")


def run_first_agent(question: str, *, tools: SafeTools, budget: Budget) -> str:
  request_id = uuid.uuid4().hex
  started = time.time()
  monitor = Monitor(max_repeat=2)
  state = {"question": question, "notes": []}

  for step in range(budget.max_steps):
      if time.time() - started > budget.max_seconds:
          return f"[{request_id}] stopped: time budget"

      action = llm_pick_action(state)  # (pseudo)
      if action.get("finish"):
          return action["finish"]

      tool = action["tool"]
      args = action["args"]
      monitor.mark(tool, args)

      obs = tools.call(tool, args=args, request_id=request_id)
      state = update_state(state, action, obs)  # (pseudo)

  return f"[{request_id}] stopped: step budget"

import crypto from "node:crypto";

function stableStringify(obj) {
if (obj === null || typeof obj !== "object") return JSON.stringify(obj);
if (Array.isArray(obj)) return "[" + obj.map(stableStringify).join(",") + "]";
const keys = Object.keys(obj).sort();
return "{" + keys.map((k) => JSON.stringify(k) + ":" + stableStringify(obj[k])).join(",") + "}";
}

export function argsHash(args) {
return crypto.createHash("sha256").update(stableStringify(args)).digest("hex").slice(0, 12);
}

export class LoopGuard extends Error {}

export class Monitor {
constructor({ maxRepeat = 2 } = {}) {
  this.maxRepeat = maxRepeat;
  this.counts = new Map();
}

mark(tool, args) {
  const key = tool + ":" + argsHash(args);
  const next = (this.counts.get(key) || 0) + 1;
  this.counts.set(key, next);
  if (next >= this.maxRepeat) throw new LoopGuard("loop guard: " + key + " repeated " + next + "x");
}
}

export class SafeTools {
constructor({ allow, impl }) {
  this.allow = allow;
  this.impl = impl;
}

async call(name, { args, requestId }) {
  if (!this.allow.has(name)) throw new Error("[" + requestId + "] tool not allowed: " + name);
  const started = Date.now();
  try {
    return await this.impl[name](args);
  } finally {
    const ms = Date.now() - started;
    console.log("[" + requestId + "] tool=" + name + " ms=" + ms + " args_hash=" + argsHash(args));
  }
}
}

export async function runFirstAgent(question, { tools, budget }) {
const requestId = crypto.randomUUID().replace(/-/g, "");
const started = Date.now();
const monitor = new Monitor({ maxRepeat: 2 });
let state = { question, notes: [] };

for (let step = 0; step < budget.max_steps; step++) {
  if ((Date.now() - started) / 1000 > budget.max_seconds) return "[" + requestId + "] stopped: time budget";

  const action = await llmPickAction(state); // (pseudo)
  if (action.finish) return action.finish;

  const tool = action.tool;
  const args = action.args;
  monitor.mark(tool, args);

  const obs = await tools.call(tool, { args, requestId });
  state = updateState(state, action, obs); // (pseudo)
}

return "[" + requestId + "] stopped: step budget";
}

Ce n’est pas “le framework parfait”. C’est une petite boucle bornée avec :

• allowlisted tools
• budgets
• loop guard
• basic logs

Ça suffit pour apprendre sans cramer la prod.

Étape 3 : une petite “tool policy” (même en read-only)

Même des tools read-only doivent être allowlistés explicitement. Ça te force à rendre la surface d’attaque visible.

Tu peux faire évoluer la classe jouet Tools vers quelque chose d’opérable :

• add args hashing
• add timing
• add error classification (retryable vs fatal)

Ne sur-engineer pas. Rends juste ça débuggable.

Tool contracts (make the model play inside the lines)

The model will happily invent args:

• wrong field names
• huge strings
• weird nested objects

Si ton wrapper accepte “any dict”, ta runtime devient la couche de validation. Ça veut dire : tu vas débugger des erreurs de validation en prod.

Commence simple :

• define an args shape per tool
• validate types and bounds
• reject unknown fields

Exemple conceptuel :

def validate_web_search(args: dict[str, Any]) -> dict[str, Any]:
  q = str(args.get("q", "")).strip()
  if not q or len(q) > 200:
      raise ValueError("invalid q")

  k = int(args.get("k", 5))
  if k < 1 or k > 8:
      raise ValueError("invalid k")

  return {"q": q, "k": k}

export function validateWebSearch(args) {
const q = String((args && args.q) || "").trim();
if (!q || q.length > 200) throw new Error("invalid q");

const k = Number((args && args.k) ?? 5);
if (!Number.isFinite(k) || k < 1 || k > 8) throw new Error("invalid k");

return { q, k };
}

Ça a l’air bête. C’est aussi ce qui empêche ton agent d’appeler web.search avec un prompt dump de 10 000 caractères. Et ça te donne une erreur propre que tu peux traiter comme fatale (pas “on retry à l’infini”).

Étape 4 : un rollout qui ne fait pas mal

Voilà comment on shippe la première version :

1. Interne uniquement (toi + un collègue)
2. Read-only (pas de write tools)
3. Canary (petit slice de trafic)
4. Kill switch (arrêt opérateur)

Si tu sautes le canary, la première boucle sera chez ton client le plus important.

Étape 4.5 : annulation (arrête de payer quand l’utilisateur part)

Si tu as une UI, des utilisateurs vont abandonner des requêtes. Si tu ne propages pas l’annulation, l’agent continue quand même. Il continue d’appeler des tools. Il continue de brûler des tokens. Il continue de générer un output que personne ne lit.

Fais de l’annulation un stop reason de première classe :

• client disconnect → abort du run
• l’abort doit se propager aux appels modèle et aux tool calls
• logge stop_reason = client_cancel

Même une implémentation grossière vaut mieux que “on dépense jusqu’à tuer le budget”.

Étape 5 : métriques de succès (pour savoir si ça s’améliore)

Choisis 3 chiffres et suis-les :

• taux de complétion (est-ce que ça finit ?)
• coût moyen par run ($ ou tokens / crédits outils)
• p95 runtime

Si le taux de complétion est haut mais que le coût explose, tes budgets sont mal faits. Si le coût est bas mais que le taux de complétion est catastrophique, ton contrat de tool est probablement mauvais.

Où l’exécuter (serverless vs workers)

Ton premier agent va probablement tourner en API route. C’est ok… jusqu’à ce que ça ne le soit plus.

Deux pièges classiques :

• cold starts : cold start long + boucle = mauvais UX
• time limits : les plateformes serverless limitent l’exécution ; ton “budget 90 secondes” peut ne pas rentrer

Souvent on sépare :

• la requête arrive (rapide)
• le run agent tourne dans un worker (budgété)
• la UI poll / stream le progrès

Si ça sonne comme “trop d’archi”, garde simple : mets des budgets assez bas pour que ta plateforme puisse réellement arrêter le run. La manière la plus rapide d’avoir des 500, c’est une longue boucle dans un request handler sans garde-fous.

Hardening v2 (toujours petit, toujours safe)

Une fois la version “un tool” stable, voilà ce qu’on ajoute ensuite :

Hash des args + dédup

Si l’agent appelle le même tool avec les mêmes args en boucle, stoppe-le. Ça attrape les boucles infinies les plus bêtes.

Add per-tool budgets

Les budgets globaux arrêtent le run. Les budgets par tool empêchent une dépendance flaky de manger tout le run.

Example:

• web.search max 3 calls

Add a kill switch

Même pour un petit agent. S’il tourne en prod, tu veux pouvoir le stopper sans déployer.

Add a “stop reason”

Rends le stop reason visible dans les logs et la UI :

• time budget
• step budget
• loop detected
• tool denied

Ça évite que les utilisateurs martèlent refresh.

Le premier write tool (comment ne pas te planter)

Quand tu ajoutes enfin un write tool :

1. make it a separate tool (don’t reuse read tool name)
2. add idempotency keys
3. require approval by default
4. log an audit event for the intended write

Si tu sautes l’idempotence, tu vas shipper des écritures dupliquées. Ce n’est pas “peut‑être”. C’est “quand”.

Erreurs classiques du “premier agent” (on les a faites)

• Glissement “juste un tool de plus”. Tu commences avec web.search et tu finis avec http.request + un token admin.
• Retries sans limites. Tu “gères les erreurs” et tu construis une boucle infinie.
• Pas de stop reason dans la UI. Les gens refresh parce qu’ils ne savent pas ce qui s’est passé, et tu payes deux fois.
• Logger seulement la réponse finale. Ensuite tu ne peux pas expliquer pourquoi le tool a été appelé 17 fois.

Si tu évites ces quatre-là, tu es déjà devant la plupart des “agent demos”. Et : teste-le avec un faux tool qui échoue une fois. Si la boucle survit sans spammer les retries, c’est bon. Si ça s’écroule, fixe maintenant. Vraiment, maintenant.

Échec réel

On a vu un “premier agent” partir en prod avec 0 budgets. Il a reçu un prompt bizarre, a commencé à brows­er, puis s’est coincé.

Résultat :

• l’utilisateur a attendu ~2 minutes
• l’agent a brûlé de l’argent
• tout le monde a blâmé le modèle

Ce n’était pas le modèle. C’étaient les budgets absents.

Quand NE PAS construire un agent (pour l’instant)

• Si un script suffit, écris le script.
• Si un workflow suffit, construis le workflow.
• Si tu ne peux pas mettre des tool permissions + des audit logs, ne shippe pas du tool calling.

Ensuite

• Commencer : Qu’est‑ce qu’un agent IA ?
• Pattern : Research agent
• Failure mode : Boucle infinie