Orchestration multi-agents sur Bedrock et LangGraph : quatre modèles
Modèles de supervision-travailleur, de pipeline, de diffusion et de débat avec LangGraph sur AWS Bedrock, ainsi que les transferts de responsabilités, les mécanismes de contrôle budgétaire, le suivi et les modes de défaillance en production.
Sources: Agentic Architectures — Article 6: Multi-Agent Orchestration Patterns
Lorsqu’un agent bute contre un mur
On pense souvent qu’un agent équipé d’un harnais bien conçu peut gérer n’importe quelle tâche. Puis arrive un projet qui exige de scanner une grande base de code, de la comparer aux alertes de sécurité, de rédiger des correctifs et de vérifier ces correctifs à l’aide des tests existants, tout cela en une seule exécution. L’agent épuise alors son contexte. Il peut aussi se concentrer sur une partie du code et oublier les alertes consultées vingt mille tokens plus tôt. Enfin, l’analyse se termine sans qu’il reste de budget pour élaborer de bons correctifs.
Aucun truc lié aux harnais ne résout ce problème, que ce soit une injection de contexte améliorée, une synthèse plus intelligente ou un ciblage plus précis des outils. La tâche est trop vaste pour qu’un seul agent puisse la gérer en même temps. Certains problèmes n’ont rien à voir avec les instructions fournies et exigent la coopération de plusieurs agents. Quatre modèles d’orchestration fonctionnent en production sur AWS Bedrock avec LangGraph, ainsi que des mesures de renforcement visant à éviter leurs défaillances. Les principes liés aux harnais pour un seul agent restent applicables : l’authentification JWT aux frontières des outils, les disjoncteurs et la détection de boucles s’étendent également au domaine multi-agent sans être remplacés.
Les questions auxquelles on répond au fil de cette analyse sont les suivantes :
- Quand un seul agent cesse-t-il d’être suffisant ?
- Quel modèle d’orchestration convient à un problème donné ?
- Comment le contexte peut-il être transmis entre les agents sans épuiser le budget de tokens ?
- Qu’est-ce qui dérape en production alors qu’il ne pose aucun problème lors des démonstrations ?
Trois plafonds
Les limites liées au contexte constituent la contrainte évidente, mais pas la seule. La conception basée sur un seul agent échoue finalement pour trois raisons distinctes.
L’épuisement du contexte attire le plus d’attention. Même des fenêtres de 200 000 tokens se remplissent rapidement lors d’opérations longues qui génèrent des résultats d’outils, des raisonnements intermédiaires et un historique de conversation. L’agent commence alors à oublier ce qu’il a récupéré précédemment, et sa capacité de raisonnement se détériore à mesure que la fenêtre se remplit.
Spécialisation versus généralisation est une question plus subtile. Un agent chargé d’être bon en tout se révèle médiocre en tout. Celui qui doit écrire du Python, évaluer les implications de sécurité et juger la couverture des tests en même temps doit gérer trois modes cognitifs différents. Un spécialiste de la sécurité qui examine ce qu’un agent de génération de code a produit le fait mieux qu’un seul agent tentant de faire les deux choses à la fois.
La parallélisation est le potentiel le plus sous-estimé. Certaines tâches sont indépendantes et peuvent être exécutées en même temps ; les effectuer une après l’autre au sein d’un seul agent fait perdre en performance.
LangGraph est utile à cet égard, car les topologies multi-agents y sont prises en compte de manière prioritaire. Un graphe permet naturellement de modéliser les schémas de coordination d’une façon que les frameworks séquentiels ne peuvent pas faire.
Quatre schémas en un coup d’œil
Ces quatre schémas ne sont pas mutuellement exclusifs, et les systèmes de production les combinent souvent. Ce tableau présente la structure de chaque schéma ainsi que son utilisation optimale :
+--------------------+--------------------------------------------------+------------------+
| 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 |
+--------------------+--------------------------------------------------+------------------+
Choisissez en fonction de la structure de la tâche et non selon le modèle qui semble le plus impressionnant. Les équipes recourent au débat lorsque une configuration simple de superviseur-travailleur aurait été plus rapide et moins coûteuse.
Superviseur-Travailleur
C’est le modèle le plus utilisé. Un agent superviseur divise une tâche complexe en sous-tâches, les attribue à des travailleurs spécialisés, rassemble ce qu’ils lui renvoient et le fusionne en un résultat final. La gestion en est sa fonction, tandis que l’exécution est confiée aux autres.
┌─────────────────────────────┐
│ SUPERVISOR AGENT │
│ (task decomposition + │
│ result synthesis) │
└──────┬───────────┬──────────┘
│ │
┌────────▼──┐ ┌────▼──────────┐ ┌──────────────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (code │ │ (security │ │ (test │
│ analysis)│ │ review) │ │ coverage) │
└────────────┘ └───────────────┘ └──────────────┘
LangGraph le met en œuvre à l’aide d’un nœud superviseur qui envoie des messages aux sous-graphes travailleurs en fonction de l’état actuel de la tâche. Chaque travailleur est un graphique compilé séparément qui peut disposer de ses propres outils, boucles de vérification et mécanismes de gestion des erreurs.
# 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"
Les travailleurs maintiennent délibérément un niveau de complexité faible : chacun prend en charge une description précise d’une sous-tâche et fournit un résultat, sans rien savoir du superviseur ni des autres travailleurs.
# 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
Ensuite, tout est relié entre eux dans un graphe d’état LangGraph, doté d’un point de contrôle mémoire :
# 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())
Le tutoriel sur le superviseur LangGraph aborde la même approche de manière plus détaillée.
Tuyau de traitement
Un tuyau de traitement est par nature séquentiel. L’agent A crée la sortie, l’agent B la restructure et l’agent C la juge, chacun ayant une tâche strictement définie. Le diagramme montre d’abord l’extraction, puis l’enrichissement, enfin la validation :
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (extraction) │────>│ (enrichment) │────>│ (validation) │
└───────────────┘ └───────────────┘ └───────────────┘
output A output B output C
becomes input B becomes input C final result
La compression du contexte constitue la principale difficulté. Avec le troisième agent, transporter l’ensemble de l’historique des messages provenant des agents A et B signifie gérer des milliers de tokens de raisonnement dont l’agent C n’a pas besoin. Ce qu’il nécessite, c’est le résultat du travail de l’agent A, et non la trace menant à ce résultat.
# 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
Cependant, la compression a un prix, et ce prix est la fidélité. Des informations se perdent. Cela peut être acceptable pour certains flux de travail. Lorsque l’Agent C doit suivre exactement le raisonnement de l’Agent A, il faut réfléchir soigneusement à ce qui est compressé et à ce qui passe intégralement.
Déploiement parallèle
L’idée est simple : diviser une tâche, exécuter les agents en même temps, puis fusionner les résultats. En pratique, c’est lors de la fusion que la plupart des implémentations échouent.
┌──────────────────┐
│ ORCHESTRATOR │
│ (task splitter) │
└──┬───┬───┬───┬──┘
│ │ │ │
┌──────────▼┐ ┌▼─┐ ┌▼──┐ ┌▼──────────┐
│ Agent 1 │ │A2│ │A3 │ │ Agent 4 │
│ (region A)│ │ │ │ │ │ (region D)│
└──────────┬┘ └┬─┘ └┬──┘ └┬──────────┘
│ │ │ │
┌──▼───▼────▼─────▼──┐
│ AGGREGATOR │
│ (result merger) │
└────────────────────┘
La concurrence en Python provient de asyncio. Sur Bedrock, c’est simple, car chaque appel d’agent concurrentuel constitue une requête API distincte.
# 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)
Le coût exige de l’honnêteté. La multiplication des agents augmente linéairement les dépenses liées à Bedrock : quatre agents travaillant en parallèle sur la même tâche coûtent environ quatre fois plus cher qu’un seul agent. Cela est généralement acceptable pour des tâches véritablement indépendantes, mais c’est une mauvaise idée lorsque la tâche peut être exécutée séquentiellement. Certaines équipes choisissent cette approche parce qu’elle semble plus puissante, et non parce que le problème l’exige. Le guide de LangGraph sur l’exécution parallèle des nœuds décrit les mécanismes de ce type d’exécution.
Débat
Deux agents traitent indépendamment le même problème, puis un troisième compare leurs résultats et rédige une synthèse. Cette méthode est coûteuse et lente, donc elle ne s’avère rentable que lorsque la qualité prime sur la vitesse : décisions à haut risque, code sensible en matière de sécurité, révision de documents juridiques.
┌─────────────┐ ┌─────────────┐
│ 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,
)
Le débat montre ce que l’évaluation par un seul agent ne peut pas faire. Lorsque deux modèles qui raisonnent de manière indépendante parviennent à la même conclusion, la confiance a des bases bien plus solides. Lorsqu’ils divergent, cette divergence constitue en elle-même un signal indiquant une véritable ambiguïté qui nécessite le jugement humain.
Transfert de contexte entre agents
Quel que soit le schéma, ce qui est transféré entre agents exige une réflexion approfondie, car le problème de la fenêtre de contexte pour un seul agent réapparaît sous une nouvelle forme. Le transfert de l’historique complet des messages fonctionne pour deux agents, mais devient inefficace à mesure que les chaînes s’allongent. Un objet de transfert structuré est plus efficace, car il sépare les résultats des traces de raisonnement :
# 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)
Modèles de défaillance de la production
Les systèmes multi-agents entraînent des défaillances que les agents individuels ne connaissent pas, et certaines d’entre elles sont subtiles.
Spirale des coûts. Lorsque les agents peuvent générer des sous-agents, les dépenses augmentent rapidement. Un superviseur qui procède à une décomposition trop agressive peut transformer une seule demande utilisateur en 20 travailleurs ; il est donc nécessaire d’établir des limites strictes :
# 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
Délégation circulaire. Il est moins probable que cela se produise qu’on ne le pense, et lorsque c’est le cas, les conséquences sont spectaculaires. Le superviseur délègue à l’Employé A, celui-ci décide de déléguer à son tour, et le superviseur délègue à nouveau. La structure de graphe de LangGraph empêche généralement cela, mais le routage dynamique le rend possible.
# 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
Traces fragmentées.Lorsqu’une requête concerne cinq agents, il est nécessaire d’avoir une seule vue des traces représentant l’ensemble du graphe, et non cinq traces disjointes. LangSmith permet d’y parvenir grâce aux arbres d’exécution, où chaque exécution de sous-agent est un enfant de son exécution parente, ce qui permet de naviguer dans toute la hiérarchie.
# 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
Leçons durables
Les systèmes multi-agents sont là où la complexité cesse d’être intéressante et doit être gérée avec soin.
Le débogage devient beaucoup plus difficile. Une exécution échouée avec un seul agent ne laisse qu’une trace à analyser, tandis qu’un système superviseur-travailleur défaillant peut en laisser douze à croiser. Investissez dans l’observabilité avant d’investir dans la topologie. Sans une vue globale de ce que font tous les agents, des jours sont consacrés à des problèmes qui devraient être résolus en quelques minutes.
Le coût augmente avec le nombre d’agents. Le schéma de raisonnement en sandwich est reproduit pour chaque agent, et une configuration superviseur-travailleur composée de cinq agents sur des modèles à réflexion étendue peut consommer une quantité étonnante de tokens pour une seule requête. Fixez des limites budgétaires strictes dès le premier jour. La protection budgétaire mentionnée ci-dessus n’est pas optionnelle, car elle permet de distinguer un modèle de coûts contrôlé d’une facture inattendue.
Une défaillance partielle est normale. Un agent isolé peut réussir ou échouer. Lors d’un essai avec plusieurs agents, il est possible que cinq sur six terminent leur tâche tandis qu’un autre échoue, il faut donc définir à l’avance ce que signifie un succès partiel. Si une analyse de sécurité fait échouer un agent, celui-ci doit-il restituer ce qu’il a accompli ou échouer complètement ? Les responsables du produit doivent trancher explicitement cette question, car il s’agit d’une décision commerciale et le comportement par défaut ne devrait pas en décider à sa place.
Les mises à jour des modèles deviennent complexes. Les outils utilisés sont spécifiques à chaque modèle, et comme les agents peuvent être basés sur des modèles différents, une mise à jour peut nécessiter de réajuster plusieurs instructions système et boucles de vérification. Un ensemble de tests pour chaque type d’agent est le seul moyen de procéder à une mise à jour en toute confiance.
Les coordinateurs constituent des points de défaillance uniques. Si le superviseur tombe en panne dans le modèle Superviseur-Travailleur, tout s’arrête. Les conceptions de production doivent conserver l’état du superviseur afin que la séance interrompue puisse reprendre. Le mécanisme de sauvegarde de LangGraph permet cela, mais il doit être configuré délibérément.
Tout le système
L’architecture complète, de la demande utilisateur en passant par le superviseur, les travailleurs, les transferts et l’observabilité, est présentée comme suit :
┌───────────────────────────────────────────┐
│ 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
Les composants, ainsi que la technologie qui les sous-tend et leur rôle, sont résumés ici :
+-----------------------------+---------------------+------------------------------+
| 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’est-ce qui manque encore
Au fur et à mesure que les systèmes d’agents se développent, un problème réapparaît constamment, et plusieurs sections précédentes l’ont abordé sans le nommer. La compression des pipelines, les résumés de transfert et un superviseur incapable de se remémorer les exécutions antérieures trouvent leur origine dans le fait que les agents oublient. Chaque exécution commence à zéro. Un superviseur qui a résolu un problème complexe mardi ne sait rien à ce sujet lorsque une tâche similaire apparaît mercredi, et un agent qui a géré un cas particulièrement difficile ne conserve cette connaissance que pour une seule exécution. Une mémoire persistante entre les exécutions constitue la prochaine étape à développer. Elle existe en plusieurs niveaux (de travail, épisodique et sémantique), et sur AWS chaque niveau dispose de sa propre implémentation.