Cuando dos agentes superan a uno: un Copilot de revisión de código LangGraph diseñado para ambos sentidos

“Multi-agente” es la palabra de moda en estos momentos. Las demostraciones muestran a cinco personajes de IA charlando en una sala simulada para extraer información de un sitio web y escribir una entrada de blog, con nombres como Investigador, Escritor y Editor, además de una terminal llena de agentes que discuten entre sí. Parece impresionante, pero no es así como se construyen los software fiables.

Añadir agentes no hace que un sistema sea más inteligente por sí solo. Solo multiplica las posibilidades de que algo falle. Tomemos la aritmética básica: un agente que realiza su tarea correctamente el 90 % de las veces tiene una fiabilidad del 0,90. Si se conectan tres de ellos, las probabilidades se multiplican, por lo que la fiabilidad básica baja a aproximadamente el 72 %. La latencia se duplica, el gasto en API se triplica, y el resultado final no es mejor.

La pregunta que merece respuesta es cuándo es cuando más de un agente realmente resuelve el problema. Una buena forma de averiguarlo es crear dos versiones de un asistente de revisión de código basado en IA. La primera versión es un revisor convencional de un solo agente. La segunda divide el trabajo entre un analizador y un revisor de riesgos, utilizando características de LangGraph como aristas condicionales, interrupt() y una reanudación basada en Command para involucrar a un humano cuando sea necesario. La comparación muestra por qué el agente único pasa por alto un grave error en la facturación y por qué la pareja lo detecta.

Por qué la revisión de código es difícil con una sola instrucción

Una buena revisión requiere dos tipos muy diferentes de pensamiento.

El primero es la comprensión. Se lee la diferencia entre versiones, se identifican las funciones que han cambiado, se determinan los archivos afectados e se infiere qué pretendía el autor. El objetivo es crear un modelo mental de la nueva función.

El segundo enfoque es adversarial. Se buscan casos límite, pruebas ausentes y fallos lógicos, y se intenta romper el código.

Si se pide a un LLM que haga ambas cosas en una sola solicitud, tiene dificultades. Resumir y criticar en la misma etapa tiende a favorecer la coherencia sobre la duda. Una vez que el modelo se decide por una versión de lo que hace el código, invierte menos esfuerzo en cuestionar esa versión y recurre a observaciones menores sobre el estilo, como la falta de strings de documentación o convenciones de nomenclatura.

Un caso concreto lo ilustra claramente. Imaginemos una solicitud de pull request ficticia contra un sistema de facturación en la que el autor agrega una función de descuento a un calculador de facturas. Este es el diferencial que reciben ambos diseños:

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

Al observar con atención, se destacan varios problemas graves. El valor de discount_pct proviene directamente de la entrada del usuario en la capa de API y se utiliza en los cálculos sin ninguna validación. Un usuario podría enviar un descuento negativo para aumentar el precio, o un valor superior a 1 para obtener un total negativo. Las pruebas no cubren este nuevo escenario. Además, la bandera de funcionalidad está configurada para llegar al 100% de los usuarios de inmediato. Se trata de un proceso de fijación de precios propenso al abuso. ¿Cómo lo maneja un revisor típico que trabaja solo?

Diseño uno: un revisor que trabaja solo

Primero está la caja de herramientas del agente: una herramienta para obtener las diferencias, otra para abrir archivos, una para buscar símbolos y otra para indicar qué pruebas existen.

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]

Luego está el estado. Un TypedDict mantiene la compatibilidad con LangGraph y alberga el ID de la PR, el historial de mensajes, la diferencia bruta y una lista de hallazgos. Un modelo Pydantic describe cada hallazgo y obliga al modelo a clasificar lo que encuentra mediante valores literales fijos, lo que facilita enormemente el procesamiento posterior.

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)

A continuación vienen los nodos. El grafo es una línea recta: un nodo de obtención que carga el contexto, y luego un nodo de revisión que ejecuta la instrucción.

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

