The Agent as an API: Wrapping It in a Service

Первый рефлекс — сделать POST /ask, внутри запустить цикл агента и вернуть ответ. Для демо сойдёт, в production развалится по трём причинам. Прогон агента длится от десятков секунд до минут — HTTP-клиенты, прокси и балансировщики обрывают такие запросы: у большинства reverse proxy таймаут 30-60 секунд по умолчанию. Обрыв соединения убивает прогон на середине — деньги потрачены, результата нет, повторить нельзя. И пока цикл крутится внутри обработчика, воркер занят — десяток параллельных пользователей кладут сервис.

Стандартное решение то же, что у любых долгих операций (video encoding, экспорт отчётов): асинхронные задания. POST /runs принимает задачу, валидирует, создаёт запись прогона со статусом queued и немедленно возвращает 202 с run_id. Сам цикл исполняется вне обработчика — фоновой задачей, отдельным воркером, очередью. Клиент опрашивает GET /runs/{id} (статус, шаги, результат) или подписывается на поток событий.

Жизненный цикл прогона стоит зафиксировать конечным автоматом: queued → running → succeeded | failed | cancelled | awaiting_approval. Последний статус — прямое продолжение урока 5: когда агент упирается в шлюз подтверждения, прогон не блокирует поток, а сохраняет состояние, переходит в awaiting_approval и ждёт; человек отвечает через POST /runs/{id}/approve или /reject, и прогон продолжается. Добавьте POST /runs/{id}/cancel — право пользователя передумать, — и следите, чтобы отмена корректно останавливала цикл между шагами, а не посреди пишущего инструмента.

Минимальный сервис — три эндпоинта и стор прогонов. Начнём с памяти процесса, но за интерфейсом, чтобы замена на Redis или Mongo не трогала обработчики.

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
import uuid

app = FastAPI()
RUNS: dict[str, dict] = {}          # позже: Redis/Mongo за тем же интерфейсом

class CreateRun(BaseModel):
    task: str
    max_usd: float = 0.50

@app.post("/runs", status_code=202)
def create_run(req: CreateRun, bg: BackgroundTasks):
    run_id = str(uuid.uuid4())
    RUNS[run_id] = {"status": "queued", "task": req.task,
                    "events": [], "result": None}
    bg.add_task(execute_run, run_id, req)   # цикл — вне обработчика
    return {"run_id": run_id}

@app.get("/runs/{run_id}")
def get_run(run_id: str):
    run = RUNS.get(run_id)
    if not run:
        raise HTTPException(404)
    return run

execute_run — это агент из прошлых уроков, только каждый шаг он дописывает в run["events"] и обновляет статус. BackgroundTasks годится для учебного сервиса; настоящий production захочет отдельные воркеры и очередь (хотя бы для того, чтобы деплой API не убивал живые прогоны), но контракт эндпоинтов при этом не изменится — в этом и смысл асинхронной формы.

Обязательная деталь — идемпотентность создания. Клиент, не дождавшийся ответа POST /runs, повторит запрос — и запустит второй платный прогон той же задачи. Принимайте заголовок Idempotency-Key: повторный POST с тем же ключом возвращает существующий run_id вместо создания нового. Вы уже строили это для инструментов в уроке 3 — теперь тот же паттерн защищает вас самих.

Валидация на входе — тоже часть контракта: Pydantic-модель CreateRun отбрасывает мусор до того, как он стоит денег, — ограничьте длину задачи, потолок max_usd, допустимый набор параметров. Всё, что прошло валидацию, фиксируйте в записи прогона как есть: воспроизводимость начинается с точного знания, что именно и с какими параметрами запускали.

Агент в production — это система, которая принимает решения без вас. Единственный способ ей доверять — видеть каждое решение. Хорошая новость: вы уже пишете JSONL-лог с урока 5; теперь он становится частью API.

Событийная модель прогона: каждый элемент run["events"] — это запись с типом (model_call, tool_call, tool_result, approval_requested, approval_resolved, budget_charge, finished), временем и полезной нагрузкой. GET /runs/{id} возвращает события — и фронтенд бесплатно получает живой лог «что агент делает прямо сейчас», а вы — материал для разбора инцидентов. Для интерактивных сценариев добавьте Server-Sent Events: GET /runs/{id}/events стримит новые события по мере появления; SSE проще WebSocket и переживает прокси.

Три правила, выстраданные практикой. Первое: сквозной trace_id (обычно run_id) во всех логах, включая исходящие HTTP-вызовы инструментов, — иначе не свяжете жалобу пользователя с конкретным прогоном. Второе: считайте деньги на уровне прогона и агрегатов — метрики usd_per_run, tokens_per_run, steps_per_run с перцентилями; аномалия в p95 стоимости — самый ранний сигнал, что агент начал ходить кругами. Третье: редакция секретов на записи, а не на чтении, — аргументы инструментов попадают в события, и токен, случайно оказавшийся в аргументах, навсегда останется в логе, если не вырезать его до записи.

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