Как вы проектируете dead letter queue для failed LLM инференс запросов?

Краткий тезис

Dead letter queue (DLQ) для failed LLM инференс запросов — это механизм изоляции и последующей обработки запросов, которые не удалось выполнить после всех повторных попыток. Типичная архитектура строится на Kafka (или аналогичной очереди) с тремя топиками: main (входящие запросы), retry (повторные попытки с задержкой) и DLQ (финальные ошибки). Из DLQ запросы либо репроцессируются вручную после анализа, либо автоматически после исправления причины, либо генерируют алерт в Sentry / PagerDuty. Ключевые аспекты: стратегия повторных попыток (backoff|exponential backoff, jitter), мониторинг глубины DLQ, идемпотентность обработчиков и дедупликация.

1. Термин: Dead Letter Queue (DLQ)

Dead letter queue — это очередь (или топик), в которую попадают сообщения, которые не удалось обработать после всех разрешённых повторных попыток. В контексте LLM инференса это означает, что запрос к модели (например, вызов OpenAI API или локальной модели) завершился ошибкой, и система исчерпала лимит retry.

Зачем нужна DLQ для LLM

• Изоляция ошибок: сбойные запросы не блокируют обработку нормальных.
• Анализ первопричин: можно изучать паттерны ошибок (таймауты, rate limits, битые промпты).
• Репроцессинг: после исправления ошибки (например, увеличения таймаута) можно повторно отправить запросы из DLQ.
• Алертинг: при росте числа сообщений в DLQ — сигнал для инженера.

2. Причины failed LLM инференс запросов

Прежде чем проектировать DLQ, нужно понимать, какие ошибки могут возникать:

Вывод: не все ошибки стоит повторять. DLQ должна получать только те запросы, которые не удалось обработать после разумного числа retry, а также те, которые заведомо не могут быть ретраированы (например, битый промпт).

3. Архитектура с Kafka: main → retry → DLQ

Типичная схема на Apache Kafka:

[Producer] → main_topic → [Consumer] → успех? → (да) завершено
                                    ↓ (нет)
                              retry_topic (с задержкой)
                                    ↓
                              [Consumer] → успех? → завершено
                                    ↓ (нет, исчерпаны retry)
                              dlq_topic

Компоненты:

• main_topic: все входящие запросы на инференс.
• retry_topic: запросы, которые нужно повторить. Каждое сообщение содержит метаданные: количество попыток, время следующей попытки, исходная ошибка.
• DLQ_topic: финальные неудачные запросы.

Реализация задержки:

• В Kafka нет встроенной задержки, поэтому используют Kafka Streams или отдельный retry consumer, который спит перед повторной отправкой.
• Альтернатива: использовать RabbitMQ с dead letter exchange и TTL, или AWS SQS с redrive policy.

Пример конфигурации топиков (в YAML для Kafka):

topics:
  - name: llm_requests_main
    partitions: 10
    replication: 3
  - name: llm_requests_retry
    partitions: 5
    replication: 3
    config:
      cleanup.policy: delete
      retention.ms: 86400000  # 1 день
  - name: llm_requests_dlq
    partitions: 3
    replication: 3
    config:
      cleanup.policy: compact  # храним последнюю версию для анализа

4. Стратегия повторных попыток (Retry Policy)

Exponential backoff с jitter — стандартный подход:

import random
import time

