Аркестрацыя агентаў на Bedrock і LangGraph: чатыры патэрны
Патэрны supervisor–worker, канвеер, fan-out і дэбаты на LangGraph у AWS Bedrock, а таксама перадачы, абмежаванні бюджэту, трасіроўка і тыповыя збоі ў прадакшне.
Крыніцы: Agentic Architectures — Article 6: Multi-Agent Orchestration Patterns
Калі агент ступае на перакантавку
Існуе распашчаная думка, што агент, які выкорыстоўвае добра створаны прыстрой для крэплення, можа справіцца з чым завгодна. Але потым з’являецца праект, які трэба аналізаваць вялікую базу коду, прапорачыць способы захісту на адповідачнасць з рэкамендацыямі па безпецы, стварыць прапанованні ўсунення бягаў і пераканацца, што гэтыя прапанованні працуюць у спадарожчэнні з існуючымі тэстамі, усё гэта за адну працу. Агент выкараняе свой кантэкст. Альбо ён занурваецца ў адны кант коду і забывае пра рэкамендацыях, якія былі запрашаны двадзець тысяч токенав раней. Альбо аналіз завершаецца, калі ў бюджэты не застаёцца нічога на стварэнне якісных способаў усунення бягаў.
Ніякі хакі з вікорыстаннем харнеса не можуць гэта выправіць, незалежна ад таго, чы гэта крашчая інжэкцыя кантэксту, розумна стасумарызацыя чы болей строгая нараджэнне інструментаў. Задача занадта вялікая, каб адны агент мог усё разам бачыць. Дзеяныя проблемы не маюць нічога спакулькаванага з прапаненнямі і выкалываюць патрэбу ў саўместнай працы калькі агентаў. Існуюць чатыры шаблону оркестрацыі, якія працуюць у рэальных умовах на AWS Bedrock з LangGraph, а таксама меры захісту, якія запобегаюць ўпадку. Прынцыпы працы з адным агентам усё ўжо застаюцца актуальнымі: аўтэнтыфікацыя JWT на межах інструментаў, прыборкі цыклаў і выявленне петляў таксама прымаюцца ў мульты-агентных сценарыях і не заменяюцца.
Усе пытанні, на якія даўся адказ, ёсць такімі:
- Калі адны агент перастае быць достатнім?
- Кой шаблон оркестрацыі падходзіць для данай проблемы?
- Як можа кантэкст пераводзіцца межы агентаў без выкарыстоўвання всіх токенав?
- Што ламаецца у рэальных умовах, але не ламаецца ў дэмавым варыянце?
Тры ўзвышэння
Ліміты кантэксту ёсць явным абмежэнням, хоць і не ўсім. Дызайн агента з аднім элементам у канцэ скульгае з трохоў разных прычын.
Выкаранне кантэксту прывлекае найбольшую увагу. Нават вікна на 200 тысяч токенаў запоўняюцца пад час дугіх задач, якія накаплююць выходы інструментаў, проміжныя міркування і історыю размов. Агент пачынае забываць тое, што раней з’явіла, а якосты яго міркування падваўляюцца ў меру запоўнення вікна.
Спецыялізацыя проты генералізацыі ёсць болей тонкай проблемай. Агент, якому кажуць быць хорашым у всім, вяршыць посерэднім у всім. Той, каму трэба ў той жа момент пісаць на Python, аналізаваць наследкі для безпекі і оцэнюваць ступень пакрыцча тэстамі, вынужаны справляцца з трома рознымі способамі мышлення. Спецяліст па безпекі, які пераглядае тое, што створыў агент для генеравання коду, работае лепшая, чым агент, які прабуе робіць гэта сам.
Паралелізацыя — гэта найбольш адказаная магчымасць. Дзеянняя ёсць незалежныя і можа выконваныя адночасова, а яе выкананне адна за іншайцю ў межах адного агента прыводзіць да некорыстнага выкарыстання ўрабочай моці.
LangGraph дапамагае ў гэтым, тым што топалогіі з колькама агентамі ў яму ёсць першага класу. Граф па свойму характару можа адобразіць шаблоны коордынацыі так, як не можа гэта зрабіць секвэнцыйныя фрэймворкі.
Чатыры шаблону проста
Чатыры шаблону не ўзаемна выключаюцься, і у прыкладных сістэмах часта ўжываецца ўзаемная комбінаванне. У гэтай табелі паказана структура кожнага шаблону і яго найкрашчая прыменэ:
+--------------------+--------------------------------------------------+------------------+
| 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 |
+--------------------+--------------------------------------------------+------------------+
Выбірайце на адміністрацыю задання, а не на тое, які варыянт звучыць найбольш прыгожа. Команды выбіраюць фармат дэбатаў, калі звычная структура «надзірнік-рабочы» была бы быстрэйшая і дэшавейшая.
Надзірнік-рабочы
Гэта найчастае выкарыстоўванае патэрн. Агент-надзірнік дзеліць складнае заданне на падзаданні, прадае іх спецыялізаваным рабочым, сабірае тое, што яны вяртаюць, і з’едынае ў фінальны рэзультат. Яго ролю — кераванне, а сама праца застаецца для іншых.
┌─────────────────────────────┐
│ SUPERVISOR AGENT │
│ (task decomposition + │
│ result synthesis) │
└──────┬───────────┬──────────┘
│ │
┌────────▼──┐ ┌────▼──────────┐ ┌──────────────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (code │ │ (security │ │ (test │
│ analysis)│ │ review) │ │ coverage) │
└────────────┘ └───────────────┘ └──────────────┘
LangGraph рэалізуе гэта за дапамою вузла-надзірніка, які надае паведамленні да падграфаў-рабоў вярха ў залежнасці ад таго, на якім етапе зараз знаходзіцца задача. Кожны рабоўскі падграф — гэта окрэма скомпіляваная структура, якая можа выкорыстоваць своія інструменты, ціклы пераканання і механізмы апрацоўкі адзінакоў.
# 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"
Рабочыя намеравана застаюцца лёгкімі: кожны берае точны опіс падзеі і вяртае рэзультат, не ведаючы ніч пра керавальніка чыраў рабочых.
# 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
Потым усё з’еднаецца ў графі стана LangGraph, дзе прыўязаны кантрольны паказчык памяці:
# 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())
У навчальным матэрыялах для керуючага LangGraph адно і тое ж падходзе расказваецца болей дакладна.
Пайплайн
Пайплайн па сваёй прыродзе ёсць последовны. Агент А стварае выходны рэзультат, агент B яго перкаштоввае, а агент C яго ацэнюе; кожны з іх мае чыста візначаную задачу. На дыяграме паказаны выкарыстоўванне дадзеных, ўдосконаленне іх, а потым перакананне ў якосці:
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ (extraction) │────>│ (enrichment) │────>│ (validation) │
└───────────────┘ └───────────────┘ └───────────────┘
output A output B output C
becomes input B becomes input C final result
Складнайшая частка — гэта стысненне кантэксту. Для трэцьяго агента неабходна вся історыя перадачы паведамленняў ад агентаў А і Б, што парадзяе тыясячы токенаў разумовых выводам, якія агенту C не патрэбны. Яму трэба толькі рэзультат работы агента А, а не весь шлях, які да ёй ведае.
# 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
Адзінакоўнае скарыстаецца певнымі наследкамі, і гэтыя наследкі — гэта падышча якосці. Інфармацыя губіцца. Для дзеяных працэсаў гэта некалічна. Але калі агент C павінен следаваць точна тым жа логічным выводам, што і агент A, трэба рэштыцаваляваць, што скарыстаецца адзінакоўна, а што будзе праходзіць без змян.
Паралельны розпад заданняў
Ідэя простая: раздзеліць заданне, запусціць агенты ў той жа час і з’едыніць рэзультаты. У практыцы саме на этапе з’едынення большасць рэалізацый выклікае проблемы.
┌──────────────────┐
│ ORCHESTRATOR │
│ (task splitter) │
└──┬───┬───┬───┬──┘
│ │ │ │
┌──────────▼┐ ┌▼─┐ ┌▼──┐ ┌▼──────────┐
│ Agent 1 │ │A2│ │A3 │ │ Agent 4 │
│ (region A)│ │ │ │ │ │ (region D)│
└──────────┬┘ └┬─┘ └┬──┘ └┬──────────┘
│ │ │ │
┌──▼───▼────▼─────▼──┐
│ AGGREGATOR │
│ (result merger) │
└────────────────────┘
Канкерэнцыя ў Python вырабляецца за дапамою asyncio. У Bedrock гэта проста, адколі кожны паралельны вызов агента ёсць самастоятным запітам 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)
Цена должна адаптавацца да прымкнула да чыстасці. Колькасць агентаў, якія выкананяюць задачу паралельна, лінейна падвайвае расходы: чатыро агентаў, якія працуюць паралельна над адной і той жа задачай, коштаюць прыблізна у чатыро разы больш, чым адны. Гэта зазвычай ўзяцца за прыемнае для справжняя незалежная праца, але ёсць паслядковым рашэнням, калі задачу можна выканаць па спрабах. Дзеякія команды выбіраюць паралельную працу таму, што гэта здаецца болей эфектываў, а не таму, што гэта неабходна для рашэння проблемы. У падрабніце LangGraph пра паралельную експансію вузлаў опісваюцца механізмы розгалужэння.
Дыскусія
Два агентаў незалежна працуюць над адной і той жа задачай, пасля чаго трэційы поручынае ўсе яныя рэзультаты і стварае сінтэз. Гэта дорога і медленная методыка, таму яна ёсць карыстай толькі тады, калі якосьць мае перавагу над швальнасцю: пры важлівых рашэннях, кодзе неабходна высокая безпека, адночасовай пераглядзе прававых дакументаў.
┌─────────────┐ ┌─────────────┐
│ 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,
)
Дыялог паказвае тое, чаго не можа зробіць адзін-агентная ацэнка. Калі два незалежна працюючыя мадэлі дасягаюць аднаго і таго ж выніку, ўпэўненасць стае набагато моцнейшай. Калі ж іхныя вынікі разніцяюцца, сама гэта разніца ёсць сігналам, які указвае на рэальную неяснасць, якой патрэбна людская ацэнка.
Перадача кантэксту між агентамі
Якім бы не быў патэрн, тое, што перадаецца між агентамі, трэба рэшыць адзін у адзін, таму што проблема вікна кантэксту адзінаго агента з’яўляецца знову, але ў новай форме. Перадача всей історыі паведамленняў працюе для двух агентаў, але стае неэфектываю, калі ланцюгі паведамленняў падлягаюць падаўгвенню. Структураваны об’ект для перадачы дастае лепыя рэзультаты, аддзеляючы рэзультаты ад следаў разумовых працэсаў:
# 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)
Фармы працэйскіх недагодзей
Сістэмы з калькамі адчынаюць недагодзеў, якіх няма ў адначленных сістэмах, і дзеянне дзяяў некалькіх калькаў можа быць супертых.
Рост віток. Калі калькі можаюць ствараць падкалькі, вытакі стрэмганна растуць. Калі калька-надзорчык агрэсывна дзеяе, ўнаследак чаго ад однай запыткі корыстувальніка можа стварыцца 20 працоўнікаў, тады неабходныя ёсць строгія ліміты:
# 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
Цыркулярная дэлегавацыя. Такое выкліканне ўсё ж мае меньшы шанс, чым можна было б спадзявацца, а калі яно выйшае, наследкі ўсё ж вельмі значныя. Кантролер дэлегуе задачу Рабоцаму А, Рабочы А прымечае рашэнне дэлегаваць яе знову, і кантролер дэлегуе ўтрымку. Структура графа у LangGraph зазвычай запобегае такой сітуацыі, але дынамічная маршрутазацыя дазволяе яе выйсці.
# 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
Раздробленыя следы. Калі запрос прабывае да пяці агентаў, патрэбны ўсёўказы следу всей графікі, а не пяць нез’ўязаных следоў. LangSmith адпавядае на гэта за дапамогою дрэва запускаў, дзе кожны пад-запуск агента ёсць дзецям свага вышэйшага запуска, таму можна навігаваць па всій іерархіі.
# 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
Урокі, які застаюцься
Сістэмы з калькамі ёсць там, дзе складнасць перастае быць захоплёнай і трэба яе адступова кармачваць.
Адлагоджэнне стае набагато сложнейшым. Неудачны запуск адна-кальковай сістэмы заставляе чакаць толькі адныя паследкі для аналізу, тады калі неудачная сістэма з надзорнікам і працоўнікамі можа заставіць трэбацца аднаццаць паследків для супораджэння. Перш чым інвеставаць у топалогію, трэба інвеставаць у можлівасць спостэрэння. Без ўсёвыходнага відображэння таго, што робяць усе калькі, на рашэнне проблем, якія зазвычай рашаюцца за калекасць хвілін, йдуць дні.
Косты растуць прырастам да колькасці калькаў. Шаблон аргументацыі множыцца для кожнай калькі, і пяці-кальковая сэтапавана сістэма з надзорнікам і працоўнікамі на мадэлях з розгледзам можа выкарыстаць нейкі значны бюджэт на адну запытку. З самага пачатку трэба задаць строгія ліміты бюджэту. Вышэўзгаданы захіст бюджэту не ёсць неабязковым, адтак я он разлічвае кантролюемую модэль костаў ад неспакоюючых рахункоў.
Частыя неудачы — гэта нормальна справа. Адзін агент можа успець чыра зазнаць невяскоў. У працэзе, у яком участвуе калька агентоў, мага з’явіцца сітуацыя, калі пяць з шасцох працавнікаў завершаюць сваю працу, а ўсё ж аднаму не вдаецца, таму значэнне частаг успеху трэба прадзеці загадзя. Якшчо пад час перагляду абяранняў безпекі агент знікае, чы трэба яму вернуць тое, што ён меў, чы ж зазнаць абсалютную невяскоў? Адпаведальныя за продукт должны чынна прыняць такое рашэнне, адколькі гэта бізнесава пытанне, і стандартная поведэнчная схема не павінна яго вялікай меры вплываць.
Абнаўленні модэляў стаюць складнымі. Абяранні ўсё чыр адносуюцца да конкрэтнага модэля, а ўсё ж, калі агенты можаць працаваць на разных модэлях, апдэйт можа значыць прадзець перэнастроек калька сістэмных запрасаў і цыклаў пераканання. Едыны спосаб апдэйтаваць з упэўненасцю — гэта стварэнне набора тэстаў для кожага тыпу агента.
Коордынатары ўтвараюць адзіны момент абякання. Калі керавальнік у модэлі «Керавальнік-рабочы» зупініцца, усё завершваецца. Праекты вырабніцтва павінны зберагаць стан керавальніка, ўпэўніваючыся таму, што праця можа будзе продажана пасля збою. Сістэма LangGraph для стварэння пунктавых контроліў падтрымляе гэта, але яе трэба спецыяльна налаштаваць.
Весь сістэма
Полная архітэктура, ад запіта корыстніка праз керавальнік, рабочыя елементы, перадачу заданняў і можлівасць адзірнання, выглядае так:
┌───────────────────────────────────────────┐
│ 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
А компаненты, тэхналогіяў, якія стояць за кожным з іх, і ўласна ўплата яных падсумаваны тут:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Што яшчэ не ўсё
Калі сістэмы агентаў растуць, апошнія проблема продавжае выступаць, і калькі разоў вышэй згадвалі яе, але не называлі. Кампрэсія канвею, кантракты перадачы дадзеных і надзірнік, які не можа прыгадаць паказаныя раней рэзультаты, — усе гэта вынікае з таго, што агенты забываюць. Кожны запуск пачынаецца з нуля. Надзірнік, які у втарклі рашыў складную проблему, нічога пра яе не ведае, калі у срок устаўае падобная задача, а працавальнік, які рашыў складны крайні кэс, зберагае гэта толькі для аднаго запуску. Памяць, якая застаецца між запускамі, — гэта наступны слой, які трэба стварыць. Яна бывае розных рангаў (рабочая, эпізодычная і семантычная), і на AWS кожны ранг мае свою рэалізацыю.