Lorsque deux agents battent un seul : un copilote d’analyse de code LangGraph fonctionnant dans les deux sens

« Multi-agent » est le mot à la mode en ce moment. Les démonstrations montrent cinq personnalités d’IA discutant dans une pièce simulée pour extraire des informations d’un site et rédiger un article de blog, avec des noms tels que Chercheur, Écrivain et Éditeur, ainsi qu’un terminal rempli d’agents qui débattent entre eux. Cela a l’air impressionnant, mais ce n’est pas ainsi que l’on développe des logiciels fiables.

Ajouter des agents ne rend pas un système plus intelligent en soi. Cela multiplie les possibilités d’échec. Prenons l’arithmétique de base : un agent qui accomplit correctement sa tâche 90 % du temps a une fiabilité de 0,90. En enchaînant trois d’eux, les probabilités se multiplient, ce qui fait chuter ce taux à environ 72 %. La latence double, la consommation d’API triple, et le résultat final n’est pas meilleur pour autant.

La question qui mérite une réponse est de savoir quand plus d’un agent parvient réellement à régler le problème. Une bonne façon de le découvrir consiste à créer deux assistants d’examen de code basés sur l’IA. La première version est un examinateur unique et conventionnel. La seconde divise le travail entre un analyseur et un examinateur des risques, en utilisant des fonctionnalités de LangGraph telles que des arêtes conditionnelles, interrupt() ainsi que la reprise du travail basée sur des Command afin d’intervenir avec un humain en cas de besoin. Cette comparaison montre pourquoi l’agent unique rate une grave erreur de facturation et pourquoi le duo la détecte.

Pourquoi l’examen de code est difficile avec une seule instruction

Un bon examen nécessite deux types de réflexion complètement différents.

Le premier est la compréhension. On lit la différence entre les versions, on identifie les fonctions qui ont changé, on détermine quels fichiers sont affectés et on infère ce que l’auteur avait en tête. L’objectif est de se forger un modèle mental de la nouvelle fonctionnalité.

Le deuxième est d’ordre adversarial. On cherche des cas limites, des tests manquants et des failles logiques, et on tente de briser le code.

Si l’on demande à un LLM de faire les deux en une seule requête, il a du mal. Résumer et critiquer en même temps tend à privilégier la cohérence aux doutes. Une fois que le modèle s’est fixé sur une interprétation de ce que fait le code, il consacre moins d’efforts à remettre en question cette interprétation et se contente de remarques mineures sur le style, comme l’absence de docstrings ou de conventions de nommage.

Un exemple concret illustre bien cela. Imaginez une demande de pull request fictive concernant un système de facturation dans laquelle l’auteur ajoute une fonction de remise à un calculateur de factures. Voici le diff reçu par les deux versions :

--- a/src/billing/invoice.py
+++ b/src/billing/invoice.py
@@ -42,7 +42,9 @@
 class InvoiceService:
-    def calculate_total(self, items):
-        return sum(i.price * i.qty for i in items)
+    def calculate_total(self, items, discount_pct=0):
+        subtotal = sum(i.price * i.qty for i in items)
+        return subtotal * (1 - discount_pct)

--- a/src/billing/api.py
+++ b/src/billing/api.py
@@ -18,6 +18,8 @@
 @router.post("/invoice")
 def create_invoice(req: InvoiceRequest):
+    discount = req.discount_pct   # NEW: from user input
     svc = InvoiceService()
-    total = svc.calculate_total(req.items)
+    total = svc.calculate_total(req.items, discount)
     return {"total": total}

--- a/config/feature_flags.yaml
+++ b/config/feature_flags.yaml
@@ -5,3 +5,4 @@
 flags:
   new_dashboard: true
+  discount_billing: true   # rollout: 100% immediately

En y regardant de plus près, plusieurs problèmes graves se détachent. La valeur discount_pct provient directement des entrées utilisateur au niveau de l’API et est utilisée dans les calculs sans aucune validation. Un utilisateur pourrait envoyer un pourcentage de remise négatif pour augmenter le prix, ou une valeur supérieure à 1 pour obtenir un total négatif. Les tests ne couvrent pas cette nouvelle situation. De plus, le paramètre de fonctionnalité est réglé pour atteindre 100 % des utilisateurs immédiatement. Il s’agit d’un processus de tarification particulièrement vulnérable aux abus. Comment un évaluateur unique gère-t-il cela ?