def next_retry_delay(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:
    delay = min(base_delay * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.1)  # 10% jitter
    return delay + jitter

Параметры:

• max_retries: обычно 3–5 для LLM (больше — слишком долго).
• base_delay: 1–2 секунды.
• max_delay: 30–60 секунд.
• retryable errors: список кодов ошибок, которые стоит повторять (429, 500, 503, таймауты).

Пример обработчика на Python с Kafka:

from kafka import KafkaConsumer, KafkaProducer
import json

consumer = KafkaConsumer('llm_requests_main', bootstrap_servers='localhost:9092')
producer = KafkaProducer(bootstrap_servers='localhost:9092')

MAX_RETRIES = 3

for msg in consumer:
    request = json.loads(msg.value)
    try:
        response = call_llm(request['prompt'])
        # успех — сохраняем результат
    except RetryableError as e:
        request['retry_count'] = request.get('retry_count', 0) + 1
        if request['retry_count'] <= MAX_RETRIES:
            delay = next_retry_delay(request['retry_count'])
            request['next_retry_at'] = time.time() + delay
            producer.send('llm_requests_retry', json.dumps(request).encode())
        else:
            producer.send('llm_requests_dlq', json.dumps(request).encode())
    except NonRetryableError as e:
        producer.send('llm_requests_dlq', json.dumps(request).encode())

5. Обработка сообщений из DLQ

После попадания в DLQ возможны три сценария:

5.1 Ручной репроцесс (manual reprocess)

• Инженер анализирует ошибку через дашборд (Grafana, Kibana).
• Исправляет причину (например, обновляет API-ключ).
• Вручную отправляет сообщение обратно в main_topic через UI или скрипт.

5.2 Автоматический репроцесс после фикса (auto-reprocess)

• Если ошибка связана с временным сбоем (например, перезагрузка модели), можно настроить scheduler, который периодически (раз в час) переотправляет сообщения из DLQ в retry_topic.
• Важно: добавить TTL (time-to-live) для сообщений в DLQ, чтобы старые запросы не репроцессились бесконечно.

5.3 Алертинг (alerting)

• При появлении нового сообщения в DLQ отправляется алерт в Sentry, PagerDuty или Slack.
• Алерт должен содержать: ID запроса, тип ошибки, время, количество попыток.

Пример алерта через Sentry:

import sentry_sdk

sentry_sdk.capture_message(
    f"LLM request failed after retries: {request_id}, error: {error_type}",
    level='error'
)

6. Мониторинг и метрики

Для эффективного управления DLQ нужно отслеживать:

Пороги для алертов:

• DLQ depth > 100 за 5 минут → warning.
• DLQ age > 1 час → critical (возможно, проблема не решается автоматически).

7. Альтернативы Kafka

Не всегда нужен Kafka. Для небольших систем подойдут:

Выбор: если у вас уже есть Kafka для других микросервисов — используйте его. Если LLM-сервис изолирован — SQS или RabbitMQ проще.

8. Best practices

• Идемпотентность: обработчик должен безопасно повторять один и тот же запрос (например, если LLM возвращает тот же результат для того же промпта).
• Дедупликация: используйте уникальный request_id в каждом сообщении. При повторной отправке проверяйте, не был ли запрос уже обработан (через Redis или БД).
• TTL для DLQ: сообщения в DLQ не должны храниться вечно. Установите retention (например, 7 дней), после чего они автоматически удаляются.
• Партиционирование: для Kafka используйте ключ партиции на основе request_id, чтобы гарантировать порядок обработки одного запроса.
• Разделение ошибок: разные типы ошибок могут направляться в разные DLQ (например, dlq_timeout, dlq_rate_limit) для упрощения анализа.

9. Пример полной реализации (схематично)

# consumer.py
from kafka import KafkaConsumer, KafkaProducer
import json, time, random

consumer = KafkaConsumer('llm_requests_main', group_id='llm_worker')
producer = KafkaProducer()

def process_request(request):
    # вызов LLM
    response = call_llm(request['prompt'], timeout=30)
    return response

for msg in consumer:
    request = json.loads(msg.value)
    try:
        result = process_request(request)
        # сохраняем результат
    except RetryableError as e:
        retry_count = request.get('retry_count', 0) + 1
        if retry_count <= 3:
            delay = min(2 ** retry_count, 60) + random.uniform(0, 2)
            request['retry_count'] = retry_count
            request['next_retry_at'] = time.time() + delay
            producer.send('llm_requests_retry', json.dumps(request).encode())
        else:
            request['final_error'] = str(e)
            producer.send('llm_requests_dlq', json.dumps(request).encode())
    except NonRetryableError as e:
        request['final_error'] = str(e)
        producer.send('llm_requests_dlq', json.dumps(request).encode())

Пет-проект для закрепления

Задача: Реализовать DLQ для LLM-сервиса, который вызывает OpenAI API, с обработкой rate limit и таймаутов.

Инструменты: Python, Kafka (или Redpanda), Docker, Prometheus + Grafana (опционально).

Шаги:

1. Разверните Kafka в Docker (один брокер, три топика: main, retry, dlq).
2. Напишите producer, который отправляет 1000 запросов с разными промптами (часть из них — заведомо битые, часть — нормальные).
3. Напишите consumer, который:
   • вызывает OpenAI API (используйте openai.ChatCompletion.create с таймаутом 10 с);
   • при ошибке 429 — retry с exponential backoff;
   • при таймауте — retry до 3 раз;
   • при ошибке 400 (bad request) — сразу в DLQ.
4. Настройте consumer для retry_topic, который проверяет next_retry_at и спит до нужного момента.
5. Настройте consumer для DLQ, который логирует ошибки и отправляет алерт в консоль (имитация Sentry).
6. (Опционально) Добавьте метрики через Prometheus client и визуализируйте глубину DLQ в Grafana.

Ожидаемый результат:

• Все успешные запросы обработаны.
• Rate limit запросы повторены и в итоге выполнены.
• Таймаутные запросы после 3 попыток попали в DLQ.
• Битые промпты сразу в DLQ.
• В логах видно, какие запросы и почему ушли в DLQ.

Связь с другими вопросами

Навигация

• Предыдущий: 239
• Следующий: 241
• Индекс: 00. Индекс разборов