LangGraph : concevoir un agent avec appel d'outils et exécution bornée pour la vérification de code

La modification autonome de code exige bien plus qu'une simple chaîne linéaire d'invites. Lorsqu'un agent modifie un dépôt, il doit lire des fichiers, appliquer des modifications, exécuter des suites de tests, analyser les sorties et s'autocorriger en cas de régression. Cependant, les architectures d'agents cycliques — souvent fondées sur le paradigme ReAct (Reasoning and Acting) — présentent des risques opérationnels s'ils ne sont pas encadrés. Un agent confronté à des dépendances cassées ou à des tests qui échouent de façon persistante peut entrer dans des boucles d'exécution infinies et coûteuses.

En s'appuyant sur l'architecture ReAct documentée par Google AI pour LangGraph (Google AI's documented ReAct architecture for LangGraph), ce guide adapte les mécaniques fondamentales des graphes d'état pour concevoir un agent de codage spécialisé et borné. L'implémentation effectue des modifications de code, valide les changements à l'aide de suites de tests automatisées et impose des limites strictes sur le nombre d'itérations.

---

Architecture fondamentale : État, nœuds et arêtes conditionnelles

LangGraph modélise les flux de travail des agents sous forme de graphes orientés. Contrairement aux graphes orientés acycliques (DAG) statiques, LangGraph prend nativement en charge les cycles, indispensables au raisonnement itératif d'un agent.

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

Le graphe repose sur trois éléments architecturaux :

1. État (State) : Une structure de données typée et centralisée qui stocke l'historique des messages, les métadonnées d'exécution et les compteurs d'étapes explicites.
2. Nœuds (Nodes) : Des fonctions Python ordinaires qui reçoivent l'état courant, exécutent des calculs isolés (appel du modèle ou exécution d'un outil) et renvoient les attributs d'état mis à jour.
3. Arêtes conditionnelles (Conditional Edges) : Des fonctions de décision qui inspectent l'état (par exemple, la présence d'appels d'outils ou le dépassement d'un plafond d'itérations) afin de déterminer s'il faut router vers l'exécuteur d'outils ou s'arrêter à END.

---

Prérequis et dépendances

Vérifiez que votre environnement de développement comprend les packages nécessaires à l'orchestration du graphe et aux pilotes de modèles :

pip install langgraph langchain-core langchain-openai

Pour connecter l'agent aux fournisseurs d'inférence via une passerelle compatible OpenAI, configurez vos variables d'environnement. L'utilisation de OneMux simplifie la gestion centralisée des clés et le contrôle des coûts sur plusieurs modèles :

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(Remarque : vérifiez toujours que le modèle cible sélectionné prend en charge les paramètres structurés d'appel d'outils via les endpoints compatibles OpenAI).

---

Étape 1 : Définir un état d'agent borné

Dans les flux conversationnels standards, MessagesState se charge d'ajouter l'historique de discussion. Pour les tâches de programmation, nous définissons explicitement AgentState à l'aide du module TypedDict de Python pour suivre les séquences de messages ainsi qu'un compteur explicite (step_count) et un seuil maximal (max_steps).

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

Le réducteur add_messages garantit que les nouvelles listes de messages sont correctement ajoutées à l'historique de conversation au lieu d'écraser les observations précédentes.

---

Étape 2 : Implémenter les outils de système de fichiers et de vérification

L'agent a besoin d'outils pour examiner les fichiers sources, appliquer des mises à jour et exécuter les tests du dépôt. Nous encapsulons ces opérations à l'aide du décorateur @tool de LangChain.

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

Étape 3 : Configuration du modèle via une API unifiée

Nous initialisons le modèle à l'aide de ChatOpenAI, en pointant vers l'endpoint proxy du fournisseur unifié. Consultez la documentation OneMux et le guide de démarrage rapide pour vérifier la compatibilité des paramètres pour l'appel de fonctions (function calling).

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

Étape 4 : Construire les nœuds du graphe et les garde-fous de boucle

Les nœuds exécutent la logique métier et incrémentent les limites. Si un agent boucle sans parvenir à corriger un test défaillant, la logique de l'arête détecte que step_count >= max_steps et oriente directement vers la fin du graphe.

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

Étape 5 : Compilation et assemblage du graphe

Assemblez les composants à l'aide de StateGraph et compilez le flux de travail :

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

Étape 6 : Exécuter les modifications de code avec vérification par tests

Pour lancer le flux de travail, définissez une tâche de programmation accompagnée de limites d'exécution strictes :

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

Limites et durcissement pour la production

Avant de déployer des agents modificateurs de code en production, mettez en place les mesures de protection suivantes :

• Isolation de l'exécution (Sandboxing) : Exécuter localement des commandes shell arbitraires comme pytest expose à des corruptions d'environnement ou à l'exécution de scripts non sécurisés. Isolez l'exécution des tests dans des conteneurs Docker éphémères ou des bacs à sable sécurisés.
• Limites de tokens dans l'état : Les sorties volumineuses des outils — telles que les traces de pile (stack traces) sur plusieurs pages — peuvent saturer rapidement la fenêtre de contexte. Tronquez les messages de sortie des tests avant de les insérer dans l'historique d'état.
• Gestion des coûts de modèles : Les boucles itératives ReAct multiplient la consommation de tokens. Suivez votre utilisation via le tableau de bord des tarifs OneMux, et évaluez des modèles de raisonnement plus économiques sur OneMux Models pour optimiser le coût par tâche.

---

Sources

• Google AI for Developers : ReAct agent from scratch with Gemini and LangGraph (Mis à jour le 2 sept. 2026).