Conception un : un évaluateur unique

Le kit d’outils de l’agent vient en premier : un outil pour obtenir les différences, un autre pour ouvrir des fichiers, un troisième pour rechercher des symboles, et enfin un outil pour indiquer quels tests existent.

from langchain_core.tools import tool
import textwrap

MOCK_DIFF = """...""" # The diff shown above
MOCK_FILES = {
    "src/billing/invoice.py": "class InvoiceService:\n    def calculate_total(self, items, discount_pct=0):\n        subtotal = sum(i.price * i.qty for i in items)\n        return subtotal * (1 - discount_pct)\n",
    "src/billing/api.py": '@router.post("/invoice")\ndef create_invoice(req: InvoiceRequest):\n    discount = req.discount_pct\n    svc = InvoiceService()\n    total = svc.calculate_total(req.items, discount)\n    return {"total": total}\n',
    "tests/test_billing.py": "def test_calculate_total():\n    # only tests no-discount path\n    assert svc.calculate_total(items) == 300\n",
}

@tool
def get_diff(pr_id: str) -> str:
    """Fetch the PR diff."""
    return MOCK_DIFF

@tool
def read_file(path: str) -> str:
    """Read a file from the repo."""
    return MOCK_FILES.get(path, f"FILE NOT FOUND: {path}")

@tool
def search_symbol(name: str) -> str:
    """Search for a symbol across the codebase."""
    if "discount" in name.lower():
        return "Found: InvoiceService.calculate_total(discount_pct) — src/billing/invoice.py:43"
    return f"No results for '{name}'"

@tool
def list_tests(path: str) -> str:
    """List test files covering a source path."""
    if "billing" in path:
        return "tests/test_billing.py — covers calculate_total (no-discount path only)"
    return "No tests found"

ALL_TOOLS = [get_diff, read_file, search_symbol, list_tests]

Puis l’état. Un TypedDict assure la compatibilité avec LangGraph et contient l’ID de la PR, l’historique des messages, la différence brute ainsi qu’une liste des résultats de vérification. Un modèle Pydantic décrit chaque résultat et oblige le modèle à classer ce qu’il trouve à l’aide de littéraux fixes, ce qui facilite grandement le traitement ultérieur.

from typing import Literal
from typing_extensions import TypedDict
from pydantic import BaseModel
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph

class ReviewFinding(BaseModel):
    severity: Literal["critical", "high", "medium", "low", "info"]
    category: Literal["bug", "regression", "missing_test", "rollout_risk",
                       "security", "edge_case", "style"]
    file: str
    description: str
    confidence: Literal["high", "medium", "low"]

class SingleAgentState(TypedDict):
    pr_id: str
    messages: list
    diff: str
    findings: list[ReviewFinding]

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)

Viennent ensuite les nœuds. Le graphe est linéaire : un nœud de récupération qui charge le contexte, suivi d’un nœud d’analyse qui exécute la demande.

def sa_fetch(state: SingleAgentState) -> dict:
    diff = get_diff.invoke({"pr_id": state["pr_id"]})
    tests = list_tests.invoke({"path": "src/billing"})
    return {"diff": diff, "messages": [
        AIMessage(content=f"Diff loaded. Test coverage: {tests}")
    ]}

def sa_review(state: SingleAgentState) -> dict:
    """Single agent does BOTH jobs: summarize + critique in one pass."""
    prompt = f"""\
You are a senior code reviewer. Read this PR diff, summarize the change,
and produce a list of risk findings. Be specific.

DIFF:
{state['diff']}

Respond with JSON: {{"summary": "...", "findings": [
  {{"severity": "...", "category": "...", "file": "...",
    "description": "...", "confidence": "..."}}
]}}
"""
    resp = llm.invoke([SystemMessage(content="You are a code review bot."),
                       HumanMessage(content=prompt)])

    # In a real app we parse the JSON response here.
    # We simulate the typical single-agent output for this diff.
    findings = [
        ReviewFinding(
            severity="medium",
            category="missing_test",
            file="tests/test_billing.py",
            description="Add unit tests for the new discount_pct parameter.",
            confidence="high"
        ),
        ReviewFinding(
            severity="low",
            category="style",
            file="src/billing/invoice.py",
            description="Consider adding type hints to the items list.",
            confidence="high"
        )
    ]
    return {"findings": findings, "messages": [resp]}

