Как приручить недетерминизм LLM-агентов с помощью Event Sourcing: практическое руководство

Почему ваш агент — гадалка, а не инженер

Вы запускаете одного и того же LLM-агента с одним промптом. Первый раз он возвращает идеальный код. Второй — падает с ошибкой. Третий — пишет поэму вместо SQL. Знакомо?

Недетерминизм встроен в природу больших языковых моделей. Температура, случайное сэмплирование, даже порядок токенов в контексте — каждый вызов /v1/chat/completions уникален. Это не баг, это фича — но только если вы пишете стихи. В production-агентах такая непредсказуемость превращает отладку в шаманство с бубном.

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

Решение, которое я предлагаю, не ново — это Event Sourcing. Но в контексте LLM-агентов он даёт суперспособность: вы фиксируете каждое «решение» агента как событие и можете переиграть его с теми же входными данными, меняя только код или промпт. Без гаданий.

Event Sourcing: не просто логирование

Многие разработчики путают Event Sourcing с обычным логом. Лог — это запись того, что произошло. Event Sourcing — это источник истины. Вы не записываете текущее состояние агента, вы сохраняете последовательность событий, которые к этому состоянию привели. И можете в любой момент воспроизвести всё заново.

Для LLM-агента событием может быть:

• Запрос к LLM (промпт + параметры)
• Ответ LLM (сгенерированный текст, логиты, температура)
• Вызов инструмента (имя инструмента + аргументы)
• Результат инструмента (успех/ошибка, данные)
• Решение о ветвлении (какой путь выбран)

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

Архитектура: CQRS + Event Store

На практике мы разделяем запись событий (команды) и чтение (запросы). Это паттерн CQRS. События пишутся в специальное хранилище — Event Store. А для агрегации состояния (например, текущий контекст агента) используем проекции.

В 2026 году зрелые решения для Event Store уже стандарт: EventStoreDB (текущая версия 24.10 LTS), Apache Kafka (версия 3.9), облачные сервисы вроде Azure Event Hubs. Для небольших агентов можно обойтись PostgreSQL в режиме логирования или SQLite с интерфейсом eventstore.

1 Проектируем схему событий

Начнём с базовой модели на Python. Будем использовать Pydantic для валидации и сериализации.

from pydantic import BaseModel, Field
from datetime import datetime, timezone
from enum import Enum
import uuid

class EventType(str, Enum):
    LLM_REQUEST = "llm_request"
    LLM_RESPONSE = "llm_response"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    HUMAN_INPUT = "human_input"
    AGENT_DECISION = "agent_decision"

class AgentEvent(BaseModel):
    session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    correlation_id: str  # связь между запросом и ответом
    event_type: EventType
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    payload: dict  # JSON с деталями
    metadata: dict = {}  # версия агента, температура, seed и т.д.

2 Строим Event Store на PostgreSQL

Для простоты реализуем слой хранения. В production используйте готовые решения, но для понимания — вот минимальный пример.

import psycopg2
from psycopg2.extras import Json, execute_values

