Простой агентный цикл — LLM получает задачу, вызывает инструмент, смотрит на результат, повторяет — работает. Но только пока задача маленькая и всё идёт по плану. Как только появляются параллельные зависимости, ошибки инструментов или необходимость доказать, что агент сделал именно то, что нужно, цикл ломается. Этот гайд показывает, как последовательно добавить к нему планирование, параллельное выполнение, верификацию и трассировку — на реальном Python-коде.
Почему простого цикла не хватает
Базовый harness выглядит примерно так: модель думает, выбирает инструмент, получает наблюдение, снова думает. Это корректно для линейных задач. Но возьмём задачу сравнения трёх городов по трём атрибутам — население, часовой пояс, описание. Это девять вызовов инструментов, каждый из которых независим от остальных, но итоговый отчёт зависит от всех девяти. Последовательный цикл выполнит их один за другим. Параллельный — за время самого медленного.
Кроме линейности, у наивного цикла есть ещё три проблемы: нет разделения ответственности (одна модель планирует и исполняет), нет верификации результата (агент сам решает, что справился), нет бюджета (цикл может крутиться бесконечно). Дальше — как это починить по шагам.
Шаг 1. Абстрагируй LLM-провайдера
Прежде чем строить что-либо поверх, нужна точка подключения мозга, которую можно заменить без переписывания всего harness. Определи базовый класс:
class LLMProvider:
"""Shared interface. Subclass to plug in a different backend."""
def complete(self, system: str, user: str, role: str = "default") -> str:
raise NotImplementedError
async def acomplete(self, system: str, user: str, role: str = "default") -> str:
return await asyncio.to_thread(self.complete, system, user, role)Асинхронный метод acomplete оборачивает синхронный вызов через asyncio.to_thread — это работает с любым SDK без переписывания. Для тестов реализуй MockProvider, который возвращает детерминированные ответы. Тогда весь harness можно прогнать без сети и без реальных токенов.
Шаг 2. Добавь DAG и параллельное выполнение
Граф зависимостей (DAG — directed acyclic graph) — это список задач, где каждая задача знает, от каких других она зависит. Планировщик строит граф, исполнитель запускает задачи, у которых все зависимости уже выполнены.
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class Task:
id: str
tool: str
args: dict
depends_on: List[str] = field(default_factory=list)
result: Optional[str] = None
status: str = "pending" # pending | running | done | failedПараллельный запуск задач без зависимостей — через asyncio.gather:
async def run_ready_tasks(tasks: dict[str, Task], provider: LLMProvider):
ready = [
t for t in tasks.values()
if t.status == "pending"
and all(tasks[dep].status == "done" for dep in t.depends_on)
]
await asyncio.gather(*[execute_task(t, provider) for t in ready])Для задачи сравнения трёх городов это означает: девять вызовов инструментов запускаются параллельно, агрегация запускается только после того, как все девять завершились. Общее время выполнения падает с суммы до максимума.
Шаг 3. Разделяй роли: Planner, Worker, Critic
Один агент, который сам себе планировщик и сам себе судья — плохая идея. Разделение ролей решает это структурно.
| Роль | Что делает | Кто вызывает |
|---|---|---|
| Planner | Разбивает цель на задачи, строит DAG | Оркестратор, один раз |
| Worker | Исполняет одну задачу, вызывает инструмент | Оркестратор, по задаче |
| Critic | Проверяет итоговый результат по критериям | Оркестратор, в конце |
Planner получает цель и возвращает список задач с зависимостями. Worker получает одну задачу и возвращает наблюдение. Critic получает цель и финальный ответ и возвращает pass или список нарушений.
def run_critic(goal: str, answer: str, provider: LLMProvider) -> dict:
system = "You are a strict verifier. Return JSON: {passed: bool, issues: list[str]}"
user = f"Goal: {goal}\n\nAnswer: {answer}"
raw = provider.complete(system=system, user=user, role="critic")
return json.loads(raw)Если Critic возвращает passed: false, оркестратор может запустить второй цикл с исправлением — или зафиксировать провал с объяснением.
Шаг 4. Введи многомерный бюджет
Бюджет — это не только лимит шагов. Реальная система контролирует несколько измерений одновременно:
@dataclass
class Budget:
max_steps: int = 20
max_tool_calls: int = 30
max_latency_ms: int = 60_000
steps_used: int = 0
tool_calls_used: int = 0
total_latency_ms: int = 0
def has_room(self) -> bool:
return (
self.steps_used < self.max_steps
and self.tool_calls_used < self.max_tool_calls
and self.total_latency_ms < self.max_latency_ms
)Аналог сигнала bingo fuel в авиации: агент знает, что ресурс заканчивается, и может вернуть частичный результат вместо того, чтобы упасть без объяснений. Бюджет передаётся в AgentState и проверяется на каждой итерации главного цикла.
Шаг 5. Пиши трассировку на каждом шаге
Каждый шаг агента должен оставлять запись в трейсе — с мыслью, действием, наблюдением и задержкой. Это единственный способ отладить поведение постфактум.
@dataclass
class Step:
index: int
thought: str
action: dict
observation: str
latency_ms: intТрейс хранится в AgentState.trace как список Step. После завершения его можно сериализовать в JSON и отдать в систему мониторинга или просто распечатать для разбора.
Где ломается
- Planner строит невалидный DAG. Если зависимости содержат цикл или несуществующий ID, исполнитель зависнет. Добавь валидацию топологической сортировки перед запуском.
- Critic слишком строгий или слишком мягкий. Промпт Critic-а нужно калибровать отдельно — иначе он будет блокировать валидные ответы или пропускать мусор.
- Параллельные инструменты с общим состоянием. Если два Worker-а пишут в один словарь без блокировки, данные испортятся. Используй
asyncio.Lockили проектируй инструменты как чистые функции. - Бюджет по времени vs. бюджет по шагам. Медленный инструмент может исчерпать
max_latency_msраньше, чем агент успеет что-то сделать. Разделяй бюджет инструментов и бюджет LLM-вызовов. - MockProvider скрывает реальные проблемы. Детерминированный мок хорош для тестов, но не показывает, как модель реагирует на неожиданные форматы ответа. Периодически прогоняй на реальном провайдере.
Что попробовать дальше
Следующий шаг после этой архитектуры — многоуровневая память: краткосрочная (контекст текущего запуска), среднесрочная (суммаризация прошлых шагов) и долгосрочная (векторное хранилище для поиска по истории). Это позволяет агенту не начинать с нуля при каждом запуске и переиспользовать результаты предыдущих задач.
Также стоит рассмотреть инструменты с типизированными схемами — когда инструмент описывает свои аргументы и возвращаемый тип через Pydantic-модель. Тогда валидация происходит до вызова, а не после, и ошибки становятся предсказуемыми.
FAQ
Чем DAG отличается от обычного списка шагов?
DAG (directed acyclic graph) явно описывает зависимости между задачами: задача B запускается только после задачи A, а задачи C и D могут идти параллельно. Обычный список всегда последовательный — каждый шаг ждёт предыдущего, даже если между ними нет реальной зависимости.
Зачем нужен Critic, если Planner уже знает цель?
Planner знает, что нужно сделать, но не видит финального результата. Critic смотрит на то, что получилось, и сверяет с исходной целью. Это разные задачи — планирование до выполнения и верификация после.
Можно ли использовать этот harness с OpenAI вместо Anthropic?
Да. Для этого и нужен абстрактный LLMProvider: реализуй подкласс с вызовом OpenAI API в методе complete, и остальной код менять не придётся.
Как тестировать агента без реальных API-вызовов?
Реализуй MockProvider, который возвращает заранее заданные строки в зависимости от параметра role. Для инструментов используй тестовый словарь CITY_FACTS вместо сетевых запросов — тогда весь ноутбук воспроизводим без интернета.
Что делать, если агент исчерпал бюджет до завершения задачи?
Зафиксируй текущий AgentState с status = "budget_exceeded", верни частичный результат и трейс. Это лучше, чем молча упасть: оркестратор или пользователь видит, сколько было сделано и где остановились.
Источники
- [Habr ML: [Перевод] Строим продвинутый каркас для ИИ-агента](https://habr.com/ru/companies/ostrovok/articles/1084568/?utm_campaign=1084568&utm_source=habrahabr&utm_medium=rss)
