Практическая статья по устройству production-ready агента
Введение
Поскольку последнее время я плотно занимаюсь разработкой ии-агента, и, по прогнозам директора, должен скоро все сдать (лол), то я решил описать в первую очередь для себя кое-какие моменты, которые стоит учесть при разработке агентской системы в 2026 году. Я планирую серию статей на основании своего опыта. Не судите строго, на платных курсах расскажут гораздо лучше. Накидать в комменты приветствуется. Перевод терминологии вольный.
Сейчас мне кажется, что весь софт, который последнее время делается - это один сплошной ии-агент, который потенциально должен уметь всё на свете. При этом пользователи в 2026 году не готовы ни к какой другой форме отношений с приложениями, кроме как промптинг. Если во время презентации продукта они видят больше одной кнопки “отправить промпт”, то сразу заявляют, что им сложно, а у тебя появляется чувство, словно ты им должен заплатить за то, чтобы они осилили твой софт. Ну ладно, мобильные телефоны в итоге ведь превратились в прямоугольники с экранами. Может, и у софта есть “финальная форма” в виде ии-агента с интерфейсом.
Базовый минимум
Когда говорят об агенте, очень часто имеют в виду “LLM с большим системным промптом и несколькими функциями, которая не останавливается пока не решит, что все сделала”, при этом вся предыдущая история диалога передается по апи в нейросеть при каждом промте.
Возможно, кто-то так и делает. И такой подход в целом позволяет работать, но быстро и часто ломается: состояние теряется после рестарта, опасные действия неотличимы от безопасных, долгие операции живут внутри HTTP-запроса, а весь контроль над поведением фактически перекладывается на вероятностную модель. Либо токенов на каждый запрос расходуется столько, что матушка-природа погибнет раньше, чем до нас доберется ии.
Зрелый агент должен быть устроен иначе. В агенте должны сочетаться durable state (долговременное состояние), явное планирование, типизированные инструменты, approval-политика, журнал событий и отдельный рантайм для фоновых задач.В правильном агенте модель является только одним из компонентов принятия решения.
Долговременное состояние (durable state)
Durable state - это не один объект некоего класса DurableState. Это слой данных, в котором агент хранит факты о своей работе: активные ходы, планы, статусы шагов, ожидающие подтверждения, события, результаты и ошибки. В коде этот слой обычно выражается набором ORM-моделей и сервисов вокруг них.
Durable state позволяет сохранять состояние (историю, текущий план, выполненные шаги, паузы, ожидания) во внешнем хранилище. В таком случае агент сможет пережить: перезапуск рантайма, смену версии модели, ожидание ответа человека дни и недели, высадку Илона Маска на марсе и многое другое. Пользователь сможет остановить агента, а через час сказать «продолжи», и агент поднимет состояние и пойдёт дальше как ни в чём не бывало.
Без долговременного состояния взаимодействие с агентом будет представлять собой список сообщений в памяти текущей сессии (in-memory). В таком случае падение сервиса приведет к потери данных. Это не считая того, что память нужно высвобождать.
Представим агента, который выполняет долгую задачу: анализирует набор документов, строит план, ждет подтверждения пользователя, затем запускает обработку.
Плохой вариант:
memory = {}
def handle_message(session_id: str, message: str):
if message == "проанализируй документы":
memory[session_id] = {
"status": "running",
"task": "analyze_documents",
"step": "started",
}
result = analyze_documents()
memory[session_id]["status"] = "completed"
memory[session_id]["result"] = result
return "Готово"
Если процесс упадет, вся память исчезнет. После рестарта агент не знает, была ли задача запущена, завершилась ли она, что пользователь подтвердил и какие шаги уже выполнены.
Но в отличие от Гигачата, вариант, который вы выкатите в продакшен, будет содержать реализацию набора классов для создания durable state. Минимально следующие классы: AgentTurn (ход), AgentPlanItem (шаг) и AgentEvent (событие), ApprovalGrant (выданные подтверждения), SessionContext (состояние сессии), BackgroundJob (фоновая задача).
Ход агента (Agent turn)
Ход, или turn, представляет собой один полный цикл работы агента на один запрос пользователя. Сперва введем класс AgentTurn. Это и будет durable-запись одного хода агентского протокола. Она хранит не только текст пользователя, но и состояние обработки: во что команда была нормализована, нужно ли подтверждение, чем выполнение завершилось и была ли ошибка. Благодаря этому агент не обязан “помнить” ход в промпте или оперативной памяти процесса. Он может восстановить состояние из базы.
class AgentTurn(Base):
#Имя таблицы, куда ORM будет сохранять ходы агента. Один объект AgentTurn в Python соответствует одной строке в таблице agent_turns.
tablename = “agent_turns” #редактор хабра не дает написать через __
# Уникальный идентификатор хода.
# По нему дальше связываются план, события, approvals и результаты выполнения.
turn_id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
# Идентификатор пользовательской сессии.
# Нужен, чтобы найти все ходы конкретного пользователя или диалога.
session_id: Mapped[str] = mapped_column(String(200), index=True)
# Сырой текст пользователя.
# Например: "проанализируй документы и сделай отчет".
input_text: Mapped[str] = mapped_column(Text)
# Нормализованная команда, полученная из input_text.
# Это уже не свободный текст, а структурированное намерение:
# {"action": "analyze_documents", "scope": "current_workspace"}.
normalized_command: Mapped[dict | None] = mapped_column(JSON, nullable=True)
# Текущий статус хода.
# Например: created, planned, awaiting_approval, running, completed, failed.
status: Mapped[str] = mapped_column(String(40), default="created")
# Требуется ли подтверждение пользователя перед выполнением.
# Например, если действие изменяет данные или запускает дорогую операцию.
needs_confirmation: Mapped[bool] = mapped_column(Boolean, default=False)
# Финальный текстовый ответ агента.
# Например: "Готово, отчет сформирован".
output_text: Mapped[str | None] = mapped_column(Text, nullable=True)
# Ошибка, если ход завершился неуспешно.
# Хранится durable, чтобы после сбоя можно было понять причину.
error: Mapped[str | None] = mapped_column(Text, nullable=True)
Допустим, пользователь пишет: “Проверь документы и сделай отчет”. Backend создает turn_id. Дальше все, что агент делает по этому запросу, привязано к этому turn_id:
turn_id = 123
запрос пользователя → план → выполнение инструментов → события прогресса → финальный ответ
Агент не ограничивается одним ходом. Если, например, ход звучит как: "Analyze documents and make a report", - то внутри него будут такие шаги:
collect_documents → analyze_documents → generate_report
У хода обычно есть состояние: running, awaiting_approval, completed, failed, cancelled
Шаг хода/плана агента (Agent Plan Item)
Шаг хода, или plan item, это отдельное действие внутри хода. Один ход может состоять из нескольких шагов.
Ход: Пользователь: “Проанализируй проект и сделай отчет”
Шаги плана: Найти документы проекта → Прочитать релевантные файлы → Проверить → Сформировать отчет
Каждый шаг плана может быть связан с конкретным инструментом: tool_name = excel_reader, tool_name = csv_reader. Шаг плана отвечает на вопрос: “Какие конкретные действия агент решил выполнить, чтобы закрыть ход?” У шага тоже есть статус: pending, running, completed, failed, skipped, awaiting_approval. Для описания шага этого нужна отдельная таблица и сущность.
class AgentPlanItem(Base):
# Таблица, где хранятся отдельные шаги планов агента.
tablename = "agent_plan_items"
# Уникальный идентификатор шага плана.
item_id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
# Ссылка на ход агента, которому принадлежит этот шаг.
# ForeignKey("agent_turns.turn_id") означает:
# это поле связано с turn_id из таблицы agent_turns.
turn_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("agent_turns.turn_id"))
# Порядковый номер шага внутри плана.
# Например: 0 - собрать данные, 1 - проанализировать, 2 - сформировать отчет.
step_index: Mapped[int] = mapped_column(default=0)
# Имя инструмента, который нужно вызвать на этом шаге.
# Например: "collect_documents", "analyze_documents", "generate_report".
tool_name: Mapped[str] = mapped_column(String(120))
# Аргументы для инструмента.
# Например: {"scope": "current_workspace", "format": "markdown"}.
args: Mapped[dict] = mapped_column(JSON, default=dict)
# Режим подтверждения.
# safe_readonly - безопасный read-only шаг;
# confirm_once - нужно подтверждение один раз;
# mutating - действие меняет состояние и требует строгой проверки.
approval_mode: Mapped[str] = mapped_column(String(40), default="safe_readonly")
# Статус конкретного шага.
# Например: created, running, completed, failed.
status: Mapped[str] = mapped_column(String(40), default="created")
# Результат выполнения шага, если он завершился успешно.
# Например: {"documents_found": 12}.
result: Mapped[dict | None] = mapped_column(JSON, nullable=True)
# Ошибка выполнения шага, если он упал.
error: Mapped[str | None] = mapped_column(Text, nullable=True)
А еще нам нужен журнал событий агента (AgentEvent).
Журнал событий Агента (Agent event)
Событие, или event, это запись о том, что что-то произошло во время хода.
Например: turn_started, plan_started, plan_ready, tool_started, tool_progress, tool_completed, approval_requested, verification_started, turn_completed,
Пример события:
{
"type": "tool_progress",
"turn_id": "123",
"tool_name": "summarize_project",
"status": "running",
"message": "Анализирую документы: 40%",
"payload": {
"percent": 40,
"done": 4,
"total": 10
}
}
Если AgentTurn отвечает на вопрос: ”Что за пользовательский ход сейчас обрабатывается?” То AgentEvent отвечает: “Что происходило во времени?” Список событий может быть таким: turn_created, plan_created, approval_requested, tool_started, tool_completed, tool_failed, turn_completed.
class AgentEvent(Base):
# Таблица, где хранится timeline агента:
# что произошло, когда произошло и к чему это относится.
tablename = "agent_events"
# Уникальный идентификатор события.
event_id: Mapped[uuid.UUID] = mapped_column(
primary_key=True,
default=uuid.uuid4,
)
# К какому ходу агента относится событие.
# Например, событие "tool_started" относится к конкретному AgentTurn.
turn_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("agent_turns.turn_id"),
)
# К какому шагу плана относится событие.
# Может быть пустым, если событие относится ко всему ходу,
# а не к конкретному шагу.
# Например:
# turn_started -> item_id = None
# approval_requested -> item_id = None
# tool_started -> item_id = id конкретного AgentPlanItem
# tool_completed -> item_id = id конкретного AgentPlanItem
item_id: Mapped[uuid.UUID | None] = mapped_column(nullable=True)
# Уникальный идентификатор сессии
session_id: Mapped[str] = mapped_column(String(200), index=True)
# Тип события.
# Например: "turn_started", "tool_started", "tool_completed", "tool_failed".
event_type: Mapped[str] = mapped_column(String(80), index=True)
# Статус, связанный с событием.
# Например: "running", "completed", "failed", "awaiting_approval".
status: Mapped[str | None] = mapped_column(String(40), nullable=True)
# Произвольные дополнительные данные события.
# Например прогресс, ошибка, имя инструмента, количество обработанных объектов.
payload: Mapped[dict] = mapped_column(JSON, default=dict)
# Время создания события.
# По нему можно построить timeline выполнения.
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
Теперь посмотрим, как оно работает вместе.
Пример работы: Turn→Plan Item → Event
Пример: пользователь написал: ”Проанализируй документы и сделай отчет”.
Система создала AgentTurn и три шага плана:
1. collect_documents
2. analyze_documents
3. generate_report
Во время выполнения будут появляться события:
AgentEvent(
turn_id=turn.turn_id,
item_id=None,
event_type="turn_started",
status="running",
payload={},
)
Потом старт первого инструмента:
AgentEvent(
turn_id=turn.turn_id,
item_id=collect_documents_item.item_id,
event_type="tool_started",
status="running",
payload={
"tool_name": "collect_documents",
"args": {"scope": "current_workspace"},
},
)
Потом завершение первого инструмента:
AgentEvent(
turn_id=turn.turn_id,
item_id=collect_documents_item.item_id,
event_type="tool_completed",
status="completed",
payload={
"tool_name": "collect_documents",
"documents_found": 12,
},
)
Потом прогресс второго шага:
AgentEvent(
turn_id=turn.turn_id,
item_id=analyze_documents_item.item_id,
event_type="tool_progress",
status="running",
payload={
"tool_name": "analyze_documents",
"done": 40,
"total": 100,
"percent": 40,
},
)
Если что-то упало:
AgentEvent(
turn_id=turn.turn_id,
item_id=analyze_documents_item.item_id,
event_type="tool_failed",
status="failed",
payload={
"tool_name": "analyze_documents",
"error": "external service timeout",
"retryable": True,
},
)
И в конце:
AgentEvent(
turn_id=turn.turn_id,
item_id=None,
event_type="turn_completed",
status="completed",
payload={
"output": "Отчет сформирован",
},
)
В результате UI становится гораздо показательнее. Без событий UI может показать только:
“Агент думает…”
С событиями UI может показывать нормальный прогресс:
Собираю документы...
Нашел 12 документов.
Анализирую документы: 40%.
Формирую отчет...
Готово.
То есть frontend не должен гадать, что происходит. Он читает события.
Пример для фронтенда:
def get_turn_events(db: Session, turn_id: uuid.UUID):
return (
db.query(AgentEvent)
.filter(AgentEvent.turn_id == turn_id)
.order_by(AgentEvent.created_at.asc())
.all()
)
Помимо хорошего UX на фронтенде, такая структура способствует лучшей отладке, восстановлению после рестарта, аудиту, тестам.
Но AgentTurn, AgentPlanItem и AgentEvent - это еще не весь durable state. Они описывают конкретный ход: что пользователь попросил, какой план построен, какие шаги выполнялись и какие события произошли. В production-агенте обычно нужны еще другие сущности: ApprovalGrant (выданные подтверждения), SessionContext (состояние сессии), BackgroundJob #фоновые задачи и, возможно, ProjectContext (состояние проекта), но о них в следующий раз.
P.S. Описания классов не полные, можете менять их любым образом
Телеграм канал автора, где он что‑то пишет про ML, NLP и разработку