def build_single_agent():
    g = StateGraph(SingleAgentState)
    g.add_node("fetch", sa_fetch)
    g.add_node("review", sa_review)
    g.add_edge(START, "fetch")
    g.add_edge("fetch", "review")
    g.add_edge("review", END)
    return g.compile()

(Dans cette démonstration, les résultats présentés sont des exemples prédéfinis représentant le type de sortie que produit généralement un examinateur ; considérez donc cette comparaison comme une illustration et non comme une référence de performance.)

C’est à peu près ainsi que sont conçus les premiers examinateurs basés sur l’IA : on leur soumet la différence entre les versions et on leur demande de détecter les bugs. En testant sur le PR simulé, l’agent remarque le test manquant et propose des indications de type. Il rate cependant les entrées non validées ainsi que les risques liés au déploiement.

La raison réside dans l’instruction donnée. Lorsqu’on lui demande de résumer le changement puis de lister les risques, le modèle consacre son attention à expliquer qu’une fonction de remise a été ajoutée. Lorsqu’il commence à énumérer les résultats, il est déjà dans un mode descriptif, se concentrant sur la syntaxe superficielle sans essayer de détecter des problèmes. La solution consiste à séparer la tâche de synthèse de celle d’analyse critique, ce qui implique l’utilisation de deux agents.

Le transfert de responsabilités : utiliser un schéma plutôt que du texte narratif

La plus grande erreur dans la conception de systèmes multi-agents est de permettre aux agents de communiquer entre eux par du texte libre. Si l’agent A écrit trois paragraphes et les transmet à l’agent B, ce dernier doit analyser le texte, deviner ce qui est important et travailler sur une base floue. C’est ainsi que le contexte se dégrade.

Les agents sont des fonctions, et les fonctions doivent transmettre des objets de type défini. Pour que un système multi-agents soit fiable, il faut définir le schéma exact de ce qui circule entre eux.

Dans l’assistant de revue, le premier agent est l’Analyseur, dont la tâche est de cartographier la demande de modification. Il doit générer un ChangeModel, un objet qui l’oblige à extraire des métadonnées spécifiques concernant la modification.

class ChangedInterface(BaseModel):
    file: str
    symbol: str
    change_type: Literal["added", "modified", "removed"]
    description: str

class ChangeModel(BaseModel):
    """Analyzer → Reviewer handoff. Structured, not prose."""
    purpose: str
    impacted_files: list[str]
    changed_interfaces: list[ChangedInterface]
    config_changes: list[str]
    migration_risk: bool
    assumptions: list[str]
    tests_touched: list[str]
    tests_likely_needed: list[str]

Chaque champ sert l’agent suivant dans la chaîne de traitement. La liste des interfaces modifiées permet à l’autre agent de savoir rapidement ce qui est important. La liste des modifications de configuration permet à l’Analyseur de séparer la logique applicative de la logique de déploiement.

Le tableau assumptions mérite une attention particulière. L’Analyseur doit préciser ce que l’auteur du code suppose, par exemple que discount_pct est un nombre entre 0 et 1. Il ne critique rien, il se contente d’énoncer des faits. Cet objet structuré devient alors l’entrée exacte pour l’Évaluateur.

Conception deux : Analyseur et Évaluateur des risques

Le nouveau schéma d’état contient le ChangeModel ainsi qu’un indicateur booléen conflict, qui déterminera par la suite s’il faut faire appel à une personne.

class TwoAgentState(TypedDict):
    pr_id: str
    messages: list
    diff: str
    change_model: ChangeModel | None
    findings: list[ReviewFinding]
    conflict: bool

Le nœud fetch reste inchangé, en récupérant les différences et le taux de couverture des tests. Viens ensuite le nœud Analyzer, avec une instruction strictement définie. Le modèle est explicitement indiqué de ne pas chercher de bugs, de sorte qu’il se concentre exclusivement sur la compréhension.

