Orquestación multiagente en Bedrock y LangGraph: cuatro patrones
Patrones de supervisor-trabajador, pipeline, fan-out y debate con LangGraph en AWS Bedrock, además de transferencias de responsabilidades, protectores de presupuesto, rastreo y modos de fallo en producción.
Fuentes: Agentic Architectures — Article 6: Multi-Agent Orchestration Patterns
Cuando un agente choca contra un muro
Existe la creencia común de que un agente equipado con un arnés bien diseñado puede manejar cualquier cosa. Luego llega un proyecto que requiere escanear una gran base de código, compararla con las advertencias de seguridad, redactar correcciones y verificar esas correcciones contra las pruebas existentes, todo en una sola ejecución. El agente agota su capacidad de procesamiento. O se queda atascado en una parte del código y olvida las advertencias obtenidas veinte mil tokens antes. O el análisis finaliza sin que quede presupuesto alguno para elaborar soluciones adecuadas.
Ningún truco con arneses de automatización soluciona eso, ya sea mediante una mejor inyección de contexto, una resumen más inteligente o un alcance más estricto de las herramientas. La tarea es demasiado amplia como para que un solo agente pueda abarcarla toda a la vez. Algunos problemas no tienen nada que ver con las instrucciones dadas y requieren la cooperación de varios agentes. A continuación se describen cuatro patrones de orquestación que funcionan en producción en AWS Bedrock con LangGraph, junto con las medidas de seguridad que evitan que fallen. Todavía es aplicable el enfoque de un solo agente: la autenticación JWT en los límites de las herramientas, los interruptores de circuito y la detección de bucles también se utilizan en entornos multiagente y no se sustituyen.
Las preguntas a las que se responde en este proceso son las siguientes:
- ¿Cuándo deja de ser suficiente un solo agente?
- ¿Qué patrón de orquestación se adapta a un problema determinado?
- ¿Cómo puede pasar el contexto entre agentes sin agotar el presupuesto de tokens?
- ¿Qué fallas ocurren en producción que nunca ocurren en las pruebas?
Tres límites
Los límites del contexto son la restricción más evidente, aunque no la única. El diseño basado en un único agente eventualmente falla por tres razones distintas.
Agotamiento del contexto recibe la mayor atención. Incluso las ventanas de 200 mil tokens se llenan en tareas largas que acumulan resultados de herramientas, razonamientos intermedios e historial de conversaciones. El agente comienza a olvidar lo que recuperó anteriormente, y su capacidad de razonamiento disminuye a medida que la ventana se llena.
Especialización versus generalización es un tema más sutil. Un agente al que se le pide ser bueno en todo termina siendo mediocre en todo. Uno que debe escribir código en Python, evaluar implicaciones de seguridad y determinar el nivel de cobertura de pruebas al mismo tiempo debe manejar tres modos cognitivos diferentes. Un especialista en seguridad que revisa lo generado por un agente de creación de código lo hace mejor que un único agente intentando hacer ambas cosas.
La paralelización es el límite más subestimado. Algunas tareas son independientes y podrían ejecutarse simultáneamente; hacerlas una tras otra dentro de un único agente deja sin aprovechar el rendimiento.
LangGraph ayuda en esto porque las topologías multiagente son de primera clase en él. Un grafo modela naturalmente los patrones de coordinación de una manera que los frameworks secuenciales no pueden hacer.
Cuatro patrones de un vistazo
Los cuatro patrones no son mutuamente excluyentes, y los sistemas de producción a menudo los combinan. Esta tabla muestra la estructura de cada patrón y su mejor uso:
+--------------------+--------------------------------------------------+------------------+
| Pattern | Structure | Best For |
+--------------------+--------------------------------------------------+------------------+
| Supervisor-Worker | One orchestrator decomposes + delegates | Complex tasks |
| | N workers execute specialized subtasks | needing expert |
| | | decomposition |
+--------------------+--------------------------------------------------+------------------+
| Pipeline | Agent A -> Agent B -> Agent C (sequential) | Transformation |
| | Each processes the previous output | chains, ETL-like |
| | | workflows |
+--------------------+--------------------------------------------------+------------------+
| Parallel Fan-out | Orchestrator sends same/related task to N agents | Research, |
| | Results aggregated into single output | analysis tasks |
| | | that parallelize |
+--------------------+--------------------------------------------------+------------------+
| Debate / Critique | Agent A produces solution, Agent B critiques | High-stakes |
| | Agent C synthesizes or adjudicates | outputs needing |
| | | adversarial QA |
+--------------------+--------------------------------------------------+------------------+
Elija según la estructura de la tarea y no según el patrón que suene más impresionante. Los equipos optan por el modelo de debate cuando una configuración sencilla de supervisor-trabajador habría sido más rápida y económica.
Supervisor-Worker
Este es el patrón más utilizado. Un agente supervisor divide una tarea compleja en subtareas, las asigna a trabajadores especializados, recopila lo que estos devuelven y lo combina en un resultado final. Su función es la gestión, mientras que la ejecución corresponde a otros.
┌─────────────────────────────┐
│ SUPERVISOR AGENT │
│ (task decomposition + │
│ result synthesis) │
└──────┬───────────┬──────────┘
│ │
┌────────▼──┐ ┌────▼──────────┐ ┌──────────────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (code │ │ (security │ │ (test │
│ analysis)│ │ review) │ │ coverage) │
└────────────┘ └───────────────┘ └──────────────┘
LangGraph lo implementa con un nodo supervisor que envía mensajes a los subgrafos trabajadores según el estado actual de la tarea. Cada trabajador es un grafo compilado por separado que puede contar con sus propias herramientas, bucles de verificación y manejo de errores.
# harness/multi_agent/supervisor.py
from typing import Literal, TypedDict, Annotated, List
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_aws import ChatBedrock
from pydantic import BaseModel
import boto3
class SupervisorState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
task_spec: str
subtasks: List[dict] # decomposed work items
worker_results: dict # keyed by subtask ID
current_worker: str # which worker is active
synthesis_complete: bool
agent_run_id: str
class SubtaskAssignment(BaseModel):
"""Structured output from supervisor decomposition."""
subtask_id: str
worker_type: Literal["code_analyst", "security_reviewer", "test_evaluator"]
description: str
depends_on: List[str] # subtask IDs this depends on
priority: int
SUPERVISOR_SYSTEM_PROMPT = """
You are an orchestrator agent. You do not write code or perform analysis yourself.
Your job is to:
1. Break the task into discrete subtasks
2. Assign each subtask to the correct specialist worker
3. Track dependencies between subtasks
4. Synthesize worker outputs into a coherent final result
Available workers:
- code_analyst: Reads and analyzes code structure, dependencies, patterns
- security_reviewer: Evaluates security implications, checks against CVEs
- test_evaluator: Assesses test coverage, identifies gaps
When decomposing, be specific. A subtask description like "analyze the auth module"
is useful. "analyze the code" is not.
Respond in JSON when asked to decompose. Respond in prose when asked to synthesize.
"""
def build_supervisor(region: str = "us-east-1") -> callable:
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Supervisor uses the heavier model — it's doing strategic reasoning
supervisor_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
model_kwargs={
"temperature": 0.2,
"max_tokens": 8000,
"thinking": {"type": "enabled", "budget_tokens": 5000}
}
)
def supervisor_node(state: SupervisorState) -> SupervisorState:
if not state.get("subtasks"):
# First pass: decompose the task
response = supervisor_model.invoke([
SystemMessage(content=SUPERVISOR_SYSTEM_PROMPT),
HumanMessage(content=f"Decompose this task into subtasks:\n{state['task_spec']}")
])
import json
try:
subtasks = json.loads(response.content)
if isinstance(subtasks, dict) and "subtasks" in subtasks:
subtasks = subtasks["subtasks"]
except json.JSONDecodeError:
subtasks = []
return {**state, "subtasks": subtasks}
# All workers done: synthesize
results_summary = "\n\n".join([
f"=== {worker} ===\n{result}"
for worker, result in state["worker_results"].items()
])
synthesis_prompt = f"""
Original task: {state['task_spec']}
Worker results:
{results_summary}
Synthesize these into a coherent final report. Highlight conflicts between
worker findings and make clear recommendations.
"""
response = supervisor_model.invoke([
SystemMessage(content=SUPERVISOR_SYSTEM_PROMPT),
HumanMessage(content=synthesis_prompt)
])
return {
**state,
"messages": state["messages"] + [response],
"synthesis_complete": True,
}
return supervisor_node
def route_to_worker(state: SupervisorState) -> str:
"""
Routes to the next worker with unfinished subtasks.
Returns END when all subtasks are complete and synthesis is done.
"""
if state.get("synthesis_complete"):
return END
# Find next unfinished subtask whose dependencies are met
completed = set(state.get("worker_results", {}).keys())
for subtask in state.get("subtasks", []):
sid = subtask["subtask_id"]
if sid in completed:
continue
deps = set(subtask.get("depends_on", []))
if deps.issubset(completed):
return subtask["worker_type"] # route to this worker
# All subtasks done, back to supervisor for synthesis
return "supervisor"
Los trabajadores mantienen un enfoque intencionalmente sencillo: cada uno toma una descripción precisa de una subtarea y devuelve un resultado, sin saber nada sobre el supervisor ni los demás trabajadores.
# harness/multi_agent/workers.py
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3
CODE_ANALYST_PROMPT = """
You are a code analysis specialist. You receive specific, bounded analysis tasks.
Focus only on what you were asked to analyze. Do not expand scope.
Return structured findings: what you found, confidence level, and specific evidence.
"""
SECURITY_REVIEWER_PROMPT = """
You are a security review specialist. You look for vulnerabilities, insecure patterns,
and CVE-relevant code. Reference specific CWE numbers when applicable.
Return structured findings with severity levels (CRITICAL, HIGH, MEDIUM, LOW).
"""
TEST_EVALUATOR_PROMPT = """
You are a test coverage specialist. You assess test quality and identify gaps.
Focus on: coverage percentage where available, missing edge cases, and untested paths.
Return structured findings with specific test gaps and suggested test cases.
"""
WORKER_PROMPTS = {
"code_analyst": CODE_ANALYST_PROMPT,
"security_reviewer": SECURITY_REVIEWER_PROMPT,
"test_evaluator": TEST_EVALUATOR_PROMPT,
}
def build_worker(worker_type: str, region: str = "us-east-1") -> callable:
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Workers use the faster model — they execute a specific, bounded task
worker_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 4000}
)
system_prompt = WORKER_PROMPTS[worker_type]
def worker_node(state: SupervisorState) -> SupervisorState:
# Find the subtask assigned to this worker type
completed = set(state.get("worker_results", {}).keys())
current_subtask = None
for subtask in state["subtasks"]:
if subtask["worker_type"] == worker_type and subtask["subtask_id"] not in completed:
deps = set(subtask.get("depends_on", []))
if deps.issubset(completed):
current_subtask = subtask
break
if not current_subtask:
return state
# Pass relevant prior results as context if this task has dependencies
context = ""
if current_subtask.get("depends_on"):
for dep_id in current_subtask["depends_on"]:
if dep_id in state.get("worker_results", {}):
context += f"\nPrevious analysis ({dep_id}):\n{state['worker_results'][dep_id]}\n"
prompt = f"Task: {current_subtask['description']}"
if context:
prompt = f"Prior context:{context}\n\n{prompt}"
response = worker_model.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=prompt)
])
updated_results = {**state.get("worker_results", {})}
updated_results[current_subtask["subtask_id"]] = response.content
return {**state, "worker_results": updated_results}
return worker_node
Luego todo se conecta en un grafo de estado LangGraph, con un puntero de verificación de memoria adjunto:
# harness/multi_agent/supervisor_graph.py
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from harness.multi_agent.supervisor import SupervisorState, build_supervisor, route_to_worker
from harness.multi_agent.workers import build_worker
def build_supervisor_graph(region: str = "us-east-1"):
graph = StateGraph(SupervisorState)
graph.add_node("supervisor", build_supervisor(region))
graph.add_node("code_analyst", build_worker("code_analyst", region))
graph.add_node("security_reviewer", build_worker("security_reviewer", region))
graph.add_node("test_evaluator", build_worker("test_evaluator", region))
graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
"supervisor",
route_to_worker,
{
"code_analyst": "code_analyst",
"security_reviewer": "security_reviewer",
"test_evaluator": "test_evaluator",
END: END,
}
)
# All workers route back to supervisor after completing their subtask
for worker in ["code_analyst", "security_reviewer", "test_evaluator"]:
graph.add_edge(worker, "supervisor")
return graph.compile(checkpointer=MemorySaver())
El tutorial del supervisor de LangGraph aborda el mismo enfoque con más detalle.
Pipeline
Un pipeline es secuencial por naturaleza. El Agente A genera la salida, el Agente B la transforma y el Agente C la evalúa, cada uno con una tarea claramente definida. El diagrama muestra la extracción, seguida del enriquecimiento y luego de la validación:
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (extraction) │────>│ (enrichment) │────>│ (validation) │
└───────────────┘ └───────────────┘ └───────────────┘
output A output B output C
becomes input B becomes input C final result
La compresión del contexto es la principal dificultad. Para el tercer agente, transportar toda la historia de mensajes de los agentes A y B implica miles de tokens de razonamiento que el agente C no puede utilizar. Lo que necesita es el resultado del trabajo del agente A, y no el rastro que conduce hasta él.
# harness/multi_agent/pipeline.py
from typing import TypedDict, Annotated, List, Optional, Any
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_aws import ChatBedrock
import boto3
class PipelineState(TypedDict):
original_input: str
stage_outputs: List[dict] # accumulates each stage's compressed result
current_stage: int
final_output: Optional[str]
agent_run_id: str
def compress_for_handoff(full_output: str, model: ChatBedrock) -> str:
"""
Compresses a stage's full output into a structured handoff summary.
This is the core of pipeline context management — the next agent gets
the substance, not the reasoning trace.
"""
response = model.invoke([
SystemMessage(content="""
Compress the following agent output into a structured handoff summary.
Include: key findings, decisions made, artifacts produced, and what the
next stage needs to know. Discard reasoning traces and intermediate steps.
Target length: 20% of original. Use bullet points for clarity.
"""),
HumanMessage(content=f"Compress this:\n\n{full_output}")
])
return response.content
def build_pipeline_stage(
stage_name: str,
system_prompt: str,
region: str = "us-east-1"
) -> callable:
bedrock = boto3.client("bedrock-runtime", region_name=region)
model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 6000}
)
# Cheaper model for compression — this is mechanical, not creative
compressor = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-3-5",
model_kwargs={"temperature": 0, "max_tokens": 2000}
)
def stage_node(state: PipelineState) -> PipelineState:
# Build context from compressed prior stage outputs only
prior_context = ""
for past_stage in state.get("stage_outputs", []):
prior_context += f"\n### {past_stage['stage']} output:\n{past_stage['compressed_output']}\n"
prompt = f"Original task: {state['original_input']}\n"
if prior_context:
prompt += f"\nPrior stage results:\n{prior_context}\n"
prompt += f"\nNow perform your stage: {stage_name}"
response = model.invoke([
SystemMessage(content=system_prompt),
HumanMessage(content=prompt)
])
full_output = response.content
# Compress before storing — next stage won't see raw output
compressed = compress_for_handoff(full_output, compressor)
updated_outputs = list(state.get("stage_outputs", []))
updated_outputs.append({
"stage": stage_name,
"full_output": full_output,
"compressed_output": compressed,
})
return {
**state,
"stage_outputs": updated_outputs,
"current_stage": state.get("current_stage", 0) + 1,
}
return stage_node
No obstante, la compresión tiene un precio, y ese precio es la fidelidad. La información se pierde. Para algunos procesos eso está bien. Pero cuando el Agente C debe seguir exactamente el razonamiento del Agente A, es necesario pensar cuidadosamente en qué se comprime y qué pasa completo.
Distribución paralela
La idea es sencilla: dividir una tarea, ejecutar los agentes al mismo tiempo y combinar los resultados. En la práctica, es en la combinación donde la mayoría de las implementaciones fallan.
┌──────────────────┐
│ ORCHESTRATOR │
│ (task splitter) │
└──┬───┬───┬───┬──┘
│ │ │ │
┌──────────▼┐ ┌▼─┐ ┌▼──┐ ┌▼──────────┐
│ Agent 1 │ │A2│ │A3 │ │ Agent 4 │
│ (region A)│ │ │ │ │ │ (region D)│
└──────────┬┘ └┬─┘ └┬──┘ └┬──────────┘
│ │ │ │
┌──▼───▼────▼─────▼──┐
│ AGGREGATOR │
│ (result merger) │
└────────────────────┘
La concurrencia en Python proviene de asyncio. En Bedrock es sencillo, ya que cada llamada a un agente concurrente constituye una solicitud de API independiente.
# harness/multi_agent/fanout.py
import asyncio
from typing import List, TypedDict, Annotated, Optional
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3
class FanoutResult(TypedDict):
agent_id: str
input_slice: str
output: str
success: bool
error: Optional[str]
async def run_agent_async(
agent_id: str,
input_slice: str,
system_prompt: str,
model: ChatBedrock,
) -> FanoutResult:
"""Run a single agent asynchronously."""
try:
response = await model.ainvoke([
SystemMessage(content=system_prompt),
HumanMessage(content=input_slice)
])
return FanoutResult(
agent_id=agent_id,
input_slice=input_slice,
output=response.content,
success=True,
error=None,
)
except Exception as e:
return FanoutResult(
agent_id=agent_id,
input_slice=input_slice,
output="",
success=False,
error=str(e),
)
async def fan_out(
task_slices: List[str],
system_prompt: str,
region: str = "us-east-1",
max_concurrent: int = 5, # don't hammer Bedrock rate limits
) -> List[FanoutResult]:
"""
Runs agents concurrently with a semaphore to cap parallelism.
max_concurrent protects against Bedrock throttling — you
will hit rate limits if you fire 20 concurrent requests.
"""
bedrock = boto3.client("bedrock-runtime", region_name=region)
model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 4000}
)
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_run(agent_id, slice_content):
async with semaphore:
return await run_agent_async(agent_id, slice_content, system_prompt, model)
tasks = [
bounded_run(f"agent_{i}", slice_content)
for i, slice_content in enumerate(task_slices)
]
return await asyncio.gather(*tasks)
def aggregate_results(
results: List[FanoutResult],
aggregator_model: ChatBedrock,
aggregation_strategy: str = "synthesize",
) -> str:
"""
Merges parallel agent outputs.
aggregation_strategy options:
- "synthesize": ask a model to merge findings coherently
- "concat": simple concatenation (fast, no model call needed)
- "vote": majority-vote for classification tasks
"""
if aggregation_strategy == "concat":
successful = [r for r in results if r["success"]]
return "\n\n---\n\n".join(r["output"] for r in successful)
failed = [r for r in results if not r["success"]]
successful = [r for r in results if r["success"]]
if failed:
# Log partial failures but don't crash — partial results are usually useful
for f in failed:
print(f"Agent {f['agent_id']} failed: {f['error']}")
outputs_for_synthesis = "\n\n".join([
f"[Agent {r['agent_id']}]:\n{r['output']}"
for r in successful
])
response = aggregator_model.invoke([
SystemMessage(content="""
You are a results aggregator. You receive outputs from multiple parallel agents
that each analyzed a different slice of the same problem. Your job is to:
1. Identify common findings across agents
2. Surface unique findings from individual agents
3. Flag any contradictions between agents and explain which to trust
4. Produce a single coherent output as if one expert had analyzed everything
Do not simply concatenate. Actively synthesize.
"""),
HumanMessage(content=f"Synthesize these {len(successful)} agent outputs:\n\n{outputs_for_synthesis}")
])
return response.content
# Convenience wrapper for synchronous callers
def run_parallel_analysis(
task_slices: List[str],
system_prompt: str,
region: str = "us-east-1",
) -> str:
results = asyncio.run(fan_out(task_slices, system_prompt, region))
bedrock = boto3.client("bedrock-runtime", region_name=region)
aggregator = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
model_kwargs={"temperature": 0, "max_tokens": 8000}
)
return aggregate_results(results, aggregator)
El costo merece honestidad. La expansión de agentes multiplica linealmente el gasto en Bedrock: cuatro agentes paralelos trabajando en la misma tarea cuestan aproximadamente cuatro veces más que uno. Esto suele ser aceptable para trabajos verdaderamente independientes, pero es una mala idea cuando la tarea podría ejecutarse de forma secuencial. Algunos equipos eligen esta expansión porque les parece más potente, y no porque el problema lo requiera. La guía de LangGraph sobre la ejecución paralela de nodos describe las mecánicas de ramificación.
Debate
Dos agentes abordan el mismo problema de forma independiente, luego un tercero compara sus resultados y elabora una síntesis. Este método es costoso y lento, por lo que solo resulta rentable cuando la calidad supera a la velocidad: en decisiones de alto riesgo, código sensible desde el punto de vista de la seguridad o revisión de documentos legales.
┌─────────────┐ ┌─────────────┐
│ Proposer │ │ Challenger │
│ (solution │ │ (solution │
│ attempt 1)│ │ attempt 2)│
└──────┬──────┘ └──────┬──────┘
│ │
└──────────┬─────────────┘
▼
┌─────────────────┐
│ ADJUDICATOR │
│ (critique + │
│ synthesis) │
└─────────────────┘
# harness/multi_agent/debate.py
from dataclasses import dataclass
from typing import Optional
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3
@dataclass
class DebateResult:
proposer_solution: str
challenger_solution: str
adjudication: str
final_recommendation: str
agreement_level: str # HIGH | MEDIUM | LOW | CONTRADICTION
PROPOSER_PROMPT = """
You are a solution proposer. Approach the problem carefully and produce your best
solution. Explain your reasoning. Do not hedge excessively — commit to a specific answer.
"""
CHALLENGER_PROMPT = """
You are a solution challenger. You will receive a problem that another agent has
already attempted. Produce your own independent solution WITHOUT seeing their work.
Approach this fresh. Your goal is not to contradict — it's to find the best solution.
"""
ADJUDICATOR_PROMPT = """
You are an adjudicator reviewing two independent solutions to the same problem.
Your job:
1. Identify where the two solutions agree (these are likely correct)
2. Identify where they diverge (these need careful evaluation)
3. For each divergence, evaluate which solution is better and why
4. Produce a final synthesis that takes the best of both
Be direct about contradictions. Do not smooth over genuine disagreements —
surface them clearly so the human reviewer can make a judgment call.
Rate the agreement level: HIGH (minor differences), MEDIUM (some significant
divergences), LOW (fundamentally different approaches), or CONTRADICTION
(mutually exclusive conclusions).
"""
def run_debate(
problem: str,
region: str = "us-east-1",
) -> DebateResult:
bedrock = boto3.client("bedrock-runtime", region_name=region)
heavy_model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
model_kwargs={
"temperature": 0.3, # slight temperature for independent solutions
"max_tokens": 6000,
"thinking": {"type": "enabled", "budget_tokens": 4000}
}
)
# Proposer works the problem
proposer_response = heavy_model.invoke([
SystemMessage(content=PROPOSER_PROMPT),
HumanMessage(content=problem)
])
proposer_solution = proposer_response.content
# Challenger works the same problem independently
# Note: Challenger does NOT see Proposer's solution
challenger_response = heavy_model.invoke([
SystemMessage(content=CHALLENGER_PROMPT),
HumanMessage(content=problem)
])
challenger_solution = challenger_response.content
# Adjudicator sees both and synthesizes
adjudicator_response = heavy_model.invoke([
SystemMessage(content=ADJUDICATOR_PROMPT),
HumanMessage(content=f"""
Problem: {problem}
Solution A (Proposer):
{proposer_solution}
Solution B (Challenger):
{challenger_solution}
Adjudicate and synthesize.
""")
])
adjudication = adjudicator_response.content
# Extract agreement level from adjudication
agreement_level = "MEDIUM"
for level in ["CONTRADICTION", "LOW", "HIGH", "MEDIUM"]:
if level in adjudication.upper():
agreement_level = level
break
return DebateResult(
proposer_solution=proposer_solution,
challenger_solution=challenger_solution,
adjudication=adjudication,
final_recommendation=adjudication,
agreement_level=agreement_level,
)
El debate muestra algo que la evaluación de un solo agente no puede hacer. Cuando dos modelos que razonan de forma independiente llegan a la misma conclusión, la confianza tiene un fundamento mucho más sólido. Cuando difieren, esa divergencia en sí misma es una señal que indica una verdadera ambigüedad que requiere el juicio humano.
Transmisión de contexto entre agentes
Cualquiera que sea el patrón, lo que se transmite entre agentes necesita una consideración cuidadosa, ya que el problema de la ventana de contexto del agente único reaparece bajo una nueva forma. Enviar toda la historia de mensajes funciona para dos agentes, pero deja de ser viable a medida que las cadenas se alargan. Un objeto de transferencia estructurado funciona mejor al separar los resultados de las huellas del razonamiento:
# harness/multi_agent/handoff.py
from dataclasses import dataclass, field
from typing import Any, Optional
from langchain_core.messages import HumanMessage
@dataclass
class AgentHandoff:
"""
Structured context passed between agents.
The split between result and trace is deliberate: downstream agents
need the result, not the full reasoning history. Keeping them separate
lets each agent decide how much context it wants to consume.
"""
source_agent: str
task_completed: str
result_summary: str # compressed, structured result
artifacts: dict = field(default_factory=dict) # files, code, structured data
reasoning_trace: Optional[str] = None # full trace, passed only if downstream needs it
confidence: str = "MEDIUM" # HIGH | MEDIUM | LOW
flags: list = field(default_factory=list) # NEEDS_REVIEW, PARTIAL_RESULT, etc.
def to_context_message(self, include_trace: bool = False) -> HumanMessage:
"""
Converts handoff to a HumanMessage for injection into next agent's context.
include_trace=True only when the downstream agent genuinely needs the reasoning.
"""
content = f"""
[HANDOFF FROM: {self.source_agent}]
Task completed: {self.task_completed}
Confidence: {self.confidence}
Flags: {', '.join(self.flags) if self.flags else 'none'}
Result summary:
{self.result_summary}
"""
if self.artifacts:
content += f"\nArtifacts available:\n"
for key, value in self.artifacts.items():
if isinstance(value, str) and len(value) < 500:
content += f" {key}: {value}\n"
else:
content += f" {key}: [available, {type(value).__name__}]\n"
if include_trace and self.reasoning_trace:
content += f"\nFull reasoning trace:\n{self.reasoning_trace}"
return HumanMessage(content=content)
Modos de fallo en la producción
Los sistemas multiagente presentan fallos que los agentes individuales no tienen, y algunos de ellos son sutiles.
Espirales de costos. Cuando los agentes pueden generar subagentes, los gastos aumentan rápidamente. Un supervisor que se descompone de forma agresiva puede convertir una solicitud del usuario en 20 trabajadores, por lo que se requieren límites estrictos:
# harness/multi_agent/budget.py
import boto3
import time
from decimal import Decimal
class AgentBudgetGuard:
"""
Tracks cumulative cost and agent count per run.
Hard-stops execution when limits are exceeded.
"""
def __init__(
self,
max_agents_per_run: int = 10,
max_total_tokens: int = 500_000,
table_name: str = "agent-budget-state",
region: str = "us-east-1",
):
self.max_agents = max_agents_per_run
self.max_tokens = max_total_tokens
self.table = boto3.resource("dynamodb", region_name=region).Table(table_name)
def register_agent_spawn(self, run_id: str, agent_id: str) -> bool:
"""
Returns True if spawn is allowed, False if budget exceeded.
Call this before spawning any sub-agent.
"""
response = self.table.update_item(
Key={"run_id": run_id},
UpdateExpression="SET agent_count = if_not_exists(agent_count, :z) + :inc",
ExpressionAttributeValues={":z": 0, ":inc": 1},
ReturnValues="UPDATED_NEW",
)
new_count = int(response["Attributes"]["agent_count"])
if new_count > self.max_agents:
raise AgentBudgetExceededError(
f"Run {run_id} attempted to spawn agent #{new_count}, "
f"but max_agents_per_run is {self.max_agents}. "
f"Either the supervisor is over-decomposing, or there is a spawn loop."
)
return True
def record_token_usage(self, run_id: str, tokens_used: int):
response = self.table.update_item(
Key={"run_id": run_id},
UpdateExpression="SET total_tokens = if_not_exists(total_tokens, :z) + :inc",
ExpressionAttributeValues={":z": 0, ":inc": tokens_used},
ReturnValues="UPDATED_NEW",
)
total = int(response["Attributes"]["total_tokens"])
if total > self.max_tokens:
raise AgentBudgetExceededError(
f"Run {run_id} consumed {total:,} tokens, exceeding limit of {self.max_tokens:,}."
)
class AgentBudgetExceededError(Exception):
pass
Delegación circular. Es menos probable que ocurra de lo que uno podría pensar, y cuando sucede el resultado es espectacular. El supervisor delega la tarea al Trabajador A, este decide delegarla de vuelta, y el supervisor vuelve a delegar. La estructura de grafos de LangGraph suele evitar esto, pero la enrutación dinámica lo hace posible.
# harness/multi_agent/deadlock.py
import boto3
import time
from typing import List
class DeadlockDetector:
"""
Tracks the delegation chain per run and detects cycles.
Stored in DynamoDB so it works across parallel agent branches.
"""
def __init__(self, table_name: str = "agent-delegation-chain", region: str = "us-east-1"):
self.table = boto3.resource("dynamodb", region_name=region).Table(table_name)
def record_delegation(self, run_id: str, from_agent: str, to_agent: str):
"""
Records a delegation event and checks for cycles.
Raises DeadlockDetectedError if a cycle is found.
"""
self.table.put_item(Item={
"run_id": run_id,
"delegation_id": f"{from_agent}->{to_agent}-{int(time.time())}",
"from_agent": from_agent,
"to_agent": to_agent,
"timestamp": int(time.time()),
})
chain = self._get_delegation_chain(run_id)
if self._has_cycle(chain):
cycle_description = self._describe_cycle(chain)
raise DeadlockDetectedError(
f"Delegation cycle detected in run {run_id}: {cycle_description}. "
f"Check supervisor decomposition logic for circular dependencies."
)
def _get_delegation_chain(self, run_id: str) -> List[tuple]:
response = self.table.query(
KeyConditionExpression="run_id = :rid",
ExpressionAttributeValues={":rid": run_id}
)
return [(item["from_agent"], item["to_agent"]) for item in response.get("Items", [])]
def _has_cycle(self, chain: List[tuple]) -> bool:
graph = {}
for from_a, to_a in chain:
graph.setdefault(from_a, set()).add(to_a)
visited, rec_stack = set(), set()
def dfs(node):
visited.add(node)
rec_stack.add(node)
for neighbor in graph.get(node, []):
if neighbor not in visited:
if dfs(neighbor): return True
elif neighbor in rec_stack:
return True
rec_stack.discard(node)
return False
return any(dfs(node) for node in graph if node not in visited)
def _describe_cycle(self, chain: List[tuple]) -> str:
return " -> ".join(f"{f}->{t}" for f, t in chain[-5:])
class DeadlockDetectedError(Exception):
pass
Rastros fragmentados. Cuando una solicitud afecta a cinco agentes, se necesita una única vista del rastro de todo el grafo y no cinco rastros desconectados. LangSmith ofrece esto a través de árboles de ejecución, donde cada ejecución de subagente es hijo de su ejecución padre, lo que permite navegar por toda la jerarquía.
# harness/multi_agent/observability.py
import os
from contextlib import contextmanager
from langsmith import Client
from langsmith.run_trees import RunTree
class MultiAgentTracer:
"""
Maintains a run tree across all agents in a multi-agent system.
Pass the parent_run_id to each agent so their traces nest correctly.
"""
def __init__(self):
self.client = Client()
@contextmanager
def agent_span(self, parent_run_id: str, agent_name: str, inputs: dict):
"""
Context manager for an individual agent's trace span.
Usage:
with tracer.agent_span(parent_run_id, "security_reviewer", {...}) as span:
result = run_security_review(...)
span.end(outputs={"result": result})
"""
run = self.client.create_run(
name=agent_name,
run_type="chain",
inputs=inputs,
parent_run_id=parent_run_id,
)
try:
yield run
except Exception as e:
self.client.update_run(run.id, error=str(e))
raise
finally:
self.client.update_run(run.id, end_time=None) # auto-sets end time
Lecciones que perduran
Los sistemas multiagente son aquellos en los que la complejidad deja de ser algo emocionante y debe gestionarse con cuidado.
Depurar se vuelve mucho más difícil. Un fallo en un sistema con un solo agente deja solo una pista para analizar, mientras que un fallo en un sistema supervisor-trabajador puede generar hasta doce pistas que deben correlacionarse. Invierta en capacidad de observabilidad antes que en la topología. Sin una visión clara de lo que están haciendo todos los agentes, se pierden días en resolver problemas que deberían resolverse en minutos.
El costo aumenta con la cantidad de agentes. El patrón de razonamiento en capas se repite en cada agente, y una configuración con cinco agentes supervisor-trabajador en modelos de pensamiento extendido puede consumir una cantidad sorprendente de recursos para una sola solicitud. Establezca límites estrictos de presupuesto desde el primer día. La protección de presupuesto mencionada anteriormente no es opcional, ya que permite separar un modelo de costos controlado de facturas inesperadas.
Un fracaso parcial es normal. Un agente individual puede tener éxito o fallar. En una ejecución con varios agentes, es posible que cinco de seis terminen su tarea mientras que uno falla, por lo que el significado del éxito parcial debe definirse de antemano. Si una revisión de seguridad hace que un agente falle, ¿debería devolver lo que ha procesado o fallar por completo? Los propietarios del producto deben tomar esa decisión de forma explícita, ya que se trata de una cuestión empresarial y el comportamiento predeterminado no debería decidirlo.
Las actualizaciones de modelos se vuelven complicadas. Los mecanismos utilizados son específicos de cada modelo, y dado que los agentes pueden operar en modelos diferentes, una actualización puede requerir ajustar varios comandos del sistema y bucles de verificación. Un conjunto de pruebas para cada tipo de agente es la única forma de realizar actualizaciones con confianza.
Los coordinadores son puntos únicos de fallo. Si el supervisor falla en el patrón Supervisor-Worker, todo se detiene. Los diseños de producción deben conservar el estado del supervisor para que una ejecución interrumpida pueda reanudarse. El sistema de puntos de control de LangGraph lo permite, pero debe configurarse intencionadamente.
El sistema completo
La arquitectura total, desde la solicitud del usuario pasando por el supervisor, los trabajadores, las transferencias de tarea y la capacidad de observabilidad, se ve así:
┌───────────────────────────────────────────┐
│ User Request │
└──────────────────┬────────────────────────┘
│
┌──────────────────▼────────────────────────┐
│ AgentHarness Runtime │
│ (budget guard, deadlock detector, tracer)│
└──────────────────┬────────────────────────┘
│
┌──────────────────▼────────────────────────┐
│ SUPERVISOR / ORCHESTRATOR │
│ Claude 3.7 + extended thinking │
│ Task decomposition + result synthesis │
└────┬──────────────┬──────────────┬────────┘
│ │ │
┌───────────▼──┐ ┌────────▼──┐ ┌───────▼───────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ Claude 3.5 │ │ Claude 3.5 │ │ Claude 3.5 │
│ specialist │ │ specialist │ │ specialist │
└───────┬──────┘ └─────┬─────┘ └──────┬────────┘
│ │ │
┌───────▼───────────────▼────────────────▼────────┐
│ Tool Execution Layer │
│ (auth, retry, circuit breaker from Art.5) │
└───────────────────────┬─────────────────────────┘
│
┌───────────────────────▼──────────────────────────┐
│ AWS Services │
│ Bedrock │ DynamoDB (budget+deadlock+loop+circ) │
│ Secrets Manager │ Knowledge Bases │
└──────────────────────────────────────────────────┘
Observability: LangSmith run trees — full hierarchy per user request
Y los componentes, junto con la tecnología que los sustenta y su función, se resumen aquí:
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Orchestration | LangGraph 0.2+ | Multi-agent graph, routing, |
| | | supervisor-worker topology |
+-----------------------------+---------------------+------------------------------+
| Supervisor Model | Claude 3.7 Sonnet | Task decomposition, |
| | (extended thinking) | result synthesis |
+-----------------------------+---------------------+------------------------------+
| Worker Models | Claude 3.5 Sonnet | Specialized execution, |
| | | bounded tasks |
+-----------------------------+---------------------+------------------------------+
| Parallel Execution | asyncio + Bedrock | Concurrent agent runs with |
| | | semaphore-gated concurrency |
+-----------------------------+---------------------+------------------------------+
| Context Compression | Claude Haiku 3.5 | Pipeline stage handoffs, |
| | | summary generation |
+-----------------------------+---------------------+------------------------------+
| Budget Guard | DynamoDB | Agent count + token limits |
| | | per run |
+-----------------------------+---------------------+------------------------------+
| Deadlock Detection | DynamoDB | Delegation cycle detection |
+-----------------------------+---------------------+------------------------------+
| Loop Detection | DynamoDB (Art. 5) | Per-resource edit tracking |
+-----------------------------+---------------------+------------------------------+
| Circuit Breaker State | DynamoDB (Art. 5) | Shared across all agents |
| | | in a run |
+-----------------------------+---------------------+------------------------------+
| Cross-Agent Observability | LangSmith run trees | Full hierarchy per request |
+-----------------------------+---------------------+------------------------------+
| Auth Propagation | CredentialManager | JWT passed to all workers |
| | (Art. 5) | via execution context |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Ollama + Docker | All patterns testable |
| | Compose | without Bedrock costs |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code | Terraform | DynamoDB tables, IAM roles |
+-----------------------------+---------------------+------------------------------+
Qué falta aún
A medida que crecen los sistemas de agentes, surge un problema constante, y varias secciones anteriores lo mencionaron sin nombrarlo. La compresión de las cadenas de procesamiento, los resúmenes de transferencia y un supervisor que no puede recordar ejecuciones anteriores se deben todos al hecho de que los agentes olvidan. Cada ejecución comienza desde cero. Un supervisor que resolvió un problema complejo el martes no sabe nada al respecto cuando aparece una tarea similar el miércoles, y un trabajador que abordó un caso límite complicado solo retiene esa información para una sola ejecución. La memoria que persiste entre ejecuciones es el siguiente nivel a desarrollar. Se divide en niveles (operativo, episódico y semántico), y en AWS cada nivel cuenta con su propia implementación.
Leer más
- Agentes paralelos de LangGraph que revisan solicitudes de pull request mediante webhooks de FastAPI — Cómo un webhook, una cola de Redis y cuatro agentes de LangGraph convierten una solicitud de pull request en una revisión estructurada y puntuada por IA en aproximadamente 30 segundos, con ngrok para pruebas locales.
- Un agente financiero de QuickBooks que funciona con un LLM local usando Ollama y LangChain — Un agente ReAct basado en Qwen 3.5 9B y una RTX 4070 lee datos en tiempo real de QuickBooks sin ningún costo relacionado con la nube. Consulte el código, los resultados y las limitaciones conocidas.