Головна / Статті / Потокове оброблення LLM із можливістю відновлення за допомогою Redis Streams як місця зустрічі за кожен крок

redis streaming llm agents gateway realtime

Потокове оброблення LLM із можливістю відновлення за допомогою Redis Streams як місця зустрічі за кожен крок

Переживайте перерви у з’єднанні та кількаминутну зупинку інструментів, публікуючи події агента у Redis Stream з обмеженим діапазоном, щоб клієнти могли продовжити роботу.

2049 слів

Що ми хотіли

Відповіді агента можуть займати десятки секунд: обробка логіки, завантаження необхідних навичок, виклик інструментів, очікування, передача токенів. У демо-версіях підтримується одне активне з’єднання. У продакшені передача даних припиняється посеред відповіді, потрібно перезапускати шлюзи, а також очікувати на інструменти, які повертаються через кілька хвилин. Мета — можливість продовження реального часу потокової передачі, щоб клієнт міг знову під’єднатися та продовжити ту саму розмову.

Шлях до рішення

Ітерація 1: клієнт спілкується безпосередньо з агентом

Просто, але крихке рішення. Будь-яка проблема в мережі призводить до переривання потоку. Горизонтальне масштабування вимагає використання «липких сеансів» або призводить до втрати даних.

Ітерація 2: потокова передача даних через gRPC між сервісами

Кращі внутрішні контракти, проте все ще незручно для браузерних клієнтів, а також проблеми з тривалими паузами під час перезапуску компонентів.

Чому не Kafka?

Чудово підходить для міцних записів; важчий, ніж потрібно, для зустрічей за кожен хід із коротким зберіганням даних та групами споживачів, які погано підходять під концепцію „однієї вкладки браузера“.

Ітерація 3 (що спрацювало): названа зустріч за кожен хід

// Agent output
{
  "type": "tool_result",
  "tool": "product_search",
  "data": {
    "items": [...]
  }
}

// Gateway -> TV
{
  "type": "product_carousel",
  "items": [...]
}

// Gateway -> Mobile
{
  "type": "product_list",
  "items": [...]
}

cursor = last_event_id or "0-0"

while True:
    entries = xread({key: cursor}, block=30_000)

    if not entries:          # the only timeout check point
        check_timeouts()
        continue

    for entry_id, event in entries:
        # writes to the socket; not an ack that the client received it
        sse.send(id=entry_id, data=event.payload)
        cursor = entry_id
        if event.type in TERMINAL:
            return

id: 1755600000123-0
data: {"type":"tool_selected","tool":"search"}

id: 1755600000871-0
data: {"type":"response_block","block":{...}}

GET /sessions/{sid}/turns/{tid}/stream
Last-Event-ID: 1755600000871-0

# turn starts: one atomic step (MULTI/EXEC, or a Lua script)
xadd(key, first_event)
expire(key, GENEROUS_TTL)

# producer finishes: bring it in
expire(key, RECONNECT_TTL)

Кожен хід користувача отримує Redis Stream (або схему stream+група споживачів), ключ якого — turn_id. Агент публікує події токена/інструменту; шлюз продовжує роботу від останнього id клієнта. Повторне підключення відбувається з цього ж місця. Поди можуть зникнути; стрім зберігає достатню кількість історії для цього ходу.

Важливі деталі

Дисципліна курсора — клієнти підтверджують останній ідентифікатор стріму; після першого підключення ніколи не починають роботу з 0.

Події серцебиття — запобігають тому, щоб проміжні компоненти закривали неактивні з’єднання під час очікування інструментом.

Цикл життя — чіткі значення turn_started / turn_paused / turn_completed / turn_failed.

TTL — автоматичне закінчення роботи потоків після завершення операції, щоб Redis не перетворювався на нескінченний архів.

Аутентифікація — turn_id не є засобом авторизації; операції прив’язуються до автентифікованої сесії.

Ситуація, яка все вирішила: кількаминутна пауза

Інструмент передав завдання іншій системі обробки, яка відповіла лише через кілька хвилин. Пряме HTTP-стрімінг припинився. За допомогою Redis Streams агент опублікував подію паузи, клієнт продовжував виконувати завдання, а пізніше токени відновили роботу після повторного підключення без необхідності перезапуску всього плану.

Що ми надсилаємо клієнту

Типові події: токени, tool_start, резюме tool_result (ніколи секретна інформація), помилки та завершення. Зберігайте об’єми даних невеликими; складні дані зберігайте у об’єктному сховищі та надсилайте лише посилання на них.

Витрати, обмеження, моменти уваги

Стежте за об’ємом пам’яті Redis, максимальною довжиною потоку та кількістю вихідних ланцюгів, якщо багато шлюзів працюють одночасно. Встановіть ліміт на кількість одночасних операцій на користувача. Проведіть тест навантаження на відновлення з’єднань після розгортання шлюзу.

Кінцевий стан

Шлюз є читачем, який бере до уваги стан виконання завдань; агент — записувачем даних; Redis Streams слугує місцем зустрічі. Реальний час у користувацькому інтерфейсі дозволяє пережити нудні збої, які руйнують архітектури демонстраційних систем.

Порада щодо експлуатації: зберігайте останній ID потоку у кукі сесії або в пам’яті клієнта, а також на сервері для можливості відтворення. Коли користувач каже, що „все зависло“, служба підтримки повинна відновити хід розмови з потоку, не просячи його чекати десять хвилин на запуск інструменту. Додайте панелі керування для відстеження частки відновлених розмов, кількості незавершених діалогів та середньої тривалості пауз, щоб компанія могла зрозуміти, чи причина у повільній роботі операторів, чи у нестабільності мережі.

Порада щодо експлуатації: зберігайте останній ID потоку у кукі сесії або в пам’яті клієнта, а також на сервері для можливості відтворення. Коли користувач каже, що „все замерло“, служба підтримки повинна відновити хід розмови з потоку, не просячи його чекати десять хвилин на запуск інструменту. Додайте панелі керування для відстеження частки відновлених розмов, кількості незавершених діалогів та середньої тривалості пауз, щоб компанія могла зрозуміти, чи причина у повільній роботі операторів, чи у нестабільності мережі.

Порада щодо експлуатації: зберігайте останній ID потоку у кукі сесії або в пам’яті клієнта, а також на сервері для можливості відтворення. Коли користувач каже, що „все зависло“, служба підтримки повинна відновити хід розмови з потоку, не просячи його чекати десять хвилин на запуск інструменту. Додайте панелі керування для відстеження частки відновлених розмов, кількості незавершених діалогів та середньої тривалості перерв, щоб компанія могла зрозуміти, чи причина у повільній роботі операторів, чи у нестабільності мережі.

Порада щодо експлуатації: зберігайте останній ID потоку у кукі сесії або в пам’яті клієнта, а також на сервері для можливості відтворення. Коли користувач каже, що „все зависло“, служба підтримки повинна відновити хід розмови з потоку, не просячи його чекати десять хвилин на запуск інструменту. Додайте панелі керування для відстеження частоти продовження розмов, кількості незавершених діалогів та середньої тривалості пауз, щоб компанія могла зрозуміти, чи причина у повільній роботі агентів чи нестабільності мережі.

Усі статті Надіслати бриф