ANALYZER_PROMPT = """\
You are the Analyzer agent. Your ONLY job: understand the change.
Do NOT critique. Do NOT hunt for bugs. Just map what changed.
Produce a structured change model."""

def ta_analyzer(state: TwoAgentState) -> dict:
    """Analyzer: compress & clarify. Optimizes for coherence."""

    # We use LLM structured output to fill the ChangeModel.
    # For this demonstration, we simulate the accurate analysis.
    cm = ChangeModel(
        purpose="Add discount percentage support to invoice billing",
        impacted_files=["src/billing/invoice.py", "src/billing/api.py",
                        "config/feature_flags.yaml"],
        changed_interfaces=[
            ChangedInterface(file="src/billing/invoice.py",
                symbol="InvoiceService.calculate_total",
                change_type="modified",
                description="Added discount_pct param (default 0)"),
            ChangedInterface(file="src/billing/api.py",
                symbol="create_invoice",
                change_type="modified",
                description="Reads discount_pct from request, passes to service"),
        ],
        config_changes=["discount_billing flag added, 100% rollout"],
        migration_risk=False,
        assumptions=[
            "discount_pct is expected to be 0..1 (fraction, not percentage)",
            "No existing callers pass discount_pct yet",
            "Feature flag controls visibility, not the calculation",
        ],
        tests_touched=[],
        tests_likely_needed=[
            "test discount path in calculate_total",
            "test boundary: discount_pct = 0, 1, >1, <0",
            "test API validation of discount_pct input",
        ],
    )
    return {
        "change_model": cm,
        "messages": [AIMessage(content=f"Analyzer: change model built. "
                               f"{len(cm.changed_interfaces)} interfaces changed, "
                               f"{len(cm.assumptions)} assumptions made.")],
    }

L’Analyseur lit la différence et enregistre ce qui s’y trouve réellement : le nouveau paramètre, le nouveau flag de fonctionnalité, ainsi que l’hypothèse implicite selon laquelle la remise se situe entre 0 et 1.

Ensuite, l’objet est transmis au Réviseur de risques, dont les instructions sont délibérément hostiles. On lui demande de ne pas faire confiance à l’Analyseur et de chercher des défauts.

REVIEWER_PROMPT = """\
You are the Risk Reviewer. The Analyzer gave you a change model.
DISTRUST it. Your job: find what's missing, broken, or dangerous.
Challenge every assumption. Check for missing tests, regressions,
rollout risks, and security issues."""

def ta_reviewer(state: TwoAgentState) -> dict:
    """Reviewer: expand & challenge. Optimizes for skepticism."""
    cm = state["change_model"]
    findings: list[ReviewFinding] = []

    # The Reviewer iterates through the Analyzer's assumptions
    for a in cm.assumptions:
        if "0..1" in a:
            findings.append(ReviewFinding(
                severity="critical",
                category="security",
                file="src/billing/api.py",
                description="ASSUMPTION CHALLENGED: discount_pct comes from user input "
                    "(req.discount_pct) with NO validation. Values <0 or >1 break billing. "
                    "Negative discount = price increase beyond subtotal. "
                    "Value >1 = negative total.",
                confidence="high"))

    # The Reviewer checks the test mapping
    if not cm.tests_touched and cm.tests_likely_needed:
        findings.append(ReviewFinding(
            severity="high",
            category="missing_test",
            file="tests/test_billing.py",
            description=f"NO tests touched but {len(cm.tests_likely_needed)} needed: "
                + "; ".join(cm.tests_likely_needed),
            confidence="high"))

    # The Reviewer checks the config changes
    for cc in cm.config_changes:
        if "100%" in cc:
            findings.append(ReviewFinding(
                severity="high",
                category="rollout_risk",
                file="config/feature_flags.yaml",
                description="Feature flag at 100% from day one — no gradual rollout. "
                    "Combined with unvalidated discount input, this is a billing incident risk.",
                confidence="high"))

    # The Reviewer looks for edge cases in the logic flow
    findings.append(ReviewFinding(
        severity="medium",
        category="edge_case",
        file="src/billing/api.py",
        description="Feature flag controls visibility but calculate_total always applies "
            "discount. If flag is off but API still receives discount_pct, discount "
            "is silently applied.",
        confidence="medium"))

    # We determine if there is a conflict worth escalating
    conflict = any(f.severity in ("critical", "high") for f in findings)

    return {
        "findings": findings,
        "conflict": conflict,
        "messages": [AIMessage(content=f"Reviewer: {len(findings)} findings, "
                               f"conflict={conflict}")],
    }

