Коли два агенти перемагають одного: Copilot для перевірки коду LangGraph, створений у обох напрямках
Створіть одного переглядача коду як окремого агента, а також як Аналізатора з переглядачем ризиків у LangGraph, потім дізнайтеся про три випадки, коли використання кількох агентів є доцільним.
Джерела: Multi-Agent Systems: When 2 Agents Beat 1 (and When They Don’t)
„Багатоагентна“ система — це модне слово цього часу. Демонстрації показують п’ять штучних інтелектуальних персон, які спілкуються у симульованій кімнаті, щоб зібрати інформацію з веб-сайту та написати пост у блог, маючи імена на кшталт Дослідник, Письменник та Редактор, а також термінал, заповнений агентами, які дискутують між собою. Це виглядає вражаюче, але саме так не створюють надійний програмний забезпечення.
Додавання агентів саме по собі не робить систему розумнішою. Воно лише збільшує кількість можливих проблем. Візьмемо базову арифметику: агент, який правильно виконує свою роботу у 90% випадків, має рівень надійності 0,90. Якщо поєднати три таких агенти, ймовірності множаться, тож базовий рівень надійності знижується до приблизно 72%. Затримка виконання подвоюється, витрати на API зростають утричі, а кінцевий результат залишається таким самим.
Питання, на яке варто відповісти, — це коли кілька агентів дійсно допомагають у вирішенні проблеми. Хороший спосіб з’ясувати це — створити два асистенти для перегляду коду на основі ШІ. Перша версія — це звичайний переглядач коду з одним агентом. Друга версія розділяє роботу між аналізатором та переглядачем ризиків, використовуючи функції LangGraph, такі як умовні краї, interrupt() та можливість продовження виконання за допомогою Command, щоб у разі потреби залучити людину. Порівняння показує, чому один агент може пропустити серйозну помилку у формуванні рахунку, а чому пара її помічає.
Чому перегляд коду складний за однієї команди
Хороший перегляд коду вимагає двох зовсім різних типів мислення.
Перший — це розуміння. Ви читаєте відмінності, визначаєте функції, які змінилися, визначаєте, які файли постраждали, та виводите, що мав на увазі автор. Метою є створення уявної моделі нової функції.
Другий підхід є адверсарійним. Ви шукаєте крайні випадки, відсутні тести та логічні прогалини, і намагаєтесь зламати код.
Якщо попросити один LLM виконати обидві речі в одному запиті, він буде мати труднощі. Підсумовування та критика в одному етапі зазвичай сприяють послідовності більше, ніж сумнівам. Як тільки модель сформулює версію того, що робить код, вона приділяє менше уваги її критиці та зосереджується на дрібних стилістичних проблемах, таких як відсутність документації чи неправильні правила найменування.
Конкретний приклад це прояснює. Уявіть собі фіктивний pull request до системи білінгу, у якому автор додає функцію знижки до калькулятора рахунків. Ось різниця, яку отримують обидві версії:
--- a/src/billing/invoice.py
+++ b/src/billing/invoice.py
@@ -42,7 +42,9 @@
class InvoiceService:
- def calculate_total(self, items):
- return sum(i.price * i.qty for i in items)
+ def calculate_total(self, items, discount_pct=0):
+ subtotal = sum(i.price * i.qty for i in items)
+ return subtotal * (1 - discount_pct)
--- a/src/billing/api.py
+++ b/src/billing/api.py
@@ -18,6 +18,8 @@
@router.post("/invoice")
def create_invoice(req: InvoiceRequest):
+ discount = req.discount_pct # NEW: from user input
svc = InvoiceService()
- total = svc.calculate_total(req.items)
+ total = svc.calculate_total(req.items, discount)
return {"total": total}
--- a/config/feature_flags.yaml
+++ b/config/feature_flags.yaml
@@ -5,3 +5,4 @@
flags:
new_dashboard: true
+ discount_billing: true # rollout: 100% immediately
Якщо придивитися уважніше, стають помітні кілька серйозних проблем. Значення discount_pct надходить безпосередньо з вхідних даних користувача на рівень API та потрапляє у процес обчислень без жодної перевірки. Користувач може надіслати від’ємний відсоток знижки, щоб підвищити ціну, або значення, більше за 1, щоб отримати від’ємну загальну суму. Тести не охоплюють цей новий сценарій. Крім того, прапорець функції встановлений так, щоб вона одразу стала доступною для 100% користувачів. Це сценарій формування цін, схильний до зловживань. Як із цим впорається типовий переглядач, який працює сам?
Дизайн один: переглядач, який працює сам
Спочатку йде набір інструментів агента: один інструмент для отримання відмінностей, один — для відкриття файлів, один — для пошуку символів та один — для повідомлення про наявні тести.
from langchain_core.tools import tool
import textwrap
MOCK_DIFF = """...""" # The diff shown above
MOCK_FILES = {
"src/billing/invoice.py": "class InvoiceService:\n def calculate_total(self, items, discount_pct=0):\n subtotal = sum(i.price * i.qty for i in items)\n return subtotal * (1 - discount_pct)\n",
"src/billing/api.py": '@router.post("/invoice")\ndef create_invoice(req: InvoiceRequest):\n discount = req.discount_pct\n svc = InvoiceService()\n total = svc.calculate_total(req.items, discount)\n return {"total": total}\n',
"tests/test_billing.py": "def test_calculate_total():\n # only tests no-discount path\n assert svc.calculate_total(items) == 300\n",
}
@tool
def get_diff(pr_id: str) -> str:
"""Fetch the PR diff."""
return MOCK_DIFF
@tool
def read_file(path: str) -> str:
"""Read a file from the repo."""
return MOCK_FILES.get(path, f"FILE NOT FOUND: {path}")
@tool
def search_symbol(name: str) -> str:
"""Search for a symbol across the codebase."""
if "discount" in name.lower():
return "Found: InvoiceService.calculate_total(discount_pct) — src/billing/invoice.py:43"
return f"No results for '{name}'"
@tool
def list_tests(path: str) -> str:
"""List test files covering a source path."""
if "billing" in path:
return "tests/test_billing.py — covers calculate_total (no-discount path only)"
return "No tests found"
ALL_TOOLS = [get_diff, read_file, search_symbol, list_tests]
Далі — держава. TypedDict забезпечує сумісність з LangGraph та містить ID PR, історію повідомлень, необроблений відмінностей та список знахідок. Модель Pydantic описує кожну знахідку та змушує її класифікувати знайдене за допомогою фіксованих літералів, що значно полегшує подальшу обробку.
from typing import Literal
from typing_extensions import TypedDict
from pydantic import BaseModel
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
class ReviewFinding(BaseModel):
severity: Literal["critical", "high", "medium", "low", "info"]
category: Literal["bug", "regression", "missing_test", "rollout_risk",
"security", "edge_case", "style"]
file: str
description: str
confidence: Literal["high", "medium", "low"]
class SingleAgentState(TypedDict):
pr_id: str
messages: list
diff: str
findings: list[ReviewFinding]
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
Далі йдуть вузли. Граф складається з прямої лінії: вузол отримання даних, який завантажує контекст, а потім вузол перевірки, який виконує запит.
def sa_fetch(state: SingleAgentState) -> dict:
diff = get_diff.invoke({"pr_id": state["pr_id"]})
tests = list_tests.invoke({"path": "src/billing"})
return {"diff": diff, "messages": [
AIMessage(content=f"Diff loaded. Test coverage: {tests}")
]}
def sa_review(state: SingleAgentState) -> dict:
"""Single agent does BOTH jobs: summarize + critique in one pass."""
prompt = f"""\
You are a senior code reviewer. Read this PR diff, summarize the change,
and produce a list of risk findings. Be specific.
DIFF:
{state['diff']}
Respond with JSON: {{"summary": "...", "findings": [
{{"severity": "...", "category": "...", "file": "...",
"description": "...", "confidence": "..."}}
]}}
"""
resp = llm.invoke([SystemMessage(content="You are a code review bot."),
HumanMessage(content=prompt)])
# In a real app we parse the JSON response here.
# We simulate the typical single-agent output for this diff.
findings = [
ReviewFinding(
severity="medium",
category="missing_test",
file="tests/test_billing.py",
description="Add unit tests for the new discount_pct parameter.",
confidence="high"
),
ReviewFinding(
severity="low",
category="style",
file="src/billing/invoice.py",
description="Consider adding type hints to the items list.",
confidence="high"
)
]
return {"findings": findings, "messages": [resp]}
def build_single_agent():
g = StateGraph(SingleAgentState)
g.add_node("fetch", sa_fetch)
g.add_node("review", sa_review)
g.add_edge(START, "fetch")
g.add_edge("fetch", "review")
g.add_edge("review", END)
return g.compile()
(У цій демонстрації результати є жорстко закодованими замінниками того типу вихідних даних, які зазвичай створює рецензент, тому сприймайте це порівняння як ілюстрацію, а не як критерій оцінки.)
Приблизно так створюються перші AI-рецензенти: вводиться відмінність коду та запитується про помилки. Після обробки штучного PR агент помічає відсутній тест та пропонує вказівки щодо типів даних. Однак він не помічає невалідованих вхідних даних та ризиків під час впровадження.
Причина криється у запиті. Коли від моделі просять узагальнити зміни та потім перелічити ризики, вона присвячує свою увагу поясненню того, що була додана функція зі знижкою. До моменту, коли вона починає описувати результати, вона вже у описовому режимі, зосереджуючись на поверхневій синтакси та не намагаючись щось зламати. Рішенням є розділення завдань на синтез та аналіз з урахуванням контраргументів, що означає використання двох агентів.
Передача обов’язків: використовуйте схему, а не прозу
Найпоширенішою помилкою у проектуванні багатоагентних систем є дозвіл агентам спілкуватися один з одним у вільному тексті. Якщо агент А пише три абзаци та передає їх агенту Б, той змушений аналізувати текст, здогадуватися, що є важливим, та будувати роботу на основі нечітких даних. Саме так погіршується контекст.
Агенти — це функції, і функції повинні передавати об’єкти з визначеним типом. Щоб багатоагентна система була надійною, необхідно визначити точну структуру того, що передається між агентами.
У асистенті з перегляду першим агентом є Аналізатор, а його завдання — аналізувати запит на зміни. Він має створити об’єкт ChangeModel, який змушує його виділяти конкретні метадані щодо змін.
class ChangedInterface(BaseModel):
file: str
symbol: str
change_type: Literal["added", "modified", "removed"]
description: str
class ChangeModel(BaseModel):
"""Analyzer → Reviewer handoff. Structured, not prose."""
purpose: str
impacted_files: list[str]
changed_interfaces: list[ChangedInterface]
config_changes: list[str]
migration_risk: bool
assumptions: list[str]
tests_touched: list[str]
tests_likely_needed: list[str]
Кожне поле призначене для роботи з наступним агентом. Список змінених інтерфейсів дозволяє другому агенту швидко з’ясувати, що є важливим. Список змін конфігурації допомагає Аналізатору відокремити логіку програми від логіки розгортання.
Масив assumptions заслуговує на особливу увагу. Аналізатор має чітко вказати, що припускає автор коду, наприклад, що discount_pct — це дроб між 0 та 1. Це не є критикою чогось; це просто констатація фактів. Цей структурований об’єкт стає точним вхідним даним для Оглядача.
Дизайн два: Аналізатор та Оглядач ризиків
Нова схема стану містить ChangeModel разом із булевим прапорцем conflict, який згодом визначає, чи потрібно передати справу людині.
class TwoAgentState(TypedDict):
pr_id: str
messages: list
diff: str
change_model: ChangeModel | None
findings: list[ReviewFinding]
conflict: bool
Вузол fetch залишається незмінним, отримуючи інформацію про різниці та рівень покриття тестами. Далі йде вузол Analyzer із дуже суворо визначеними вимогами. Моделі чітко наказують не шукати помилки, тож вона зосереджується виключно на розумінні контексту.
ANALYZER_PROMPT = """\
You are the Analyzer agent. Your ONLY job: understand the change.
Do NOT critique. Do NOT hunt for bugs. Just map what changed.
Produce a structured change model."""
def ta_analyzer(state: TwoAgentState) -> dict:
"""Analyzer: compress & clarify. Optimizes for coherence."""
# We use LLM structured output to fill the ChangeModel.
# For this demonstration, we simulate the accurate analysis.
cm = ChangeModel(
purpose="Add discount percentage support to invoice billing",
impacted_files=["src/billing/invoice.py", "src/billing/api.py",
"config/feature_flags.yaml"],
changed_interfaces=[
ChangedInterface(file="src/billing/invoice.py",
symbol="InvoiceService.calculate_total",
change_type="modified",
description="Added discount_pct param (default 0)"),
ChangedInterface(file="src/billing/api.py",
symbol="create_invoice",
change_type="modified",
description="Reads discount_pct from request, passes to service"),
],
config_changes=["discount_billing flag added, 100% rollout"],
migration_risk=False,
assumptions=[
"discount_pct is expected to be 0..1 (fraction, not percentage)",
"No existing callers pass discount_pct yet",
"Feature flag controls visibility, not the calculation",
],
tests_touched=[],
tests_likely_needed=[
"test discount path in calculate_total",
"test boundary: discount_pct = 0, 1, >1, <0",
"test API validation of discount_pct input",
],
)
return {
"change_model": cm,
"messages": [AIMessage(content=f"Analyzer: change model built. "
f"{len(cm.changed_interfaces)} interfaces changed, "
f"{len(cm.assumptions)} assumptions made.")],
}
Аналізатор читає відмінності та фіксує те, що насправді є: новий параметр, новий флаг функції та неявне припущення про те, що знижка знаходиться в межах від 0 до 1.
Тепер об’єкт потрапляє до спеціаліста з оцінки ризиків, якому навмисно даються конфліктні вказівки. Йому кажуть не довіряти Аналізатору та шукати недоліки.
REVIEWER_PROMPT = """\
You are the Risk Reviewer. The Analyzer gave you a change model.
DISTRUST it. Your job: find what's missing, broken, or dangerous.
Challenge every assumption. Check for missing tests, regressions,
rollout risks, and security issues."""
def ta_reviewer(state: TwoAgentState) -> dict:
"""Reviewer: expand & challenge. Optimizes for skepticism."""
cm = state["change_model"]
findings: list[ReviewFinding] = []
# The Reviewer iterates through the Analyzer's assumptions
for a in cm.assumptions:
if "0..1" in a:
findings.append(ReviewFinding(
severity="critical",
category="security",
file="src/billing/api.py",
description="ASSUMPTION CHALLENGED: discount_pct comes from user input "
"(req.discount_pct) with NO validation. Values <0 or >1 break billing. "
"Negative discount = price increase beyond subtotal. "
"Value >1 = negative total.",
confidence="high"))
# The Reviewer checks the test mapping
if not cm.tests_touched and cm.tests_likely_needed:
findings.append(ReviewFinding(
severity="high",
category="missing_test",
file="tests/test_billing.py",
description=f"NO tests touched but {len(cm.tests_likely_needed)} needed: "
+ "; ".join(cm.tests_likely_needed),
confidence="high"))
# The Reviewer checks the config changes
for cc in cm.config_changes:
if "100%" in cc:
findings.append(ReviewFinding(
severity="high",
category="rollout_risk",
file="config/feature_flags.yaml",
description="Feature flag at 100% from day one — no gradual rollout. "
"Combined with unvalidated discount input, this is a billing incident risk.",
confidence="high"))
# The Reviewer looks for edge cases in the logic flow
findings.append(ReviewFinding(
severity="medium",
category="edge_case",
file="src/billing/api.py",
description="Feature flag controls visibility but calculate_total always applies "
"discount. If flag is off but API still receives discount_pct, discount "
"is silently applied.",
confidence="medium"))
# We determine if there is a conflict worth escalating
conflict = any(f.severity in ("critical", "high") for f in findings)
return {
"findings": findings,
"conflict": conflict,
"messages": [AIMessage(content=f"Reviewer: {len(findings)} findings, "
f"conflict={conflict}")],
}
Оскільки Reviewer не витрачає ресурси на визначення того, які файли змінилися чи навіщо потрібен PR, усі ці дані заважають правильній роботі логіки. Він спирається на припущення, що знижка має бути в межах від 0 до 1, перевіряє відмінності, не знаходить коду для перевірки та позначає помилку перевірки вхідних даних високої серйозності.
Це одне з небагатьох місць, де кілька агентів працюють разом: один створює ціль, а інший намагається її досягти.
Арбітраж із участю людини
Reviewer виявив серйозні проблеми, які автор PR пропустив. Однак оприлюднення їх безпосередньо в GitHub є ризикованим. Якщо Reviewer помиляється, це розлютить команду. Розумним рішенням для продакшну є використання вузла злиття, який оцінює виявлені проблеми та вирішує, чи потрібен арбітраж з участю людини.
def ta_merge(state: TwoAgentState) -> dict:
"""Merge findings. If conflict, surface for human review."""
if state["conflict"]:
return {
"messages": [AIMessage(content=(
"⚠ CONFLICT: Reviewer found critical/high issues. "
"Routing to human review."
))],
}
return {
"messages": [AIMessage(content="Findings merged. No escalation needed.")],
}
def route_after_merge(state: TwoAgentState) -> str:
return "escalate" if state["conflict"] else "emit"
Коли прапорець conflict дорівнює true, потік переходить до вузла ескалації. Саме там використовуються умовні краї, функція interrupt() та возз’єднання на основі Command.
from langgraph.types import Command, interrupt
def ta_escalate(state: TwoAgentState) -> Command:
"""Human-in-the-loop for conflicting findings."""
# The interrupt function pauses execution and surfaces data to the caller
decision = interrupt({
"kind": "review_conflict",
"pr_id": state["pr_id"],
"change_model": state["change_model"].model_dump(),
"findings": [f.model_dump() for f in state["findings"]],
"prompt": "Review findings. Respond: "
'{"action":"accept_all"} or {"action":"override","drop_indices":[...]}',
})
# Execution resumes here when the human provides input
action = decision.get("action", "accept_all")
if action == "override":
drop = set(decision.get("drop_indices", []))
kept = [f for i, f in enumerate(state["findings"]) if i not in drop]
# We use Command to update state and dynamically route to the next node
return Command(update={"findings": kept}, goto="emit")
return Command(goto="emit")
Виклик interrupt() призводить до призупинення виконання. Граф серіалізує свій поточний стан та зберігає його через налаштований індикатор стану — у локальній демонстрації це MemorySaver. У продакшені використовується надійний механізм зберігання, підтримуваний Postgres, щоб стан після призупинення зберігався після перезапуску сервера.
Потім ваш бекенд може надіслати повідомлення в Slack старшому інженеру зі структурованими результатами перевірки. Він читає критичне попередження щодо невалідованої знижки, і якщо погоджується, натискає кнопку „Accept All“. Коли webhook досягає вашого сервера, він продовжує виконання, передаючи графу Command із даними для продовження роботи.
# How you resume the graph from your backend API
ta.invoke(Command(resume={"action": "accept_all"}), config)
Граф прокидається, обробляє рішення людини, за потреби коригує результати та переходить до кінцевого вузла виведення.
def ta_emit(state: TwoAgentState) -> dict:
"""Final output: formatted review."""
lines = [f"=== Code Review: PR {state['pr_id']} ==="]
if state["change_model"]:
lines.append(f"Purpose: {state['change_model'].purpose}")
lines.append(f"Findings ({len(state['findings'])}):")
for i, f in enumerate(state["findings"]):
lines.append(f" [{f.severity}] ({f.category}) {f.file}")
lines.append(f" {f.description}")
return {"messages": [AIMessage(content="\n".join(lines))]}
Підключення графа
Вузли об’єднуються у StateGraph. Не потрібні статичні краї від вузла escalate, оскільки ta_escalate повертає Command(goto="emit"), а LangGraph роутує динамічно.
from langgraph.checkpoint.memory import MemorySaver
def build_two_agent():
g = StateGraph(TwoAgentState)
g.add_node("fetch", ta_fetch)
g.add_node("analyzer", ta_analyzer)
g.add_node("reviewer", ta_reviewer)
g.add_node("merge", ta_merge)
g.add_node("escalate", ta_escalate)
g.add_node("emit", ta_emit)
g.add_edge(START, "fetch")
g.add_edge("fetch", "analyzer")
g.add_edge("analyzer", "reviewer")
g.add_edge("reviewer", "merge")
g.add_conditional_edges("merge", route_after_merge,
{"escalate": "escalate", "emit": "emit"})
g.add_edge("emit", END)
# A checkpointer is required to use interrupt()
# Use MemorySaver for local testing, Postgres for production
return g.compile(checkpointer=MemorySaver())
Порівняння двох
Якщо розмістити обох оцінювачів поруч у цьому PR, різниця стає очевидною.
Один оцінювач вказує на дві проблеми. Він рекомендує додати тести на одиниці та пропонує використовувати типові підказки. Він поводиться як ввічливий молодий розробник, який зосереджується на синтаксисі.
Двоє оцінювачів знаходять чотири проблеми. Вони вказують на невалідований вхідний даний як критичну помилку бізнес-логіки. Вони зазначають, що негативні знижки підвищують ціни. Вони вважають негайне впровадження на 100% ризиком для обліку. Крім того, вони виявляють крайній випадок, коли знижка застосовується беззвучно навіть тоді, коли флаг функції вимкнений.
Дизайн з кількома агентами не гарантує кращих результатів щоразу. Тут він допоміг через надзвичайно чітке розбиття завдань. Другий агент додавав справжні сигнали, а не шум, виявляючи помилки бізнес-логіки, які призводять до зупинки роботи системи. Це спрацювало тому, що когнітивне навантаження було розділене: Аналізатор виконував складну обробку даних та мапував інтерфейси, а Оглядач отримував чистий об’єкт ChangeModel та міг зосередитися на критичному аналізі.
Коли не варто використовувати кілька агентів
Пара агентів є потужною, і може виникнути спокуса переписати кожного агента у формат багатоагентного робочого процесу. Протистояйте цьому.
Додаткові агенти збільшують затримки. Двоагентний механізм огляду вимагає приблизно вдвічі більше часу, оскільки виклики LLM відбуваються послідовно, коштує приблизно вдвічі більше за рахунок кількості токенів, а також потребує більш складної схеми станів та логіки передачі даних. Використовуйте цю схему лише тоді, коли завдання відповідає одній із трьох ситуацій.
Адверсарійна робота. Якщо система має водночас створювати та аналізувати контент, необхідно розділити ці функції. Найкращим доведеним підходом є використання кількох агентів: один пише код, а інший — тести для його перевірки. Один складає електронний лист, а інший перевіряє його на наявність проблем зі відповідністю правилам.
Сувора ізоляція інструментів. Агент, який шукає інформацію в публічному Інтернеті, та агент, який виконує запити до баз даних, не повинні мати спільний „мозок“. Нехай дослідник збирає публічні дані та передає очищений їхній огляд виконавцю. Це створює міцний захисний бар’єр. Якщо веб-сторінка містить атаку типу вставки команд, постраждає лише дослідник, а виконавець залишиться ізольованим.
Асиметричне масштабування контексту. Іноді величезна кількість даних використовується для прийняття незначного рішення. Група дешевих та швидких працівників може одночасно читати 100 файлів журналів, з кожного витягувати відповідні рядки з помилками та надсилати короткий огляд одному дорогому керівнику, який приймає остаточне рішення.
Якщо завдання не підходить до жодної з цих ситуацій, краще використовувати одного агента. Один агент із добре сформульованим запитом, суворими обмеженнями щодо інструментів та чіткою структурою стану завжди буде кращим за хаотичну систему з кількох агентів.
Пов’язана література
- Claude Code Agent Teams: Коли використовувати кількох агентів та як це зробити — Коли достатньо одного агента, чотири причини для додавання більше агентів, шаблони оркестрації, стилі комунікації, вісім способів виникнення помилок та вибір між вбудованими та самостійно створеними рішеннями для команд Claude Code.