Когда два агента побеждают одного: Copilot для ревью кода LangGraph, работающий в обоих направлениях
Создайте одного рецензента кода в качестве отдельного агента, а затем — в качестве Анализатора плюс Рецензента рисков в LangGraph; после этого изучите три случая, когда использование нескольких агентов оправдывает затраты.
Источники: Multi-Agent Systems: When 2 Agents Beat 1 (and When They Don’t)
«Мультиагентные» системы — это модный термин сегодняшнего дня. В демонстрациях показано, как пять искусственных интеллектов общаются в симулированной комнате для сбора информации с сайта и написания блог-поста; у них есть такие имена, как Исследователь, Писатель и Редактор, а также терминал, полный сообщений агентов, обсуждающих друг друга. Это выглядит впечатляюще, но именно так не создают надежное программное обеспечение.
Добавление агентов само по себе не делает систему умнее. Оно увеличивает количество возможных сбоев. Возьмем простую арифметику: агент, выполняющий свою задачу правильно в 90% случаев, имеет уровень надежности 0,90. Если связать вместе три таких агента, вероятности умножаются, и базовый уровень надежности снижается примерно до 72%. Задержки удваиваются, расходы на API утраиваются, а конечный результат остается прежним.
Важный вопрос заключается в том, когда выгодно использовать не одного, а несколько агентов. Хороший способ выяснить это — создать двух помощников для ревью кода на основе ИИ. Первая версия представляет собой обычного ревизора с одним агентом. Вторая разделяет работу между анализатором и специалистом по оценке рисков, используя функции LangGraph, такие как условные связи, метод interrupt() и возобновление работы на основе команд, чтобы при необходимости привлечь человека. Сравнение показывает, почему один агент может упустить серьезную ошибку в счетах, а почему два агента её обнаруживают.
Почему ревью кода сложно выполнить с одним запросом
Для качественного ревью требуются два совершенно разных типа мышления.
Первый — это понимание. Вы читаете отчет о различиях, определяете измененные функции, выясняете, какие файлы затронуты, и пытаетесь понять намерения автора. Цель — сформировать в уме модель новой функции.
Второй подход носит конфронтационный характер. Вы ищете крайние случаи, отсутствующие тесты и логические ошибки, и пытаетесь нарушить работу кода.
Если попросить одну большую языковую модель сделать и то, и другое в одном запросе, у неё возникнут трудности. Когда необходимо одновременно подводить итоги и критиковать, это обычно способствует сохранению целостности повествования в ущерб сомнениям. Как только модель сформирует представление о том, что делает код, она тратит меньше усилий на его критику и ограничивается мелкими замечаниями по стилю, такими как отсутствие документации или неправильные правила именования.
Конкретный пример проиллюстрирует это. Представьте фиктивный запрос на внесение изменений в систему биллинга, в котором автор добавляет функцию скидок в калькулятор счетов. Вот разница между двумя версиями кода:
--- 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 пул-реквеста, историю сообщений, необработанные изменения и список находок. Модель 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()
(В этой демонстрации результаты представляют собой заранее заданные примеры того типа выводов, который обычно генерирует рецензент, поэтому рассматривайте это сравнение как иллюстрацию, а не как критерий оценки.)
Примерно так создаются первые версии ИИ-рецензентов: вводится разница между версиями, запрашиваются ошибки. При работе с фиктивным PR агент замечает отсутствующий тест и предлагает указания типов. Однако он упускает невалидированный ввод и риски, связанные с внедрением изменений.
Причина кроется в запросе. Когда от модели требуют кратко описать изменения, а затем перечислить риски, она сосредотачивается на объяснении того, что была добавлена функция скидок. К моменту, когда она начинает составлять список выявленных проблем, она уже находится в описательном режиме, анализируя поверхностную синтаксис и не пытаясь что-либо нарушить. Решение заключается в разделении задач синтеза и анализа на конфликтные сценарии, что подразумевает использование двух агентов.
Передача данных: использовать схему, а не прозу
Самая распространённая ошибка при проектировании многопроцессорных систем — позволять агентам обмениваться информацией в свободном текстовом формате. Если агент A пишет три абзаца и передаёт их агенту B, тот вынужден анализировать текст, догадываться, что важно, и работать на основе нечёткой структуры. Именно так ухудшается контекст.
Агенты представляют собой функции, и функции должны передавать объекты с определённым типом. Чтобы многопроцессорная система была надёжной, необходимо чётко определить схему того, что передаётся между агентами.
В помощнике для ревью первым агентом является Анализатор, задача которого — анализировать патч. Он должен создавать объект 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}")],
}
Поскольку рецензент тратит ресурсы на определение того, какие файлы изменились и в чём заключается смысл пула изменений, это сильно мешает правильной работе логики. Он исходит из предположения, что скидка должна находиться в диапазоне от 0 до 1, проверяет различия между версиями, не находит кода для верификации и отмечает серьёзную ошибку валидации входных данных.
Это одно из немногих мест, где несколько агентов работают вместе: один создаёт целевую версию, а другой проверяет её.
Арбитраж с участием человека
Рецензент выявил серьёзные проблемы, которые упустил автор пула изменений. Однако публикация этих замечаний непосредственно в GitHub сопряжена с рисками. Если рецензент ошибается, это разозлит команду. Разумным решением в производственной среде будет использование узла слияния, который оценит выявленные проблемы и решит, требуется ли арбитраж со стороны человека.
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() и возобновление работы на основе команд.
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 старшему инженеру с структурированной информацией об обнаруженных проблемах. Он прочитает критическое предупреждение о невалидированной скидке, и если согласен, нажмет кнопку «Принять всё». Когда 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, что позволяло ему сосредоточиться на критическом анализе.
Когда не следует использовать несколько агентов
Такая комбинация агентов очень мощна, но может возникнуть соблазн преобразовать каждого агента в элемент многоагентной работы. Сопротивляйтесь этому желанию.
Дополнительные агенты увеличивают задержки. Система с двумя агентами-проверщиками работает примерно в два раза медленнее, поскольку вызовы больших языковых моделей выполняются последовательно, это увеличивает затраты на токены, а также требуется более сложная структура состояний и логика передачи данных. Используйте эту схему только в тех случаях, когда задача соответствует одной из трёх определённых ситуаций.
Работа в условиях конфликта интересов. Если система должна одновременно создавать и анализировать контент, необходимо разделить эти функции. Наиболее эффективным решением является использование двух агентов: генератора и дискриминатора. Один агент пишет код, а другой — тесты для его нахождения ошибок. Один составляет электронное письмо, а другой проверяет его на наличие проблем с соответствием правилам.
Строгая изоляция инструментов. Агент, который ищет информацию в открытом Интернете, и агент, который выполняет запросы к базе данных, не должны использовать один и тот же механизм обработки данных. Пусть исследователь собирает публичные данные и передает их в отфильтрованном виде исполнителю. Это создаёт прочную защитную границу. Если на веб-странице присутствует инъекция команд, пострадает только исследователь, а исполнитель останется изолированным.
Асимметричное масштабирование контекста. Иногда огромный объем данных используется для принятия незначительного решения. Группа дешевых и быстрых агентов может параллельно читать 100 файлов журнала, извлекать из каждого соответствующие строки с ошибками и передавать краткое резюме одному дорогостоящему агенту-руководителю, который принимает окончательное решение.
Если задача не соответствует ни одному из этих случаев, достаточно использовать одного агента. Один агент с хорошо составленным запросом, строгими ограничениями на инструменты и четкой структурой состояния всегда превзойдет сложную систему с несколькими агентами.
Связанные материалы
- Claude Code Agent Teams: Когда использовать несколько агентов и как это сделать — Когда достаточен один агент, четыре причины для добавления большего количества агентов, шаблоны оркестрации, стили общения, восемь способов сбоев и выбор между встроенными решениями и самостоятельной настройкой для команд Claude Code.