Comme l’Évaluateur ne consomme pas de tokens pour déterminer quels fichiers ont changé ou à quoi sert la PR, tout son contexte vient perturber la logique. Il part du principe que la remise doit être comprise entre 0 et 1, vérifie la différence, ne trouve aucun code de validation, et signale alors une erreur de validation des entrées de gravité élevée.

C’est l’un des rares cas où plusieurs agents interviennent ensemble : l’un construit l’objectif et l’autre le teste.

Arbitrage avec un humain impliqué

L’Évaluateur a identifié des problèmes graves que l’auteur de la PR a manqués. Cependant, les publier directement sur GitHub serait risqué. Si l’Évaluateur se trompe, cela irritera l’équipe. Une solution pratique en environnement de production serait d’ajouter un nœud de fusion qui évalue ces constats et décide s’il faut faire appel à un humain pour arbitrer.

def ta_merge(state: TwoAgentState) -> dict:
    """Merge findings. If conflict, surface for human review."""
    if state["conflict"]:
        return {
            "messages": [AIMessage(content=(
                "⚠ CONFLICT: Reviewer found critical/high issues. "
                "Routing to human review."
            ))],
        }
    return {
        "messages": [AIMessage(content="Findings merged. No escalation needed.")],
    }

def route_after_merge(state: TwoAgentState) -> str:
    return "escalate" if state["conflict"] else "emit"

Lorsque le drapeau conflict est activé, le flux se dirige vers un nœud d’escalade. C’est là que les arêtes conditionnelles, interrupt() ainsi que la reprise basée sur des Command entrent en jeu.

from langgraph.types import Command, interrupt

def ta_escalate(state: TwoAgentState) -> Command:
    """Human-in-the-loop for conflicting findings."""

    # The interrupt function pauses execution and surfaces data to the caller
    decision = interrupt({
        "kind": "review_conflict",
        "pr_id": state["pr_id"],
        "change_model": state["change_model"].model_dump(),
        "findings": [f.model_dump() for f in state["findings"]],
        "prompt": "Review findings. Respond: "
                  '{"action":"accept_all"} or {"action":"override","drop_indices":[...]}',
    })

    # Execution resumes here when the human provides input
    action = decision.get("action", "accept_all")

    if action == "override":
        drop = set(decision.get("drop_indices", []))
        kept = [f for i, f in enumerate(state["findings"]) if i not in drop]

        # We use Command to update state and dynamically route to the next node
        return Command(update={"findings": kept}, goto="emit")

    return Command(goto="emit")

L’appel à interrupt() provoque une pause de l’exécution. Le graphe serialise son état actuel et le stocke via le point de contrôle configuré, qui est MemorySaver dans une démo locale. En production, on utiliserait un mécanisme de sauvegarde fiable soutenu par Postgres afin que l’état en pause survive aux redémarrages du serveur.

Votre backend peut ensuite envoyer un message Slack à un ingénieur senior contenant les résultats structurés. Celui-ci lit l’avertissement critique concernant la remise non validée et, s’il est d’accord, appuie sur le bouton « Accepter tout ». Lorsque le webhook atteint votre serveur, celui-ci reprend l’exécution en transmettant au graphe une Commande contenant les données de reprise.

# How you resume the graph from your backend API
ta.invoke(Command(resume={"action": "accept_all"}), config)

Le graphe s’active, traite la décision humaine, ajuste les résultats si nécessaire et se dirige vers le nœud d’émission final.

def ta_emit(state: TwoAgentState) -> dict:
    """Final output: formatted review."""
    lines = [f"=== Code Review: PR {state['pr_id']} ==="]
    if state["change_model"]:
        lines.append(f"Purpose: {state['change_model'].purpose}")
    lines.append(f"Findings ({len(state['findings'])}):")

    for i, f in enumerate(state["findings"]):
        lines.append(f"  [{f.severity}] ({f.category}) {f.file}")
        lines.append(f"    {f.description}")

    return {"messages": [AIMessage(content="\n".join(lines))]}