class PostgresEventStore:
    def __init__(self, dsn):
        self.conn = psycopg2.connect(dsn)
        self._create_tables()

    def _create_tables(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS agent_events (
                    id SERIAL PRIMARY KEY,
                    session_id UUID NOT NULL,
                    correlation_id UUID NOT NULL,
                    event_type VARCHAR(50) NOT NULL,
                    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    payload JSONB NOT NULL,
                    metadata JSONB DEFAULT '{}'
                );
                CREATE INDEX IF NOT EXISTS idx_events_session
                    ON agent_events (session_id, timestamp);
                CREATE INDEX IF NOT EXISTS idx_events_correlation
                    ON agent_events (correlation_id);
            """)
        self.conn.commit()

    def append_event(self, event: AgentEvent):
        with self.conn.cursor() as cur:
            cur.execute(
                """INSERT INTO agent_events 
                   (session_id, correlation_id, event_type, payload, metadata) 
                   VALUES (%s, %s, %s, %s, %s)""",
                (event.session_id, event.correlation_id, event.event_type.value,
                 Json(event.payload), Json(event.metadata))
            )
        self.conn.commit()

    def get_events_for_session(self, session_id: str) -> list[AgentEvent]:
        with self.conn.cursor() as cur:
            cur.execute(
                """SELECT session_id, correlation_id, event_type, 
                          timestamp, payload, metadata 
                   FROM agent_events 
                   WHERE session_id = %s 
                   ORDER BY timestamp""", (session_id,)
            )
            rows = cur.fetchall()
        return [AgentEvent(
            session_id=row[0], correlation_id=row[1],
            event_type=EventType(row[2]), timestamp=row[3],
            payload=row[4], metadata=row[5]
        ) for row in rows]

3 Оборачиваем LLM-вызовы в события

Теперь модифицируем вашего агента так, чтобы каждый вызов LLM сначала сохранял событие запроса, а после получения ответа — событие ответа.

import openai  # версия 1.56+ на 2026

class EventedLLMAgent:
    def __init__(self, event_store: PostgresEventStore, client: openai.OpenAI):
        self.store = event_store
        self.client = client
        self.session_id = str(uuid.uuid4())

    async def chat(self, messages: list[dict], **kwargs) -> str:
        # Генерируем correlation_id для этой пары запрос-ответ
        correlation_id = str(uuid.uuid4())
        
        # Записываем событие запроса
        request_event = AgentEvent(
            session_id=self.session_id,
            correlation_id=correlation_id,
            event_type=EventType.LLM_REQUEST,
            payload={"messages": messages, **kwargs},
            metadata={"model": kwargs.get("model", "gpt-4o"), "temperature": kwargs.get("temperature", 1.0)}
        )
        self.store.append_event(request_event)

        # Выполняем запрос (недетерминированный)
        response = await self.client.chat.completions.create(
            model=kwargs.get("model", "gpt-4o"),
            messages=messages,
            **kwargs
        )
        
        # Записываем событие ответа
        response_event = AgentEvent(
            session_id=self.session_id,
            correlation_id=correlation_id,
            event_type=EventType.LLM_RESPONSE,
            payload={"choices": [c.model_dump() for c in response.choices]},
            metadata={"finish_reason": response.choices[0].finish_reason}
        )
        self.store.append_event(response_event)
        
        return response.choices[0].message.content

Воспроизведение сцены: повторяем с тем же seed

Главная фишка Event Sourcing — возможность воспроизвести сессию агента в точности, как она была, если мы зафиксируем всё, от seed до температуры. Для LLM важно не только сохранить промпт, но и параметры генерации (seed, temperature, top_p).

В OpenAI API начиная с моделей GPT-4o (2025) и новее параметр seed работает детерминированно, если задать одинаковые seed и temperature=0. На 31 мая 2026 года последняя версия GPT-4o с суффиксом -2026-05-15 поддерживает deterministic mode через seed + temperature=0 и response_format.

# Воспроизведение: извлекаем события, повторяем с теми же параметрами
async def replay_agent_session(event_store: PostgresEventStore, session_id: str):
    events = event_store.get_events_for_session(session_id)
    request_events = [e for e in events if e.event_type == EventType.LLM_REQUEST]
    
    for req in request_events:
        payload = req.payload
        messages = payload.pop("messages")
        metadata = req.metadata
        seed = metadata.get("seed", 42)  # сохранённый seed
        temperature = metadata.get("temperature", 0.0)
        
        # Повторяем запрос
        response = await openai.chat.completions.create(
            model=metadata.get("model", "gpt-4o"),
            messages=messages,
            seed=seed,
            temperature=temperature,
            **payload
        )
        # Сравниваем с оригинальным ответом
        original = [e for e in events if e.correlation_id == req.correlation_id and e.event_type == EventType.LLM_RESPONSE][0]
        is_identical = response.choices[0].message.content == original.payload["choices"][0]["message"]["content"]
        print(f"Correlation {req.correlation_id}: {'✓ Match' if is_identical else '✗ Diff'}")

Продвинутые техники: сага для агентов

Когда агент взаимодействует с внешними сервисами (API, базы данных), недетерминизм умножается на сетевые задержки и состояния. Event Sourcing в комбинации с паттерном Saga позволяет откатывать или компенсировать действия, если один из шагов провалился.

Каждое внешнее действие — тоже событие. Если на этапе replay мы не хотим реально вызывать API, мы можем заменить вызов на событие из прошлого (mock). Так мы проверяем, как новый код агента реагирует на те же внешние ответы.

4 Mock внешних вызовов через Event Store

Предположим, наша сессия содержит события TOOL_CALL и TOOL_RESULT. При повторном прогоне мы можем проверить, действительно ли новый код агента сгенерирует те же вызовы инструментов. Если нет — значит изменение кода влияет на логику, и это нужно осознанно принять.

class ReplayContext:
    def __init__(self, events: list[AgentEvent]):
        self.events_by_type = defaultdict(list)
        for e in events:
            self.events_by_type[e.event_type].append(e)
        self._tool_result_index = 0

    async def call_tool(self, name: str, args: dict) -> dict:
        # Ищем сохранённый результат для этого вызова
        # (упрощённо: берём по порядку)
        if self._tool_result_index < len(self.events_by_type[EventType.TOOL_RESULT]):
            result_event = self.events_by_type[EventType.TOOL_RESULT][self._tool_result_index]
            self._tool_result_index += 1
            return result_event.payload  # возвращаем оригинальный результат
        else:
            raise RuntimeError("No recorded result for tool call")

# Использование в replay:
# replay_ctx = ReplayContext(events)
# response = await agent.run(replay_ctx=replay_ctx)

Ошибки, которые я совершал (и вы, вероятно, тоже)

Поделюсь шишками, которые набил за последние два года внедрения Event Sourcing в агентные системы.

• Игнорирование версий модели. Если вы обновили модель с gpt-4o на gpt-4o-mini, replay сломается. Всегда сохраняйте model и version (например, openai/gpt-4o-2026-04-15).
• Не сохраняли seed и температуру. Без seed детерминизма не будет. Фиксируйте все параметры генерации.
• Слишком большой объём событий. Каждый вызов LLM может генерировать мегабайты токенов. Используйте сжатие или храните payload отдельно (например, в S3).
• Забыли про временные метки в событиях внешних систем. Если агент ждал 10 секунд, а при replay ждёт 0 — это может изменить поведение (таймауты).
• Не тестировали с другими провайдерами. Один и тот же seed у OpenAI и Anthropic даёт разные результаты. Event Sourcing должен хранить провайдера как часть metadata.

Как Event Sourcing помогает с эпистемической асимметрией

В статье про «молчаливого учёного» обсуждается, как агенты не знают о своих предыдущих действиях и теряют контекст. Event Sourcing решает это радикально: при каждом новом запросе мы передаём агенту не просто краткое резюме, а все релевантные события в структурированном виде. Агент видит последовательность своих решений, а не только финальное состояние.

Это напрямую борется с «забывчивостью» долгоживущих агентов. Вместо того чтобы сжимать историю в один контекст, мы даём модель саму решить, какие события важны.

Интеграция с Agent Skills и RLM

Event Sourcing отлично сочетается с подходом Agent Skills. Каждый навык — это не только промпт, но и набор событий, которые он порождает. Мы можем переиспользовать события между разными агентами, строить дашборды и аналитику.

А метод RLM снижает контекстный дрейф, а Event Sourcing даёт сырьё для его проекций. Мы можем построить проекцию «текущая цель» из событий и подавать её агенту как контекст.

Инструменты, которые стоит присмотреть в 2026

На рынке уже есть зрелые решения для durable execution с Event Sourcing, которые напрямую поддерживают LLM-агентов:

• Temporal (v1.25+) — отличный выбор для надёжных оркестраций. В 2026 добавили встроенный детектор недетерминированных активностей.
• Inngest — лёгкий, бессерверный, с поддержкой step functions.
• Durable Functions (Azure) — классика, но требует проприетарного стека.
• LangGraph (от LangChain) — встроенный Checkpointer, который по сути и есть Event Sourcing. На 2026 версия 0.3.x стабильна.

Однако все они «зашивают» свою логику хранения событий. Если вам нужна полная прозрачность и кастомные проекции — строить свой Event Store поверх PostgreSQL не сложнее, чем кажется.

Ответы на холиварные вопросы (FAQ style)

В качестве напутствия

Недетерминизм — не враг, а материал. Но работать с ним нужно как с радиоактивным веществом: в защитном костюме Event Sourcing. Не пытайтесь сделать агента полностью детерминированным — это невозможно. Вместо этого сделайте каждый его шаг воспроизводимым и прозрачным. Тогда вы сможете уверенно менять промпты, обновлять модели и не бояться, что production рассыплется.

Попробуйте внедрить хотя бы минимальную версию Event Store в вашего текущего агента. Поверьте, когда в следующий раз баг будет воспроизводиться с трёх попыток из пяти — вы скажете спасибо.

Полный код проекта (с интеграцией Temporal и LangGraph) я выложил в свой репозиторий. Ссылка — в профиле.

Подписаться на канал