# Один пилот не ведёт воздушную операцию: как вырастить базовый агентный цикл в боевую систему
Каноническая страница: https://plainews.ru/posts/advanced-agentic-harness-python-guide
Опубликовано: 2026-09-25T07:01:42.121Z

Как добавить к базовому агентному циклу DAG, параллельное выполнение, многоуровневую память и роли Planner/Worker/Critic — пошаговый гайд на Python.
Простой агентный цикл — LLM получает задачу, вызывает инструмент, смотрит на результат, повторяет — работает. Но только пока задача маленькая и всё идёт по плану. Как только появляются параллельные зависимости, ошибки инструментов или необходимость доказать, что агент сделал именно то, что нужно, цикл ломается. Этот гайд показывает, как последовательно добавить к нему планирование, параллельное выполнение, верификацию и трассировку — на реальном Python-коде.

## Почему простого цикла не хватает

Базовый harness выглядит примерно так: модель думает, выбирает инструмент, получает наблюдение, снова думает. Это корректно для линейных задач. Но возьмём задачу сравнения трёх городов по трём атрибутам — население, часовой пояс, описание. Это девять вызовов инструментов, каждый из которых независим от остальных, но итоговый отчёт зависит от всех девяти. Последовательный цикл выполнит их один за другим. Параллельный — за время самого медленного.

Кроме линейности, у наивного цикла есть ещё три проблемы: нет разделения ответственности (одна модель планирует и исполняет), нет верификации результата (агент сам решает, что справился), нет бюджета (цикл может крутиться бесконечно). Дальше — как это починить по шагам.

## Шаг 1. Абстрагируй LLM-провайдера

Прежде чем строить что-либо поверх, нужна точка подключения мозга, которую можно заменить без переписывания всего harness. Определи базовый класс:

```python
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 можно прогнать без сети и без реальных токенов.

> **К сведению.** Параметр `role` позволяет передавать разные системные промпты для Planner, Worker и Critic — одному провайдеру, но с разным контекстом.

## Шаг 2. Добавь DAG и параллельное выполнение

Граф зависимостей (DAG — directed acyclic graph) — это список задач, где каждая задача знает, от каких других она зависит. Планировщик строит граф, исполнитель запускает задачи, у которых все зависимости уже выполнены.

```python
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:

```python
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 или список нарушений.

```python
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. Введи многомерный бюджет

Бюджет — это не только лимит шагов. Реальная система контролирует несколько измерений одновременно:

```python
@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. Пиши трассировку на каждом шаге

Каждый шаг агента должен оставлять запись в трейсе — с мыслью, действием, наблюдением и задержкой. Это единственный способ отладить поведение постфактум.

```python
@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)