Agent de code : lire un dépôt, planifier, éditer et lancer les tests

Les outils de lecture (sans danger)

L’exploration ne doit rien modifier. On expose des outils en lecture seule : lister, lire, chercher. Le périmètre minimal du deep-dive sécurité commence ici.

from langchain_core.tools import tool
from pathlib import Path

RACINE = Path("/sandbox/repo")        # le dépôt, dans un environnement isolé

@tool
def lister(chemin: str = ".") -> str:
    """Liste les fichiers et dossiers sous un chemin du dépôt."""
    base = (RACINE / chemin).resolve()
    if RACINE not in base.parents and base != RACINE:    # anti path-traversal
        return "Chemin hors du dépôt."
    return "\n".join(p.name for p in base.iterdir())

@tool
def lire(fichier: str) -> str:
    """Lit le contenu d'un fichier du dépôt (numéroté)."""
    f = (RACINE / fichier).resolve()
    if RACINE not in f.parents:
        return "Chemin hors du dépôt."
    return "\n".join(f"{i+1}: {l}" for i, l in enumerate(f.read_text().splitlines()))

@tool
def chercher(motif: str) -> str:
    """Recherche un motif (texte) dans le code du dépôt et renvoie les emplacements."""
    return grep_dans_repo(RACINE, motif)         # ripgrep côté implémentation

L’outil de test (l’évaluateur objectif)

Lancer les tests est une exécution de commande — donc à isoler. On exécute dans le sandbox, avec un timeout, et on renvoie la sortie brute (utile à l’agent).

import subprocess

@tool
def lancer_tests() -> str:
    """Lance la suite de tests du dépôt et renvoie le résultat (succès/échecs + logs)."""
    proc = subprocess.run(
        ["pytest", "-q"], cwd=RACINE, capture_output=True, text=True, timeout=120,
    )
    statut = "SUCCÈS" if proc.returncode == 0 else "ÉCHEC"
    return f"{statut}\n{proc.stdout[-4000:]}\n{proc.stderr[-2000:]}"

L’outil d’écriture (derrière garde-fou)

L’écriture ne propose qu’un diff ; elle ne l’applique pas directement. C’est ce diff qui passera devant l’humain avant d’être committé.

@tool
def proposer_diff(diff_unifie: str) -> str:
    """Propose un correctif sous forme de patch unifié (git diff). N'écrit rien encore."""
    if not diff_unifie.lstrip().startswith(("diff --git", "---")):
        return "Format invalide : fournis un patch unifié."
    return "Diff enregistré pour validation."     # stocké dans l'état par le nœud appelant

Nœud d’exploration : localiser le code

Un agent ReAct outillé en lecture seule explore le dépôt et identifie les fichiers en cause. On le borne (il ne doit pas explorer indéfiniment).

from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model

modele = init_chat_model("anthropic:claude-sonnet-4-6")
explorateur = create_react_agent(modele, [lister, lire, chercher])

def explorer(etat: Etat) -> dict:
    res = explorateur.invoke({"messages": [{"role": "user", "content":
        f"Localise le(s) fichier(s) concerné(s) par cette issue. Liste-les précisément.\n"
        f"Issue : {etat['issue']}"}]},
        config={"recursion_limit": 20})
    return {"fichiers_cibles": extraire_chemins(res["messages"][-1].content)}

Nœud d’édition : proposer un correctif

L’agent lit les fichiers cibles, comprend, et produit un diff. S’il itère après un échec de tests, on lui injecte la sortie des tests précédente — c’est ce retour qui rend la boucle efficace.

def editer(etat: Etat) -> dict:
    contexte = "\n\n".join(lire.invoke(f) for f in etat["fichiers_cibles"])
    retour = f"\n\nLes tests ont échoué :\n{etat['sortie_tests']}" if etat.get("sortie_tests") else ""
    rep = modele.invoke(
        f"Issue : {etat['issue']}\n\nCode actuel :\n{contexte}{retour}\n\n"
        f"Produis un patch unifié (git diff) qui corrige le problème, sans rien casser d'autre."
    )
    return {"diff": extraire_diff(rep.content), "tentatives": etat["tentatives"] + 1}

Nœud de test : appliquer dans le sandbox et juger

On applique le diff dans le sandbox (jamais sur le vrai dépôt), on lance les tests, on enregistre le verdict.

def tester(etat: Etat) -> dict:
    appliquer_diff_sandbox(RACINE, etat["diff"])      # patch temporaire, isolé
    sortie = lancer_tests.invoke({})
    return {"tests_ok": sortie.startswith("SUCCÈS"), "sortie_tests": sortie}

L’aiguillage : vert, on valide ; rouge, on réitère

from typing import Literal
from langgraph.graph import END

def apres_tests(etat: Etat) -> Literal["valider", "editer", "abandonner"]:
    if etat["tests_ok"]:
        return "valider"
    if etat["tentatives"] >= 4:           # garde-fou anti-boucle
        return "abandonner"
    return "editer"                       # ré-essaie avec la sortie des tests en contexte

Validation humaine avant d’écrire

Les tests passent — mais un humain valide le diff avant qu’il ne touche le vrai dépôt. Dernier garde-fou, le plus important.

from langgraph.types import interrupt, Command

def valider(etat: Etat) -> Command:
    decision = interrupt({
        "diff": etat["diff"],
        "tests": "verts",
        "demande": "Appliquer ce correctif au dépôt ?",
    })
    if decision == "approuver":
        appliquer_diff_reel(etat["diff"])             # le seul endroit qui écrit vraiment
        return Command(goto=END, update={"messages": [{"role": "assistant",
            "content": "Correctif appliqué, tests au vert."}]})
    return Command(goto=END, update={"messages": [{"role": "assistant",
        "content": "Correctif rejeté, aucune modification appliquée."}]})

Câbler le graphe

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver

def abandonner(etat: Etat) -> dict:
    return {"messages": [{"role": "assistant",
        "content": "Je n'ai pas réussi à faire passer les tests en 4 tentatives. "
                   "Voici ma meilleure proposition et les logs pour reprise humaine."}]}

g = StateGraph(Etat)
g.add_node("explorer", explorer)
g.add_node("editer", editer)
g.add_node("tester", tester)
g.add_node("valider", valider)
g.add_node("abandonner", abandonner)

g.add_edge(START, "explorer")
g.add_edge("explorer", "editer")
g.add_edge("editer", "tester")
g.add_conditional_edges("tester", apres_tests,
    {"valider": "valider", "editer": "editer", "abandonner": "abandonner"})
g.add_edge("abandonner", END)

agent_code = g.compile(checkpointer=InMemorySaver())   # HITL ⇒ checkpointer