(En esta demostración, los resultados son sustitutos predefinidos del tipo de salida que suele generar un revisor, por lo que considere esta comparación como una ilustración y no como un punto de referencia.)

Así es más o menos como se crean los revisores de IA de primera pasada: se introduce la diferencia y se solicitan errores. Al probarlo con el PR simulado, el agente detecta la prueba faltante y sugiere indicaciones de tipo. No detecta la entrada no validada ni el riesgo de despliegue.

La razón está en la instrucción dada. Al pedirle que resuma el cambio y luego enumere los riesgos, el modelo dedica su atención a explicar que se añadió una función de descuento. Para cuando comienza a elaborar el array de hallazgos, ya está en un tono descriptivo, centrado en la sintaxis superficial y sin intentar romper nada. La solución es separar la tarea de síntesis de la tarea adversarial, lo que implica utilizar dos agentes.

La transferencia: utilice un esquema y no prosa

El error más común en el diseño de sistemas multiagente es permitir que los agentes se comuniquen entre sí mediante texto libre. Si el agente A escribe tres párrafos y se los entrega al agente B, este debe analizar el texto, adivinar qué es importante y trabajar sobre una base poco precisa. Así es como el contexto se degrada.

Los agentes son funciones, y las funciones deben pasar objetos tipados. Para que un sistema multiagente sea fiable, se debe definir el esquema exacto de lo que se transmite entre ellos.

En el asistente de revisión, el primer agente es el Analizador, y su tarea es mapear la solicitud de cambios. Debe generar un ChangeModel, un objeto que le obliga a extraer metadatos específicos sobre el cambio.

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]

Cada campo sirve al agente posterior en el flujo de trabajo. La lista de interfaces modificadas permite al segundo agente consultar rápidamente qué elementos son importantes. La lista de cambios de configuración ayuda al Analizador a separar la lógica de la aplicación de la lógica de despliegue.

El array assumptions merece especial atención. El Analizador debe especificar qué supone el autor del código, como que discount_pct es un valor fraccionario entre 0 y 1. No está criticando nada, sino que simplemente expone hechos. Ese objeto estructurado se convierte en la entrada exacta para el Revisor.

Diseño dos: Analizador y Revisor de Riesgos

El nuevo esquema de estado contiene el ChangeModel además de una bandera booleana conflict, que posteriormente determina si es necesario escalonar el problema a una persona.

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

El nodo de fetch permanece igual, obteniendo la diferencia y el porcentaje de cobertura de pruebas. Luego viene el nodo Analyzer, con una instrucción estrictamente limitada. Se le indica explícitamente al modelo que no busque errores, por lo que se concentra únicamente en la comprensión.

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.")],
    }

El Analizador lee la diferencia y registra lo que realmente está allí: el nuevo parámetro, la nueva bandera de funcionalidad y la suposición implícita de que el descuento está entre 0 y 1.

A continuación, el objeto pasa al Revisor de Riesgos, cuya instrucción es deliberadamente adversarial; se le pide que desconfíe del Analizador y que busque defectos.

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}")],
    }

Dado que el Revisor no utiliza tokens para determinar qué archivos han cambiado o cuál es el propósito de la PR, todo su contexto contribuye a romper la lógica del sistema. Este asume que el descuento debe estar entre 0 y 1, verifica la diferencia entre versiones, no encuentra código de validación y marca un error de validación de entrada de alta gravedad.

Este es uno de los pocos casos en los que varios agentes colaboran para resolver un problema: uno crea el objetivo y el otro lo evalúa.

Arbitraje con la participación de un humano

El Revisor ha detectado problemas graves que el autor de la PR pasó por alto. Sin embargo, publicarlos directamente en GitHub sería arriesgado. Si el Revisor está cometiendo errores, eso molestará al equipo. Una adición sensata en producción sería un nodo de fusión que evalúe esos problemas y decida si es necesario que un humano intervenga en el arbitraje.

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"

Cuando la bandera conflict está activa, el flujo se dirige a un nodo de escalada. Allí es donde entran en juego los bordes condicionales, interrupt() y la reanudación basada en Command.

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

