Orkiestracja wielu agentów w Bedrock i LangGraph: cztery wzorce
Wzorce nadzoru pracownika, łańcuchów przetwarzania, rozprzestrzeniania się informacji i debat przy użyciu LangGraph w AWS Bedrock, a także mechanizmy przekazywania zadań, ochrony budżetu, śledzenia oraz tryby awarii w produkcji.
Źródła: Agentic Architectures — Article 6: Multi-Agent Orchestration Patterns
Gdy jeden agent napotyka barierę
Powszechnie uważa się, że agent wyposażony w dobrze zaprojektowany pas bezpieczeństwa może poradzić sobie z czymkolwiek. Następnie pojawia się projekt, który wymaga skanowania dużego zbioru kodu, porównania go z zaleceniami bezpieczeństwa, opracowania poprawek oraz weryfikacji tych poprawek pod kątem istniejących testów, wszystko to w jednym przebiegu. Agent wyczerpuje swój kontekst. Albo zagłębia się w jakiś fragment kodu i zapomina o zaleceniach uzyskanych dwadzieścia tysięcy tokenów wcześniej. Albo analiza kończy się bez żadnych środków w budżecie na napisanie skutecznych rozwiązań.
Żaden trik z użyciem harnessu nie rozwiąże tego problemu, czy to lepsza iniekcja kontekstu, inteligentniejsze podsumowywanie, czy bardziej precyzyjne określenie zakresu narzędzi. Zadanie jest zbyt rozległe, by jeden agent mógł je jednocześnie objąć swoją uwagą. Niektóre problemy nie mają nic wspólnego z formułowaniem promptów i wymagają współpracy kilku agentów. Opisano cztery wzorce orkiestracji, które są używane w produkcji na AWS Bedrock z LangGraph, wraz z mechanizmami wzmacniającymi ich stabilność. Nadal obowiązuje podejście oparte na pojedynczym agencie: autoryzacja JWT przy granicach narzędzi, wyłączniki obwodowe oraz detekcja pętli są stosowane również w środowiskach wieloagentowych i nie są zastępowane.
Odpowiedzi na pytania, które pojawiają się w trakcie tego procesu, to:
- Kiedy pojedynczy agent przestaje być wystarczający?
- Który wzorzec orkiestracji pasuje do danego problemu?
- Jak kontekst może być przekazywany między agentami bez nadmiernego zużycia bufora tokenów?
- Co psuje się w produkcji, a nie psuje się w demonstracji?
Trzy rodzaje ograniczeń
Oczywistym, choć nie jedynym ograniczeniem są limity kontekstu. Projektowanie agenta jednoagentowego ostatecznie zawodzi z trzech różnych powodów.
Wyczerpanie kontekstu przyciąga najwięcej uwagi. Nawet okna o pojemności 200 tysięcy tokenów wypełniają się podczas długich zadań, które gromadzą wyniki działań narzędzi, pośrednie rozumowania oraz historię rozmów. Agent zaczyna zapominać to, co wcześniej pobrał, a jego zdolność do rozumowania pogarsza się w miarę wypełniania się okna.
Specjalizacja kontra uogólnianie jest bardziej subtelne. Agent, któremu każe się być dobry we wszystkim, okazuje się przeciętny we wszystkim. Taki, który musi jednocześnie pisać kod w Pythonie, oceniać implikacje bezpieczeństwa i sprawdzać stopień pokrycia testami, musi zarządzać trzema różnymi trybami poznawczymi. Specjalista od bezpieczeństwa, który analizuje wyniki pracy agenta generującego kod, radzi sobie lepiej niż jeden agent próbujący robić to samo.
Paralelizacja to najmniej doceniany potencjał. Niektóre zadania są niezależne i mogą być wykonywane jednocześnie, a ich realizacja po kolei w ramach jednego agenta sprawia, że wydajność pozostaje niewykorzystana.
LangGraph pomaga w tym, ponieważ topologie wieloagentowe są w nim pierwszorzędnym elementem. Graf naturalnie modeluje wzorce koordynacji w sposób, którego nie potrafią frameworki sekwencyjne.
Cztery wzorce na pierwszy rzut oka
Cztery te wzorce nie są ze sobą wykluczające się, a systemy produkcyjne często łączą je ze sobą. W tej tabeli przedstawiono strukturę każdego wzorca oraz jego najlepsze zastosowanie:
+--------------------+--------------------------------------------------+------------------+
| 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 |
+--------------------+--------------------------------------------------+------------------+
Nie wybieraj na podstawie struktury zadania, lecz takiej, której schemat wydaje się najbardziej imponujący. Zespoły decydują się na model debaty wtedy, gdy zwykła struktura nadzorcy-pracownika byłaby szybsza i tańsza.
Nadzorca-Pracownik
To najczęściej używany schemat. Agent nadzorczy dzieli skomplikowane zadanie na podzadania, przydziela je specjalistycznym pracownikom, zbiera to, co ci przekazują, i łączy to w ostateczny rezultat. Jego rolą jest zarządzanie, a wykonywanie zadań pozostawia się innym.
┌─────────────────────────────┐
│ SUPERVISOR AGENT │
│ (task decomposition + │
│ result synthesis) │
└──────┬───────────┬──────────┘
│ │
┌────────▼──┐ ┌────▼──────────┐ ┌──────────────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (code │ │ (security │ │ (test │
│ analysis)│ │ review) │ │ coverage) │
└────────────┘ └───────────────┘ └──────────────┘
LangGraph realizuje to za pomocą węzła nadzorczego, który wysyła wiadomości do podgrafów roboczych w zależności od aktualnego stanu zadania. Każdy podgraf roboczy to oddzielnie skompilowany graf, który może posiadać własne narzędzia, pętle weryfikacyjne oraz mechanizmy obsługi błędów.
# 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"
Pracownicy celowo pozostają lekko zorganizowani: każdy bierze dokładny opis podzadania i zwraca wynik, nie wiedząc nic o kierowniku ani o innych pracownikach.
# 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
Następnie wszystko jest połączone w grafie stanowym LangGraph, z dołączonym wskaźnikiem sprawdzania pamięci:
# 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())
Tutorial dla nadzorcy LangGraph omawia ten sam podejście w większym stopniu szczegółowo.
Kanał przetwarzania
Kanał przetwarzania jest z natury sekwencyjny. Agent A tworzy wynik, Agent B go przekształca, a Agent C go ocenia – każdy z nich ma ściśle określone zadanie. Diagram pokazuje najpierw ekstrakcję, następnie wzbogacanie danych, a potem walidację:
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (extraction) │────>│ (enrichment) │────>│ (validation) │
└───────────────┘ └───────────────┘ └───────────────┘
output A output B output C
becomes input B becomes input C final result
Kompresja kontekstu stanowi główną trudność. Dla trzeciego agenta przenoszenie całej historii wiadomości od agentów A i B oznacza tysiące tokenów rozumowania, które agent C nie może wykorzystać. Potrzebuje on jedynie wyniku pracy agenta A, a nie śladów prowadzących do niego.
# 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
Kompresja ma jednak swoją cenę, a tą ceną jest wierność oryginałowi. Informacje giną. W niektórych przypadkach to nie stanowi problemu. Gdy agent C musi dokładnie naśladować tok rozumowania agenta A, konieczne jest staranne przemyślenie tego, co zostanie skompresowane, a co przejdzie bez zmian.
Rozproszenie równoległe
Idea jest prosta: podzielić zadanie, uruchomić agenty jednocześnie i połączyć wyniki. W praktyce to właśnie etap łączenia jest tym, w którym większość implementacji zawodzi.
┌──────────────────┐
│ ORCHESTRATOR │
│ (task splitter) │
└──┬───┬───┬───┬──┘
│ │ │ │
┌──────────▼┐ ┌▼─┐ ┌▼──┐ ┌▼──────────┐
│ Agent 1 │ │A2│ │A3 │ │ Agent 4 │
│ (region A)│ │ │ │ │ │ (region D)│
└──────────┬┘ └┬─┘ └┬──┘ └┬──────────┘
│ │ │ │
┌──▼───▼────▼─────▼──┐
│ AGGREGATOR │
│ (result merger) │
└────────────────────┘
Konkurencja w Pythonie pochodzi z asyncio. W Bedrock jest to proste, ponieważ każde równoległe wezwanie agenta stanowi odrębną prośbę API.
# 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)
Koszt wymaga uczciwości. Rozgałęzienie zadań mnoży wydatki związane z Bedrockiem liniowo: cztery niezależne agenty pracujące nad tym samym zadaniem kosztują mniej więcej cztery razy tyle co jeden. Jest to zazwyczaj akceptowalne przy naprawdę niezależnej pracy, ale słaba decyzja, gdy zadanie może być wykonywane sekwencyjnie. Niektóre zespoły wybierają takie rozgałęzienie, ponieważ sprawia to wrażenie większej skuteczności, a nie dlatego, że problem tego wymaga. Przewodnik LangGraph dotyczący równoległej eksploatacji węzłów opisuje zasady tworzenia gałęzi.
Dyskusja
Dwa agenty niezależnie rozwiązują to samo zadanie, a następnie trzeci porównuje ich wyniki i tworzy syntezę. Jest to drogie i powolne, więc opłaca się to tylko wtedy, gdy jakość ma większe znaczenie niż szybkość: przy decyzjach o dużym znaczeniu, kodzie wymagającym środków bezpieczeństwa czy przeglądaniu dokumentów prawniczych.
┌─────────────┐ ┌─────────────┐
│ 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,
)
Dyskusja pokazuje to, czego nie potrafi ocena jednego agenta. Gdy dwa modele rozumujące niezależnie dochodzą do tego samego wniosku, ich pewność siebie ma znacznie mocniejsze podstawy. Gdy dochodzi między nimi do rozbieżności, ta sama jest sygnałem wskazującym na rzeczywistą niejednoznaczność, która wymaga ludzkiego osądu.
Przekazywanie kontekstu między agentami
Niezależnie od wzorca, to, co jest przekazywane między agentami, wymaga starannego przemyślenia, ponieważ problem okna kontekstowego jednego agenta pojawia się w nowej formie. Przekazywanie pełnej historii wiadomości działa w przypadku dwóch agentów, ale przestaje być skuteczne wraz ze wzrostem długości łańcuchów. Strukturyzowany obiekt przekazywania radzi sobie lepiej, dzieląc wyniki od śladów rozumowania:
# 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)
Warianty awarii w produkcji
Systemy wielu agentów powodują awarie, których nie ma w systemach z jednym agentem, a niektóre z nich są subtelne.
Spirala kosztów. Gdy agenci mogą tworzyć podagenty, wydatki szybko rosną. Nadzorca, który agresywnie dzieli zadania, może przekształcić jedną prośbę użytkownika w 20 pracowników, dlatego konieczne są ścisłe ograniczenia:
# 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
Kolokwialna delegacja. Zdarza się to rzadziej, niż można by pomyśleć, a gdy już do tego dochodzi, skutki są spektakularne. Kierownik deleguje zadanie Pracownikowi A, Pracownik A postanawia zdelegować je z powrotem, a kierownik deleguje ponownie. Struktura grafowa LangGraph zazwyczaj temu zapobiega, ale dynamiczne routowanie umożliwia to.
# 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
Rozproszone ślady. Gdy żądanie dotyka pięciu agentów, potrzebny jest jeden widok śladu całego grafu, a nie pięć oddzielnych śladów. LangSmith umożliwia to za pomocą drzew wykonywań, w których każde wykonywanie podagenta jest dzieckiem swojego rodzica, dzięki czemu można przemieszczać się po całej hierarchii.
# 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
Lekcje, które pozostają
Systemy wielu agentów to sytuacje, w których złożoność przestaje być ekscytująca i wymaga ostrożnego zarządzania.
Diagnozowanie staje się znacznie trudniejsze. Nieudana próba działania pojedynczego agenta pozostawia jeden ślad do analizy, natomiast nieudany system nadzorcy-pracownika może pozostawić aż dwanaście śladów do skorelowania. Inwestuj w możliwości obserwacji przed inwestowaniem w topologię. Bez jednolitego widoku tego, co robią wszystkie agenty, na rozwiązanie problemów, które powinny zająć kilka minut, tracimy dni.
Koszty rosną wraz z liczbą agentów. Wzorzec rozumowania typu „kanapka” jest powielany dla każdego agenta, a pięcioagentowy układ nadzorcy-pracownika w modelach o rozszerzonej zdolności myślenia może zużyć zaskakująco dużo zasobów na jedną prośbę. Ustal sztywne limity budżetu od pierwszego dnia. Powyższy mechanizm kontroli budżetu nie jest opcjonalny, ponieważ oddziela kontrolowany model kosztów od niespodziewanych rachunków.
Częściowa niepowodzenie jest normalne. Pojedynczy agent może odnieść sukces lub ponieść porażkę. W symulacjach z wieloma agentami może się zdarzyć, że pięć z sześciu „pracowników” ukończy zadanie, a jeden zawiedzie, dlatego znaczenie częściowego sukcesu musi zostać ustalone z góry. Jeśli weryfikacja bezpieczeństwa pozbawi się jednego agenta, czy powinien on zwrócić to, co ma, czy ponieść całkowitą porażkę? Właściciele produktów powinni jasno podjąć tę decyzję, ponieważ jest to kwestia biznesowa, a domyślne zachowanie nie powinno jej decydować.
Uaktualniania modeli stają się skomplikowane. Narzędzia są specyficzne dla danego modelu, a ponieważ agenci mogą pracować na różnych modelach, aktualizacja może oznaczać konieczność ponownego dostosowania kilku instrukcji systemowych oraz pętli weryfikacyjnych. Jedynym sposobem na pewne aktualizowanie jest posiadanie zestawu testów dla każdego typu agenta.
Koordynatorzy stanowią pojedyncze punkty awarii. Jeśli w schemacie Supervisor-Worker nadzorca przestanie działać, wszystko się zatrzymuje. Projekty produkcyjne powinny przechowywać stan nadzorcy, aby można było wznowić zatrzymaną pracę. LangGraph umożliwia takie przechowywanie stanów, ale wymaga to celowej konfiguracji.
Cały system
Pełna architektura, od żądania użytkownika, przez nadzorcę, roboty, procesy przekazywania zadań po możliwości obserwacji, wygląda w ten sposób:
┌───────────────────────────────────────────┐
│ 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
A komponenty wraz z technologią stojącą za każdym z nich oraz ich rolą są podsumowane tutaj:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Co jeszcze brakuje
W miarę rozwoju systemów agentowych pojawia się jeden powtarzający się problem, który został wspomniany w kilku wcześniejszych sekcjach, choć bez podania jego nazwy. Kompresja łańcucha przetwarzania, streszczenia przenoszonych danych oraz nadzorca, który nie może przywołać informacji z wcześniejszych wykonań, wynikają wszystkie z faktu, że agenci zapominają. Każde wykonywanie zaczyna się od zera. Nadzorca, który we wtorek rozwiązał trudny problem, nic o nim nie wie, gdy w środę pojawia się podobne zadanie, a pracownik, który uporał się z skomplikowanym przypadkiem szczególnym, pamięta to tylko przy jednym wykonywaniu. Pamięć, która przetrwa między kolejnymi wykonywaniami, to następny element do opracowania. Dzieli się ona na poziomy (pracujący, epizodyczny i semantyczny), a w AWS każdy z tych poziomów ma swoją własną implementację.
Literatura pokrewna
- Agenty LangGraph działające równolegle, które przeglądają pull requests za pomocą webhooków FastAPI — Jak webhook, kolejka Redis oraz cztery agenty LangGraph przekształcają pull request w uszeregowany, strukturyzowany przegląd AI w ciągu około 30 sekund, przy użyciu ngrok do testów lokalnych.
- Agent finansowy QuickBooks działający na lokalnym modelu LLM z Ollama i LangChain — Agent ReAct na bazie Qwen 3.5 9B oraz karty RTX 4070 odczytuje dane z QuickBooks w czasie rzeczywistym bez żadnych kosztów chmurowych. Zobacz kod, wyniki oraz znane ograniczenia.