Connexion du graphe

Les nœuds sont assemblés en un StateGraph. Aucun bord statique n’est nécessaire en provenance du nœud d’escalade, car ta_escalate renvoie une Command(goto="emit") et LangGraph gère les routages de manière dynamique.

from langgraph.checkpoint.memory import MemorySaver

def build_two_agent():
    g = StateGraph(TwoAgentState)

    g.add_node("fetch", ta_fetch)
    g.add_node("analyzer", ta_analyzer)
    g.add_node("reviewer", ta_reviewer)
    g.add_node("merge", ta_merge)
    g.add_node("escalate", ta_escalate)
    g.add_node("emit", ta_emit)

    g.add_edge(START, "fetch")
    g.add_edge("fetch", "analyzer")
    g.add_edge("analyzer", "reviewer")
    g.add_edge("reviewer", "merge")

    g.add_conditional_edges("merge", route_after_merge,
        {"escalate": "escalate", "emit": "emit"})

    g.add_edge("emit", END)

    # A checkpointer is required to use interrupt()
    # Use MemorySaver for local testing, Postgres for production
    return g.compile(checkpointer=MemorySaver())

Comparaison des deux

En plaçant les deux reviewers côte à côte dans ce PR, l’écart devient évident.

L’agent unique identifie deux problèmes. Il recommande d’ajouter des tests unitaires et suggère des indications de type. Il se comporte comme un développeur junior poli qui se concentre sur la syntaxe.

Le duo en trouve quatre. Il signale que les données non validées constituent une faille critique dans la logique métier. Il note que des remises négatives augmenteraient les prix. Il considère un déploiement immédiat à 100 % comme un risque pour la facturation. De plus, il identifie un cas limite dans lequel la remise est appliquée en silence même lorsque le paramètre de fonctionnalité est désactivé.

La conception multi-agents ne garantit pas toujours de meilleures évaluations. Elle a été utile ici car la décomposition était exceptionnellement claire. Le deuxième agent a apporté des informations réelles et non du bruit, permettant de détecter les défauts de logique métier qui perturbent le fonctionnement. Cela a fonctionné car la charge cognitive a été répartie : l’Analyseur s’est chargé de l’analyse approfondie et de la mise en correspondance des interfaces, tandis que l’Évaluateur recevait un ChangeModel propre et pouvait se concentrer sur son travail d’évaluation.

Quand ne pas utiliser plusieurs agents

Cette combinaison est puissante, et l’envie sera grande de réécrire chaque agent en un flux de travail multi-agents. Résistez-y.

Des agents supplémentaires augmentent la latence. L’approche à deux agents prend environ deux fois plus de temps, car les appels à l’LLM sont séquentiels, coûte environ deux fois plus en tokens, et implique un schéma d’état et une logique de transfert plus complexes. Utilisez ce modèle uniquement si la tâche correspond à l’une des trois situations spécifiques.

Travail antagoniste. Si un système doit à la fois produire et critiquer du contenu, séparez ces rôles. La conception à plusieurs agents avec un générateur et un discriminateur est celle qui s’est le mieux avérée. Un agent écrit du code et un autre écrit des tests pour le faire échouer. Un agent rédige un e-mail et un autre le vérifie pour détecter d’éventuels problèmes de conformité.

Isolement strict des outils. Un agent qui explore le web public et un agent qui exécute des requêtes de base de données ne devraient pas partager le même mécanisme de traitement. Permettez à un chercheur de collecter des données publiques et de transmettre un résumé nettoyé à un exécutant. Cela crée une barrière de sécurité solide. Si une page web contient une injection de commande, seul le chercheur est compromis tandis que l’exécutant reste isolé.

Échelle contextuelle asymétrique. Parfois, une énorme quantité de données alimente une décision mineure. Une flotte d’agents bon marché et rapides peut lire 100 fichiers de journal en parallèle, en extraire les lignes d’erreur pertinentes et transmettre un bref résumé à un seul superviseur coûteux qui prend la décision finale.

Si une tâche ne correspond à aucun de ces cas, privilégiez un seul agent. Un agent unique doté d’une instruction bien formulée, de contraintes strictes sur les outils et d’un schéma d’état clair surpassera toujours une configuration complexe avec plusieurs agents.