Al llegar a la llamada interrupt() se pausa la ejecución. El grafo serializa su estado actual y lo almacena a través del puntero de verificación configurado, que en una demo local es MemorySaver. En entornos de producción se utilizaría un mecanismo de almacenamiento duradero respaldado por Postgres para que el estado pausado sobreviva a los reinicios del servidor.

Luego, su backend puede enviar un mensaje a Slack a un ingeniero senior con los resultados estructurados. Este lee la advertencia crítica sobre el descuento no validado y, si está de acuerdo, pulsa el botón “Aceptar todo”. Cuando el webhook llega a su servidor, este continúa la ejecución al pasarle al grafo un Command que contiene los datos para reanudar.

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

El grafo se activa, procesa la decisión humana, ajusta los resultados si es necesario y pasa al nodo de emisión 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))]}

Conexión del grafo

Los nodos se agrupan en un StateGraph. No se necesita ningún enlace estático desde el nodo de escalada, ya que ta_escalate devuelve un Command(goto="emit") y LangGraph enruta de forma dinámica.

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

Comparación de los dos

Si se colocan a los dos revisores uno al lado del otro en esta PR, la diferencia es evidente.

El único revisor señala dos problemas. Recomienda agregar pruebas unitarias y sugiere usar indicadores de tipo. Se comporta como un desarrollador junior educado que se enfoca en la sintaxis.

El dúo identifica cuatro problemas. Marca la entrada no validada como un defecto crítico en la lógica de negocio. Señala que los descuentos negativos elevarían los precios. Considera que la implementación inmediata al 100% representa un riesgo para los cobros. Además, identifica un caso límite en el que el descuento se aplica en silencio incluso cuando la bandera de la función está desactivada.

El diseño multiagente no garantiza siempre mejores revisiones. Aquí ayudó porque la descomposición fue excepcionalmente clara. El segundo agente aportó señales reales y no ruido, detectando los defectos en la lógica empresarial que causan problemas en la producción. Funcionó porque la carga cognitiva se distribuyó: el Analizador se encargó de leer en profundidad y mapear las interfaces, mientras que el Revisor recibió un ChangeModel limpio y pudo concentrarse en su labor crítica.

Cuándo no usar múltiples agentes

La combinación de dos agentes es poderosa, y habrá tentación de reescribir cada agente como parte de un flujo de trabajo multiagente. Resístase a ello.

Los agentes adicionales generan mayor latencia. El sistema con dos agentes revisores tarda aproximadamente el doble, ya que sus llamadas a modelos de lenguaje grande son secuenciales, consumen casi el doble de tokens y implican un esquema de estado y lógica de transferencia más complejos. Utilice este patrón solo si la tarea se ajusta a una de las tres situaciones indicadas.

Trabajo adversarial. Si un sistema debe tanto generar como criticar contenido, divida esas funciones. El diseño de múltiples agentes con generador y discriminador es el más probado. Un agente escribe el código y otro escribe pruebas para detectar fallos en él. Uno redacta un correo electrónico y otro lo revisa en busca de problemas de conformidad.

Aislamiento estricto de herramientas. Un agente que busca en la web pública y otro que ejecuta consultas a la base de datos no deberían compartir el mismo sistema. Deje que un Investigador recopile datos públicos y envíe un resumen sanitizado a un Ejecutor. Eso crea una barrera de seguridad sólida. Si una página web contiene inyección de comandos, solo el Investigador se ve comprometido y el Ejecutor permanece aislado.

Escala asimétrica de contexto. A veces, una enorme cantidad de datos sirve para tomar una decisión mínima. Una flota de trabajadores económicos y rápidos puede leer 100 archivos de registro en paralelo, extraer de cada uno las líneas de error relevantes y enviar un breve resumen a un único supervisor costoso que toma la decisión final.

Si una tarea no encaja en ninguno de estos casos, utilice un solo agente. Un agente único con una instrucción bien redactada, restricciones estrictas en las herramientas y un esquema de estado claro superará siempre a una configuración compleja con múltiples agentes.