LangGraph: Construye un agente con llamada a herramientas y ejecución acotada para verificación de código

La modificación autónoma de código requiere algo más que una cadena de prompts lineal estándar. Cuando un agente modifica un repositorio, debe leer archivos, aplicar ediciones, ejecutar suites de tests, inspeccionar salidas y autocorregirse cuando aparecen regresiones. Sin embargo, los patrones cíclicos de agentes —a menudo basados en el paradigma ReAct (Reasoning and Acting)— plantean riesgos operativos si no se delimitan. Un agente que se topa con dependencias rotas o fallos persistentes en los tests puede entrar en bucles de ejecución costosos e infinitos.

Basándose en la arquitectura ReAct para LangGraph documentada por Google AI, esta guía adapta la mecánica central de los grafos de estado a un agente de código especializado y acotado. La implementación ejecuta ediciones de código, verifica los cambios frente a suites de pruebas automatizadas y aplica límites estrictos de iteración.

---

Arquitectura central: Estado, nodos y aristas condicionales

LangGraph modela los flujos de trabajo de los agentes como grafos dirigidos. A diferencia de los grafos acíclicos dirigidos (DAG) estáticos, LangGraph admite ciclos de forma nativa, lo cual resulta esencial para el razonamiento iterativo de los agentes.

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

El grafo se apoya en tres elementos arquitectónicos:

1. Estado (State): Una estructura de datos tipada y centralizada que almacena el historial de mensajes, los metadatos de ejecución y contadores de pasos explícitos.
2. Nodos (Nodes): Funciones estándar de Python que reciben el estado actual, ejecutan computaciones aisladas (llamar al modelo o ejecutar una herramienta) y devuelven atributos de estado actualizados.
3. Aristas condicionales (Conditional Edges): Funciones de decisión que inspeccionan el estado (como la presencia de llamadas a herramientas o la superación de un límite de iteraciones) para decidir si enrutan hacia el ejecutor de herramientas o terminan en END.

---

Requisitos previos y dependencias

Asegúrate de que tu entorno de desarrollo incluya los paquetes necesarios para la orquestación del grafo y los controladores del modelo:

pip install langgraph langchain-core langchain-openai

Para conectar el agente a proveedores de inferencia a través de una pasarela compatible con OpenAI, configura tus variables de entorno. El uso de OneMux simplifica la gestión unificada de claves y el control de gasto entre múltiples modelos:

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

(Nota: Verifica siempre que el modelo de destino seleccionado admita parámetros estructurados de tool calling a través de endpoints compatibles con OpenAI).

---

Paso 1: Definición del estado acotado del agente

En los flujos conversacionales estándar, MessagesState se encarga de anexar el historial del chat. Para tareas de programación, definimos explícitamente AgentState mediante TypedDict de Python para rastrear secuencias de mensajes junto con un contador explícito (step_count) y un umbral máximo (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

El reductor add_messages garantiza que las listas de mensajes entrantes se añadan limpiamente al historial de la conversación en lugar de sobrescribir observaciones anteriores.

---

Paso 2: Implementación de herramientas de sistema de archivos y verificación

El agente necesita herramientas para inspeccionar archivos de código fuente, aplicar actualizaciones y ejecutar los tests del repositorio. Encapsulamos estas operaciones utilizando el decorador @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}

---

Paso 3: Configuración del modelo mediante la API unificada

Inicializamos el modelo utilizando ChatOpenAI, apuntando al endpoint proxy unificado del proveedor. Consulta la documentación de OneMux y la guía de inicio rápido para comprobar la compatibilidad de parámetros en el 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)

---

Paso 4: Construcción de nodos del grafo y protectores de bucle

Los nodos ejecutan la lógica e incrementan los límites. Si un agente entra en bucle repetidamente sin corregir un test fallido, la lógica de la arista detecta que step_count >= max_steps y redirige directamente a la finalización del grafo.

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"

---

Paso 5: Compilación y ensamblaje del grafo

Ensambla los componentes utilizando StateGraph y compila el flujo de trabajo:

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()

---

Paso 6: Ejecución de cambios de código con verificación mediante tests

Para ejecutar el flujo de trabajo, define una tarea de programación junto con límites de ejecución estrictos:

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.

---

Limitaciones y robustecimiento para producción

Al desplegar agentes que modifican código en entornos de producción, ten en cuenta las siguientes medidas de protección:

• Aislamiento de la ejecución (Sandboxing): Ejecutar comandos de shell arbitrarios como pytest en local entraña el riesgo de corromper el entorno o ejecutar scripts de forma insegura. Aísla la ejecución de tests dentro de contenedores Docker efímeros o sandboxes seguros.
• Límites de tokens en el estado: Las salidas extensas de herramientas —como los rastreos de pila (stack traces) de varias páginas— pueden agotar rápidamente las ventanas de contexto. Trunca los mensajes de salida de los tests antes de escribirlos en el historial de estado.
• Gestión de costes del modelo: Los bucles iterativos de ReAct multiplican el consumo de tokens. Monitorea el uso mediante el panel de precios de OneMux y evalúa modelos de razonamiento de menor coste en modelos de OneMux para optimizar el coste por tarea.

---

Fuentes

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Actualizado el 2 de septiembre de 2026).