Коротко про те, що взагалі робимо зараз на проекті: у нас є окремий hardware сервер (hostname == “Matrix”), на якому з llama.cpp запускаємо свої self-hosted моделі.

В нашому Kubernetes є LiteLLM AI Gateway для наших клієнтів – Backend API і інших сервісів проекту.

Клієнти відправляють запити до OpenAI/Anthropic/OpenRouter до LiteLLM, а LiteLLM передає їх до провайдеру, логує це все, записує traces, контролює доступ і $$$.

Задача зараз складається з двох частин:

• треба налаштувати traffic mirroring: клієнти продовжують працювати з primary provider/model типу OpenAI, а “під капотом” ми на LiteLLM додатково відправляємо запит клієнта на нашу self-hosted LLM (зараз це Gemma)
• потім нам потрібно виконувати LLM Evaluations: після отримання відповідей від моделей OpenAI та Gemma на нашому Matrix – треба оцінити якість відповідей одної і другої моделі

Далі, маючи оцінку відповідей “primary model” типу OpenAI GPT 5.6 Terra і нашої self-hosted, наприклад Gemma – ми зможемо порівнювати якість роботи моделей і підібрати якусь для себе.

А пізніше, можливо, навіть займемося якимось fine-tuning.

Налаштування traffic mirroring описував в пості LiteLLM: Custom Callback для traffic mirroring та OTel tracing до VictoriaTraces – в цьому пості вона теж буде – але зі змінами, бо то був більше PoC і з деякими багами.

Чому робив саме через custom callback – розбирав у LiteLLM: Traffic Mirroring та Batch Completions і трафік до двох провайдерів.

Описане в цьому пості рішення робоче, але це все ж більше просто приклад того, як можна таку задачу вирішити.

Вже під час використання виявились деякі нюанси – про них скажу в кінці.

Код тут – той, що у нас зараз в production, хоча скоріш за все ще будуть зміни.

Ну і щось міг пропустити – бо зазвичай пости пишу прямо під час розборки/налаштування і все записую – але цього разу описую вже готовий проект і щось міг забути.

Пост виглядає довгим – але це за рахунок коду, бо додав всі скрипти повністю.

LLM Evaluations with Phoenix або Opik

Взагалі оцінку відповідей LLM можна робити з Phoenix – див. Running Evals on Traces або Opik – див. Evaluation Overview, і ми їх теж тестуємо (і будемо все ж робити з Opik).

Але зараз задача стояла зробити просте і швидке в реалізації рішення, яке до того ж буде у вигляді звичайних Prometheus-метрик і Grafana dashboard.

Для Opik ми зараз готуємо “golden dataset” – дані, з якими будуть порівнюватись відповіді наших self-hosted моделей, і це трохи окремий геморой – бо навіть в production відповіді не завжди коректні.

В тому рішенні, що описано в цьому пості відповідь primary-моделі використовується як baseline – з ним Judge LLM порівнює відповідь self-hosted моделі.

По суті, це не повноцінний golden dataset, бо він не містить заздалегідь перевірених еталонних відповідей і критеріїв оцінювання – але ми вважаємо відповідь GPT 5.6 як основу, на яку треба рівнятись нашим self-hosted LLMs.

Загальна архітектура

Якщо коротко, то маємо три компоненти:

• LiteLLM:
  • отримує запит від клієнта, шле до primary model (OpenAI)
  • потім викликає наш скрипт з Custom Callback, який записує дані по реквесту від клієнта в Redis
• Mirror Worker:
  • читає дані з Redis, виконує запит до Matrix, передаючи клієнтський запит
  • отриманий результат знов записує до Redis в інший стрім
• Evaluation Worker:
  • читає дані з Redis, передає текст запиту клієнта та обидві відповіді до “Judge Model”
  • Judge Model виконує оцінку якості відповідей і повертає свій score по кожній
  • Evaluation Worker віддає результат у вигляді метрики litellm_evaluation_score з лейблами provider та model до VictoriaMetrics

Схематично це може виглядати так – нижче буде більш детальний опис всього flow:

І по кожному кроку:

• LiteLLM – отримує запити від клієнтів (наших внутрішніх сервісів):
  • проксює ці запити до primary provider/model (OpenAI / GPT 5.6)
  • має свій OpenTelemetry context, створює trace для цього запиту – primary span, child spans, пише їх до VictoriaTraces (див. LiteLLM: метрики, traces та інтеграція з VictoriaMetrics Stack)
  • на кожен успішний запит викликає наш TrafficMirrorCallback – скрипт в LiteLLM Pod
• Custom Callback – виконує Traffic Mirroring:
  • отримує prompt і відповідь основної моделі
  • нормалізує Chat Completions або Responses API input
  • додає поточний OpenTelemetry context від LiteLLM (аби мати загальний трейс)
  • записує “mirror job” у Redis Stream з іменем traffic_mirror:requests
  • Callback не чекає відповіді Matrix – тому доступність і швидкість self-hosted LLM не впливають на запит клієнтів і роботу LiteLLM
    • це основна відмінність від версії з попереднього посту
  • Callback віддає метрику litellm_traffic_mirror_enqueue_total для VictoriaMetrics – скільки задач для Mirror Worker додали в Redis
• Redis:
  • Redis використовується як тимчасова черга між LiteLLM і фоновими workers, тому persistence і всякі fault tolerance не потрібні, максимально простий сетап
  • тримає два Redis Streams для воркерів – traffic_mirror:requests для Mirror Worker і traffic_mirror:evaluations для Evaluation Worker
• Mirror Worker:
  • постійно слухає Redis Stream
  • читає jobs із traffic_mirror:requests через Redis Consumer Group
  • відправляє запит з текстом від клієнта до Matrix і Gemma
  • із Redis разом з іншими даними отримує OTel context, який створював LiteLLM, щоб primary і mirror spans мали спільний trace_id
    • створює окремий “mirror span” з name="traffic_mirror", в атрибутах якого пише модель, ім’я провайдеру як “llama.cpp“, оригінальний prompt від клієнта, текст відповіді від Matrix model
    • в результаті в VictoriaTraces маємо повний trace – primary span від LiteLLM + Matrix span від Mirror Worker
  • після отримання відповіді:
    • створює OpenTelemetry span із Matrix request і response
    • записує результат у логи до VictoriaLogs
    • формує пару original prompt від клієнта + primary response (від моделі в OpenAI) + response від нашої Gemma (чи whatever it will be) на нашому сервері “Matrix”
    • записує її в Redis Stream traffic_mirror:evaluations
  • генерує метрику litellm_traffic_mirror_requests_total з лейблою “outcome” і значеннями success, failed або timeout
• Evaluation Worker:
  • читає готові пари відповідей із Redis stream traffic_mirror:evaluations
  • формує запит до Judge LLM (тут в прикладах це GPT 5.6 Terra, але є нюанси, допишу в кінці), відправляє його “судді”
  • Judge LLM:
    • порівнює обидві відповіді та повертає до Evaluation Worker пояснення свого рішення (reasoning), оцінку primary response, оцінку Matrix response, переможця: primary, matrix або tie (tie – відповіді мають практично однакову якість, тому немає однозначного переможця)
  • Evaluation Worker записує отриманий від Judge LLM результат в лог та генерує Prometheus-метрики:
    • litellm_evaluation_score, litellm_evaluation_errors_total, litellm_evaluation_duration_seconds
    • VictoriaMetrics має VMServiceScrape і збирає ці метрики з ендпоінта на Evaluation Worker

Реалізація – Python та Helm

Взагалі починав писати покрокове створення – але вийшло забагато тексту, тому просто вже фінальний (на даний момент) результат з поясненнями основних моментів.

Ще думав тут описувати і основні функції – але це теж прям полотенце тексту вийшло, тому просто попросив Codex додати детальні коментарі прямо в коді скриптів.

Ну і сам код і Helm на 90% писались з Codex і 5.6 Sol Lite – але код “вичитаний”, протестований.

Отже, для системи Traffic Mirroring та LLM Evaluations є три компоненти, скрипти на Python:

• TrafficMirrorCallback (скрипт traffic_mirror_callback.py): зберігає запис про отриманий від клієнта request до Redis
• Mirror Worker (скрипт mirror_worker.py): отримує дані з Redis, робить запит до Matrix, отримує відповідь від self-hosted LLM, записує результат до Redis для обробки Evaluation Worker
• Evaluation Worker (скрипт evaluation_worker.py): читає з Redis текст запиту клієнта, відповіді обох моделей, відправляє їх на оцінку до Judge model, віддає результат оцінки у вигляді метрик для VictoriaMetrics

LiteLLM деплоїться з Helm-чарту (див. LiteLLM: AI Gateway в Kubernetes та метрики до VictoriaMetrics), тому Traffic Mirroring та LLM Evaluation теж робиться через нього:

• evaluation-redis.yaml: маніфест з Kubernetes Deployment та Service для Redis
• mirror-worker.yaml: Kubernetes Deployment, ConfigMap, Service та VMServiceScrape для Mirror Worker
• evaluation-worker.yaml: описує Kubernetes Deployment, ConfigMap, Service та VMServiceScrape для Evaluation Worker

Деплой Redis

Деплоїться одним маніфестом evaluation-redis.yaml з усіма ресурсами:

{{- if .Values.evaluationRedis.enabled }}
apiVersion: v1
kind: Service
metadata:
  name: {{ .Release.Name }}-evaluation-redis
  labels:
    app.kubernetes.io/name: evaluation-redis
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  selector:
    app.kubernetes.io/name: evaluation-redis
    app.kubernetes.io/instance: {{ .Release.Name }}
  ports:
    - name: redis
      port: 6379
      targetPort: redis
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-evaluation-redis
  labels:
    app.kubernetes.io/name: evaluation-redis
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels:
      app.kubernetes.io/name: evaluation-redis
      app.kubernetes.io/instance: {{ .Release.Name }}
  template:
    metadata:
      labels:
        app.kubernetes.io/name: evaluation-redis
        app.kubernetes.io/instance: {{ .Release.Name }}
    spec:
      containers:
        - name: redis
          image: {{ .Values.evaluationRedis.image | quote }}
          imagePullPolicy: IfNotPresent
          args:
            - redis-server
            - --save
            - ""
            - --appendonly
            - "no"
            - --maxmemory
            - "96mb"
            - --maxmemory-policy
            - "noeviction"
          ports:
            - name: redis
              containerPort: 6379
              protocol: TCP
          readinessProbe:
            exec:
              command: ["redis-cli", "ping"]
            initialDelaySeconds: 2
            periodSeconds: 5
          livenessProbe:
            exec:
              command: ["redis-cli", "ping"]
            initialDelaySeconds: 5
            periodSeconds: 10
          resources:
{{ toYaml .Values.evaluationRedis.resources | indent 12 }}
          volumeMounts:
            - name: data
              mountPath: /data
      volumes:
        - name: data
          emptyDir: {}
      affinity:
{{ toYaml .Values.evaluationRedis.affinity | indent 8 }}
      tolerations:
{{ toYaml .Values.evaluationRedis.tolerations | indent 8 }}
{{- end }}

Values:

...
evaluationRedis:
  enabled: true
  image: redis:7.4-alpine
  resources:
    requests:
      cpu: 25m
      memory: 64Mi
    limits:
      cpu: 100m
      memory: 128Mi
  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
          - matchExpressions:
              - key: component
                operator: In
                values:
                  - llm
  tolerations:
    - key: LLMOnly
      operator: Exists
      effect: NoSchedule
...

LiteLLM Custom Callback

Важлива різниця з тим, що описано в попередньому пості: тепер Callback виконує тільки запис в Redis – але не виконує запит до Matrix, як було в попередньому варіанті.

Причина – LiteLLM і його LoggingWorker, який дропав виконання запитів довше двох секунд.

Крім того, в попередній версії був баг, бо формат даних, який повертає Responses API, очікувано відрізняється від Chat Completions API, з яким ми робимо запити до llama.cpp – тому тут вже додана нормалізація запитів в форматі, який “зрозумілий” Chat Completions API.

Скрипт traffic_mirror_callback.py

Викликається LiteLLM після успішної відповіді primary-моделі:

• має sampling через TRAFFIC_MIRROR_SAMPLE_RATE: можна налаштувати, яку частину запитів пересилати на Matrix-  зменшуємо навантаження на Matrix і витрати на Judge LLM
• отримує оригінальний prompt, primary response, модель, провайдера та response ID:
  • нормалізує Chat Completions і Responses API input до формату messages, який підтримує Matrix/llama.cpp Chat Completions API
• зберігає поточний OTEL context у trace_context, щоб Mirror Worker міг продовжити той самий trace
• записує job у Redis Stream traffic_mirror:requests
• через /metrics endpoint самого LiteLLM (бо callback виконується всередині процесу LiteLLM)  повертає метрику litellm_traffic_mirror_enqueue_total з результатом enqueued, skipped, failed або cancelled, метрику забирає VMAgent
• працює за принципом fail-open: проблеми з mirroring не впливають на відповідь LiteLLM клієнту

Весь скрипт:

"""Queue successful LiteLLM requests for asynchronous traffic mirroring.

The callback runs inside the LiteLLM proxy process after the primary provider
has returned a successful response. It performs only bounded local work:

* captures response data that LiteLLM's Responses API OTEL span omits;
* applies probabilistic sampling;
* normalizes Chat Completions or Responses API input for llama.cpp;
* injects the active W3C trace context into the queued payload; and
* writes one job to the ``traffic_mirror:requests`` Redis Stream.

Matrix is deliberately not called here. Keeping the callback short prevents a
slow or unavailable self-hosted model from delaying LiteLLM logging workers or
affecting the response that has already been returned to the client.
"""

import asyncio
import json
import os
import random
from datetime import datetime
from typing import Any

from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs
from opentelemetry import trace
from opentelemetry.propagate import inject
from prometheus_client import Counter
from redis.asyncio import Redis


# Matrix target metadata is stored in the queued job and metric labels. The
# callback does not contact Matrix directly; mirror_worker.py does that.
MATRIX_MODEL = os.environ["TRAFFIC_MIRROR_MODEL"]

# Fraction of successful primary requests to mirror: 0.0 disables mirroring,
# 1.0 mirrors every request, and 0.1 mirrors approximately one request in ten.
MATRIX_SAMPLE_RATE = float(os.environ["TRAFFIC_MIRROR_SAMPLE_RATE"])

# Environment name (for example, test or ops) used as a low-cardinality label.
LITELLM_ENV = os.environ["LITELLM_ENV"]

# Dedicated Redis Stream used to hand work off from LiteLLM to Mirror Worker.
MIRROR_REDIS_URL = os.environ["TRAFFIC_MIRROR_REDIS_URL"]
MIRROR_REDIS_STREAM = os.environ["TRAFFIC_MIRROR_REDIS_STREAM"]

# Approximate maximum number of entries retained in the Stream. This bounds
# memory usage; it is not a limit on prompt or response length.
MIRROR_REDIS_STREAM_MAXLEN = int(
    os.environ["TRAFFIC_MIRROR_REDIS_STREAM_MAXLEN"]
)

# Keep Redis operations short so an unavailable queue cannot delay LiteLLM.
MIRROR_REDIS_TIMEOUT_SECONDS = float(
    os.environ["TRAFFIC_MIRROR_REDIS_TIMEOUT_SECONDS"]
)

if not 0.0 <= MATRIX_SAMPLE_RATE <= 1.0:
    raise ValueError("TRAFFIC_MIRROR_SAMPLE_RATE must be between 0.0 and 1.0")

# The client is shared by callback invocations and creates connections lazily.
mirror_redis = Redis.from_url(
    MIRROR_REDIS_URL,
    decode_responses=True,
    socket_connect_timeout=MIRROR_REDIS_TIMEOUT_SECONDS,
    socket_timeout=MIRROR_REDIS_TIMEOUT_SECONDS,
)
# Records whether sampling skipped the request or Redis accepted/rejected it.
TRAFFIC_MIRROR_ENQUEUE_TOTAL = Counter(
    "litellm_traffic_mirror_enqueue_total",
    "Traffic mirror callback enqueue attempts by outcome.",
    ["litellm_env", "outcome", "matrix_model", "error_type"],
)


def _record_enqueue_outcome(
    outcome: str,
    error_type: str = "none",
    amount: float = 1,
) -> None:
    """Increment the bounded-cardinality mirror enqueue counter.

    ``outcome`` describes the callback result (enqueued, skipped, failed, or
    cancelled). ``error_type`` contains only the exception class, never its
    message, so arbitrary error text cannot create unbounded Prometheus series.
    ``amount=0`` is used during startup to pre-create dashboard time series.
    """
    TRAFFIC_MIRROR_ENQUEUE_TOTAL.labels(
        litellm_env=LITELLM_ENV,
        outcome=outcome,
        matrix_model=MATRIX_MODEL,
        error_type=error_type,
    ).inc(amount)


# Pre-create every expected time series so Grafana can display zero before the
# first event instead of showing "No data".
for outcome in ("enqueued", "failed", "cancelled", "skipped"):
    _record_enqueue_outcome(outcome, amount=0)


def _json(value: Any) -> str:
    """Serialize SDK objects and plain Python values into a JSON string.

    LiteLLM/OpenAI responses may be Pydantic models, hence ``model_dump``.
    ``ensure_ascii=False`` keeps prompts readable in logs, while ``default=str``
    prevents uncommon SDK value types from breaking the fail-open callback.
    """
    if hasattr(value, "model_dump"):
        value = value.model_dump()
    return json.dumps(value, ensure_ascii=False, default=str)


def _primary_input(kwargs: dict[str, Any]) -> Any:
    """Return the original caller input from LiteLLM callback kwargs.

    Chat Completions uses ``messages`` and Responses API uses ``input``. This
    value is preserved without normalization for logs, traces, and evaluation;
    ``_matrix_messages`` separately builds the format accepted by Matrix.
    """
    return kwargs.get("messages") or kwargs.get("input")


def _content_text(content: Any) -> str | None:
    """Flatten message content into text accepted by Chat Completions.

    Chat Completions commonly supplies a string, while Responses API supplies a
    list of typed content parts such as ``input_text``. Unsupported/non-text
    parts are ignored because Matrix currently receives text-only messages.
    """
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return None

    # Preserve the original order when a message consists of several text parts.
    texts: list[str] = []
    for part in content:
        if hasattr(part, "model_dump"):
            part = part.model_dump()
        if not isinstance(part, dict):
            continue
        text = part.get("text")
        if isinstance(text, str):
            texts.append(text)
    return "\n".join(texts) or None


def _normalize_message_items(items: list[Any]) -> list[dict[str, str]]:
    """Convert Chat Completions/Responses items for Matrix chat completions.

    Matrix exposes a Chat Completions-compatible llama.cpp endpoint. Responses
    API items, developer messages, and tool calls therefore need to be reduced
    to ordinary system/user/assistant text messages before they are queued.
    """
    messages: list[dict[str, str]] = []
    for item in items:
        # LiteLLM may pass either dictionaries or Pydantic SDK objects depending
        # on the endpoint and the stage at which the callback was invoked.
        if hasattr(item, "model_dump"):
            item = item.model_dump()
        if not isinstance(item, dict):
            continue

        role = item.get("role")
        content = _content_text(item.get("content"))
        # llama.cpp follows the Chat Completions role set. Responses API's
        # developer role has equivalent precedence here and becomes system.
        if role in {"system", "developer", "user", "assistant"} and content:
            messages.append(
                {
                    "role": "system" if role == "developer" else role,
                    "content": content,
                }
            )
        # Preserve tool output as text instead of dropping useful conversation
        # context. A user message is broadly supported by chat templates.
        elif role == "tool" and content:
            messages.append(
                {
                    "role": "user",
                    "content": (
                        f"Tool result for "
                        f"{item.get('tool_call_id', 'unknown')}:\n{content}"
                    ),
                }
            )

        # Chat Completions stores function calls inside assistant.tool_calls.
        # Convert each call to readable text because Matrix need not execute it.
        for tool_call in item.get("tool_calls") or []:
            if hasattr(tool_call, "model_dump"):
                tool_call = tool_call.model_dump()
            if not isinstance(tool_call, dict):
                continue
            function = tool_call.get("function") or {}
            if hasattr(function, "model_dump"):
                function = function.model_dump()
            if not isinstance(function, dict):
                function = {}
            messages.append(
                {
                    "role": "assistant",
                    "content": (
                        f"Tool call {function.get('name', 'unknown')}: "
                        f"{function.get('arguments', '')}"
                    ),
                }
            )

        # Responses API represents function calls and their outputs as separate
        # top-level items rather than nested Chat Completions messages.
        item_type = item.get("type")
        if item_type == "function_call":
            messages.append(
                {
                    "role": "assistant",
                    "content": (
                        f"Tool call {item.get('name', 'unknown')}: "
                        f"{item.get('arguments', '')}"
                    ),
                }
            )
        elif item_type == "function_call_output":
            output = item.get("output")
            if not isinstance(output, str):
                output = _json(output)
            messages.append(
                {
                    "role": "user",
                    "content": (
                        f"Tool result for {item.get('call_id', 'unknown')}:\n"
                        f"{output}"
                    ),
                }
            )

    # Silently sending an empty prompt would create a misleading evaluation.
    if not messages:
        raise ValueError("Primary input contains no mirrorable text")
    return messages


def _matrix_messages(kwargs: dict[str, Any]) -> list[dict[str, str]]:
    """Select and normalize the request shape used for the Matrix call.

    LiteLLM may expose ``messages`` even for some Responses API calls, so it is
    checked first and normalized rather than returned verbatim. A plain string
    Responses input becomes one user message; list input follows the shared
    item normalizer.
    """
    chat_messages = kwargs.get("messages")
    if isinstance(chat_messages, list):
        return _normalize_message_items(chat_messages)

    responses_input = kwargs.get("input")
    if isinstance(responses_input, str):
        return [{"role": "user", "content": responses_input}]
    if isinstance(responses_input, list):
        return _normalize_message_items(responses_input)
    raise ValueError("Primary request has no supported messages or input")


def _primary_output(response_obj: Any) -> Any:
    """Return the primary assistant result from either response object shape.

    Chat Completions returns ``choices[0].message.content``. Responses API may
    expose the convenient ``output_text`` property or only structured ``output``
    items; the latter are preserved as-is for later evaluation and tracing.
    """
    choices = getattr(response_obj, "choices", None)
    if choices:
        message = getattr(choices[0], "message", None)
        return getattr(message, "content", None)

    output_text = getattr(response_obj, "output_text", None)
    if output_text:
        return output_text
    return getattr(response_obj, "output", None)


def _output_messages(response_obj: Any) -> list[Any]:
    """Convert Responses API output items to OTEL-serializable dictionaries."""
    items = getattr(response_obj, "output", None) or []
    return [
        item.model_dump() if hasattr(item, "model_dump") else item
        for item in items
        if item is not None
    ]


# Use-case labels a caller sends in the request ``metadata`` object. They name
# the calling feature, so a dataset can be built per use case instead of per key.
USE_CASE_KEYS = ("route_key", "category", "surface", "operation", "user_id")


def _requester_metadata(kwargs: dict[str, Any]) -> dict[str, Any]:
    """Read the ``metadata`` object the caller sent in the request body.

    The proxy copies that object to ``requester_metadata``. It keeps its own
    request metadata under ``metadata`` for most routes, but under
    ``litellm_metadata`` for the Responses API, so the LiteLLM helper resolves
    the correct dict. The OpenTelemetry exporter reads ``requester_metadata``
    for metric attributes only, so the labels never reach a span.
    """
    metadata = get_litellm_metadata_from_kwargs(kwargs) or {}
    requester = metadata.get("requester_metadata") if isinstance(metadata, dict) else None
    return requester if isinstance(requester, dict) else {}


def _record_output_on_span(
    kwargs: dict[str, Any], response_obj: Any, response_id: str | None, model: Any
) -> None:
    """Record the response and the use case of every successful request.

    The LiteLLM OpenTelemetry v2 exporter writes ``gen_ai.output.messages``
    for Chat Completions only. Responses API traffic therefore reaches
    VictoriaTraces with the request but without the response, which makes it
    unusable for evaluation.

    The proxy request span has already ended when LiteLLM runs success
    callbacks, so attributes cannot be added to it. This emits a small
    ``response_capture`` span in the same trace instead. Join it to the
    ``chat <model>`` span that holds ``gen_ai.input.messages`` by ``trace_id``.

    The span also carries the caller use-case labels as ``hos.*`` attributes.
    It is emitted for Chat Completions too, which the upstream exporter already
    covers, because those requests need the same labels.
    """
    try:
        if not trace.get_current_span().get_span_context().is_valid:
            return
        # Chat Completions responses are already covered by the upstream
        # exporter, which reads choices[]. Do not write them a second time.
        covered = bool(getattr(response_obj, "choices", None))
        messages = [] if covered else _output_messages(response_obj)
        labels = _requester_metadata(kwargs)
        if not messages and not labels:
            return
        span = trace.get_tracer("litellm.traffic_mirror.callback").start_span(
            "response_capture"
        )
        try:
            if messages:
                span.set_attribute("gen_ai.output.messages", _json(messages))
            span.set_attribute("gen_ai.request.model", str(model))
            if response_id:
                span.set_attribute("gen_ai.response.id", str(response_id))
            for key in USE_CASE_KEYS:
                value = labels.get(key)
                if value:
                    span.set_attribute(f"hos.{key}", str(value))
        finally:
            span.end()
    except Exception as error:
        # Tracing must never affect an already successful primary response.
        verbose_proxy_logger.warning(
            "TRAFFIC_MIRROR_SPAN_OUTPUT_FAILED response_id=%s error=%r",
            response_id,
            error,
        )


def _primary_provider(kwargs: dict[str, Any]) -> str:
    """Read the provider LiteLLM actually selected for the primary request.

    This is OpenAI, OpenRouter, Anthropic, and so on—not Matrix. Router metadata
    normally stores it under ``litellm_params``; the top-level lookup supports
    callback shapes where LiteLLM exposes it directly.
    """
    litellm_params = kwargs.get("litellm_params") or {}
    if isinstance(litellm_params, dict):
        provider = litellm_params.get("custom_llm_provider")
        if provider:
            return str(provider)
    return str(kwargs.get("custom_llm_provider") or "unknown")


async def _enqueue_mirror(payload: dict[str, Any]) -> str:
    """Add a bounded mirror job to Redis and return the generated entry ID.

    ``MAXLEN ~`` trims approximately rather than on every exact boundary, which
    is cheaper for Redis. The limit controls the number of Stream entries, not
    the byte or token length of an individual prompt/response payload.
    """
    return await mirror_redis.xadd(
        MIRROR_REDIS_STREAM,
        {"payload": _json(payload)},
        maxlen=MIRROR_REDIS_STREAM_MAXLEN,
        approximate=True,
    )


class TrafficMirrorCallback(CustomLogger):
    """LiteLLM hook that quickly queues sampled successful primary requests."""

    async def async_log_success_event(
        self,
        kwargs: dict[str, Any],
        response_obj: Any,
        start_time: datetime,
        end_time: datetime,
    ) -> None:
        """Handle one successful LiteLLM request without changing its result.

        LiteLLM supplies request data in ``kwargs`` and the completed primary
        response in ``response_obj``. ``start_time`` and ``end_time`` are part of
        the CustomLogger interface; only ``end_time`` is stored as job creation
        time. All failures are logged and swallowed, except cancellation, so
        mirroring remains fail-open for clients.
        """
        response_id = getattr(response_obj, "id", None)

        # Trace capture is not sampled: VictoriaTraces must hold the response
        # of every successful request, not only of mirrored ones.
        _record_output_on_span(kwargs, response_obj, response_id, kwargs.get("model"))

        # Decide before serialization or Redis I/O to keep skipped calls cheap.
        if random.random() >= MATRIX_SAMPLE_RATE:
            _record_enqueue_outcome("skipped")
            verbose_proxy_logger.debug(
                "TRAFFIC_MIRROR_SKIPPED response_id=%s sample_rate=%s",
                response_id,
                MATRIX_SAMPLE_RATE,
            )
            return

        # Keep two forms of the same request: the untouched input is used by the
        # Judge, while matrix_messages is normalized for llama.cpp compatibility.
        primary_input = _primary_input(kwargs)
        primary_response = _primary_output(response_obj)
        verbose_proxy_logger.info(
            "TRAFFIC_MIRROR_CALLBACK primary_model=%s response_id=%s "
            "input=%s primary_response=%s",
            kwargs.get("model"),
            response_id,
            _json(primary_input),
            _json(primary_response),
        )

        try:
            # Capture W3C trace headers from LiteLLM's current span. Mirror
            # Worker restores them and creates its span in the same trace.
            trace_context: dict[str, str] = {}
            inject(trace_context)
            # The job is self-contained: Mirror Worker requires no LiteLLM DB
            # lookup to reconstruct the primary request, response, or trace.
            stream_entry_id = await _enqueue_mirror(
                {
                    "created_at": end_time.isoformat(),
                    "trace_context": trace_context,
                    "sample_rate": MATRIX_SAMPLE_RATE,
                    "prompt": primary_input,
                    "matrix_messages": _matrix_messages(kwargs),
                    "primary": {
                        "provider": _primary_provider(kwargs),
                        "model": kwargs.get("model"),
                        "response_id": response_id,
                        "response": primary_response,
                    },
                }
            )
            _record_enqueue_outcome("enqueued")
            verbose_proxy_logger.info(
                "TRAFFIC_MIRROR_ENQUEUED primary_response_id=%s "
                "stream=%s entry_id=%s",
                response_id,
                MIRROR_REDIS_STREAM,
                stream_entry_id,
            )
        except asyncio.CancelledError as error:
            # Cancellation must propagate so LiteLLM can manage task shutdown.
            _record_enqueue_outcome("cancelled", type(error).__name__)
            verbose_proxy_logger.exception(
                "TRAFFIC_MIRROR_ENQUEUE_CANCELLED primary_response_id=%s "
                "error_type=%s error=%r",
                response_id,
                type(error).__name__,
                error,
            )
            raise
        except Exception as error:
            # Redis/normalization failures must not affect the primary response.
            _record_enqueue_outcome("failed", type(error).__name__)
            verbose_proxy_logger.exception(
                "TRAFFIC_MIRROR_ENQUEUE_FAILED primary_response_id=%s "
                "stream=%s error_type=%s error=%r",
                response_id,
                MIRROR_REDIS_STREAM,
                type(error).__name__,
                error,
            )


# LiteLLM imports this module-level object from proxy-config.yaml.
traffic_mirror_callback = TrafficMirrorCallback()

Helm templates та values

Скрипт traffic_mirror_callback.py запускається самим LiteLLM, який додається через callbacks в litellm_settings:

...
    litellm_settings:
      # Monitoring settings
      ...
      enable_end_user_cost_tracking_prometheus_only: true
      callbacks:
      - prometheus
      - arize_phoenix
      - traffic_mirror_callback.traffic_mirror_callback
...

Сам скрипт при деплої записується у ConfigMap:

apiVersion: v1
kind: ConfigMap
metadata:
  name: litellm-traffic-mirror-callback
data:
  traffic_mirror_callback.py: |
{{ .Files.Get "files/traffic_mirror_callback.py" | indent 4 }}

Який потім підключається до LiteLLM Pods:

...
volumes:
  - name: traffic-mirror-callback
    configMap:
      name: litellm-traffic-mirror-callback

volumeMounts:
  - name: traffic-mirror-callback
    mountPath: /etc/litellm/traffic_mirror_callback.py
    subPath: traffic_mirror_callback.py
...

В values чарту задаються змінні для підключення до Redis:

envVars:
  LITELLM_ENV: "ops"

  TRAFFIC_MIRROR_REDIS_URL: "redis://litellm-evaluation-redis:6379/0"
  TRAFFIC_MIRROR_REDIS_STREAM: "traffic_mirror:requests"
  TRAFFIC_MIRROR_REDIS_STREAM_MAXLEN: "200"
  TRAFFIC_MIRROR_REDIS_TIMEOUT_SECONDS: "0.5"

  TRAFFIC_MIRROR_MODEL: "gemma-4-26b-a4b-it-q8"
  TRAFFIC_MIRROR_SAMPLE_RATE: "0.1"

Mirror Worker

Задача – відправка запитів до нашої self-hosted моделі, збереження її відповіді, яку потім використовує Evaluation Worker.

Скрипт mirror_worker.py

• постійно слухає Redis Stream traffic_mirror:requests через Redis Consumer Group
  • отримує jobs, які записав TrafficMirrorCallback
  • відновлює OTEL context із trace_context
• надсилає запит до Matrix/llama.cpp і Gemma
• створює span traffic_mirror <model_name> у тому самому trace, що й primary-запит
• записує в атрибути цього span дані з prompt, primary response, Matrix response, reasoning, model, response ID і finish_reason
• формує payload із prompt та двома відповідями від обох моделей, записує його у Redis Stream traffic_mirror:evaluations
• обробляє кілька запитів паралельно відповідно до MIRROR_CONCURRENCY
• підтверджує оброблені Redis jobs через XACK
• на власному /metrics endpoint повертає метрику litellm_traffic_mirror_requests_total з результатом success, failed або timeout, яку забирає VMAgent
• деталі помилок, включно з HTTP status і response body від Matrix, записує в логи

Весь скрипт:

"""Consume mirror jobs, call Matrix, and queue response pairs for evaluation.

The worker is isolated from the LiteLLM request path. It continuously reads the
``traffic_mirror:requests`` Redis Stream through a consumer group, restores the
trace context captured by the callback, and invokes Matrix's OpenAI-compatible
llama.cpp endpoint. A successful primary/Matrix pair is then written to the
``traffic_mirror:evaluations`` Stream for evaluation_worker.py.

Matrix failures are terminal in this PoC: they are logged, counted, and ACKed
instead of being retried forever. Evaluation enqueue failures are also logged,
but do not retroactively change a successful Matrix span into a failed request.
"""

import asyncio
import json
import logging
import os
import socket
from typing import Any

import httpx
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.propagate import extract
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import SpanKind, Status, StatusCode
from prometheus_client import Counter, start_http_server
from redis.asyncio import Redis
from redis.exceptions import ResponseError


# Source queue populated by TrafficMirrorCallback. A consumer group distributes
# jobs between worker replicas without processing a new entry twice.
REDIS_URL = os.environ["MIRROR_REDIS_URL"]
MIRROR_STREAM = os.environ["MIRROR_REDIS_STREAM"]
MIRROR_CONSUMER_GROUP = os.environ["MIRROR_REDIS_CONSUMER_GROUP"]

# Maximum number of Matrix jobs processed concurrently by this worker process.
MIRROR_CONCURRENCY = int(os.environ["MIRROR_CONCURRENCY"])

# Destination queue consumed by evaluation_worker.py after Matrix responds.
EVALUATION_STREAM = os.environ["EVALUATION_REDIS_STREAM"]
EVALUATION_STREAM_MAXLEN = int(os.environ["EVALUATION_REDIS_STREAM_MAXLEN"])

# OpenAI-compatible llama.cpp endpoint and request timeout.
MATRIX_URL = os.environ["TRAFFIC_MIRROR_URL"]
MATRIX_MODEL = os.environ["TRAFFIC_MIRROR_MODEL"]
MATRIX_TIMEOUT_SECONDS = float(os.environ["TRAFFIC_MIRROR_TIMEOUT_SECONDS"])

# Prometheus exporter and environment label used for worker-owned metrics.
METRICS_PORT = int(os.environ["MIRROR_METRICS_PORT"])
LITELLM_ENV = os.environ["LITELLM_ENV"]

# OTLP destination and resource metadata for the Matrix client span.
OTEL_ENDPOINT = os.environ["OTEL_ENDPOINT"]
OTEL_PROJECT_NAME = os.environ["OTEL_PROJECT_NAME"]

# Hostname distinguishes this Redis consumer from other worker pods.
CONSUMER_NAME = socket.gethostname()
# XREADGROUP blocks for five seconds, then returns so the event loop can cycle.
READ_BLOCK_MS = 5_000

# Kubernetes collects stdout/stderr, so use timestamped single-line messages
# that VictoriaLogs can search by stable event names such as MIRROR_JOB_COMPLETED.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("mirror-worker")

# One async client is shared for group creation, reads, writes, and ACKs.
# Decoding to strings lets JSON payloads go directly into json.loads.
redis = Redis.from_url(REDIS_URL, decode_responses=True)

# Export Matrix spans with the same resource fields as LiteLLM spans.
provider = TracerProvider(
    resource=Resource.create(
        {
            "service.name": "litellm",
            "deployment.environment": LITELLM_ENV,
            "openinference.project.name": OTEL_PROJECT_NAME,
        }
    )
)
# Batch export keeps OTLP network I/O off the hot processing coroutine.
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint=OTEL_ENDPOINT))
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("litellm.traffic_mirror.worker")

# Counts Matrix processing outcomes without request IDs or other unbounded labels.
TRAFFIC_MIRROR_REQUESTS_TOTAL = Counter(
    "litellm_traffic_mirror_requests_total",
    "Traffic mirror requests by outcome.",
    ["litellm_env", "outcome", "matrix_model", "error_type"],
)


def _record_mirror_outcome(
    outcome: str,
    error_type: str = "none",
    amount: float = 1,
) -> None:
    """Record one bounded Matrix outcome for Prometheus dashboards.

    ``error_type`` contains an exception class only. Full error messages and
    HTTP response bodies belong in logs because using them as labels would cause
    unbounded metric cardinality. ``amount=0`` initializes empty series.
    """
    TRAFFIC_MIRROR_REQUESTS_TOTAL.labels(
        litellm_env=LITELLM_ENV,
        outcome=outcome,
        matrix_model=MATRIX_MODEL,
        error_type=error_type,
    ).inc(amount)


# Initialize known series so dashboards show zero before the first event.
for outcome in ("success", "failed", "timeout"):
    _record_mirror_outcome(outcome, amount=0)


def _json(value: Any) -> str:
    """Serialize payloads, log values, and OTEL attributes consistently."""
    return json.dumps(value, ensure_ascii=False, default=str)


async def _request_matrix(messages: list[dict[str, Any]]) -> dict[str, Any]:
    """Send normalized messages to Matrix and return decoded response JSON.

    ``raise_for_status`` turns 4xx/5xx responses into HTTPStatusError so
    ``_handle_entry`` can log the status and bounded response body separately
    from network timeouts and unexpected processing errors.
    """
    async with httpx.AsyncClient(timeout=MATRIX_TIMEOUT_SECONDS) as client:
        response = await client.post(
            MATRIX_URL,
            json={
                "model": MATRIX_MODEL,
                "messages": messages,
                "temperature": 0,
                "max_tokens": 1024,
            },
        )
        response.raise_for_status()
        return response.json()


async def _enqueue_evaluation(payload: dict[str, Any]) -> str:
    """Queue a completed response pair and return its Redis Stream entry ID.

    Approximate trimming bounds the ephemeral queue with less Redis overhead.
    The entry remains available until Evaluation Worker processes and ACKs it.
    """
    return await redis.xadd(
        EVALUATION_STREAM,
        {"payload": _json(payload)},
        maxlen=EVALUATION_STREAM_MAXLEN,
        approximate=True,
    )


async def _create_consumer_group() -> None:
    """Create the source Stream and consumer group on worker startup.

    ``id=0`` makes entries already present in a newly created Stream group
    eligible for delivery. ``mkstream=True`` also supports a fresh ephemeral
    Redis. BUSYGROUP simply means another startup already created the group.
    """
    try:
        await redis.xgroup_create(
            MIRROR_STREAM,
            MIRROR_CONSUMER_GROUP,
            id="0",
            mkstream=True,
        )
        logger.info(
            "MIRROR_CONSUMER_GROUP_CREATED stream=%s group=%s",
            MIRROR_STREAM,
            MIRROR_CONSUMER_GROUP,
        )
    except ResponseError as error:
        if "BUSYGROUP" not in str(error):
            raise


async def _process(job: dict[str, Any]) -> dict[str, Any]:
    """Process one decoded job and return IDs useful for completion logs.

    The function restores distributed tracing, records request/primary context,
    calls Matrix, records Matrix output, and finally queues the complete pair.
    Matrix exceptions are attached to the span and re-raised for classification
    by ``_handle_entry``.
    """
    # Restore the W3C context injected by the callback before it queued the job.
    parent_context = extract(job.get("trace_context") or {})
    # Primary contains provider, model, response ID, and response text captured
    # after LiteLLM completed the original request.
    primary = job["primary"]

    with tracer.start_as_current_span(
        f"traffic_mirror {MATRIX_MODEL}",
        context=parent_context,
        kind=SpanKind.CLIENT,
    ) as span:
        # Attributes use GenAI semantic conventions where an equivalent exists;
        # traffic_mirror.* fields preserve primary-specific comparison data.
        span.set_attribute("gen_ai.operation.name", "chat")
        span.set_attribute("gen_ai.system", "llama.cpp")
        span.set_attribute("gen_ai.provider.name", "llama.cpp")
        span.set_attribute("gen_ai.request.model", MATRIX_MODEL)
        span.set_attribute("gen_ai.request.max_tokens", 1024)
        span.set_attribute("gen_ai.request.temperature", 0.0)
        span.set_attribute(
            "gen_ai.input.messages",
            _json(job["matrix_messages"]),
        )
        span.set_attribute("traffic_mirror.primary.input", _json(job["prompt"]))
        span.set_attribute(
            "traffic_mirror.sample_rate",
            float(job["sample_rate"]),
        )
        span.set_attribute("traffic_mirror.primary.model", str(primary["model"]))
        span.set_attribute(
            "traffic_mirror.primary.response_id",
            str(primary["response_id"]),
        )
        span.set_attribute(
            "traffic_mirror.primary.response",
            _json(primary["response"]),
        )

        # The network request lives inside the CLIENT span so duration and any
        # exception represent the actual call to the external Matrix service.
        try:
            matrix_response = await _request_matrix(job["matrix_messages"])
        except Exception as error:
            span.record_exception(error)
            span.set_status(Status(StatusCode.ERROR, str(error)))
            raise

        # Matrix uses the Chat Completions response shape. Missing required
        # fields intentionally raise and are counted as processing failures.
        matrix_choice = matrix_response["choices"][0]
        matrix_message = matrix_choice["message"]
        # Convert the numeric OTEL trace ID to the 32-character representation
        # used by VictoriaTraces and carry it into the evaluation log payload.
        trace_id = f"{span.get_span_context().trace_id:032x}"

        span.set_attribute("gen_ai.response.id", str(matrix_response.get("id")))
        span.set_attribute(
            "gen_ai.response.model",
            str(matrix_response.get("model", MATRIX_MODEL)),
        )
        span.set_attribute(
            "gen_ai.response.finish_reasons",
            _json([matrix_choice.get("finish_reason")]),
        )
        span.set_attribute("gen_ai.output.messages", _json([matrix_message]))
        span.set_attribute(
            "traffic_mirror.matrix.response",
            _json(matrix_message.get("content")),
        )
        span.set_attribute(
            "traffic_mirror.matrix.reasoning",
            _json(matrix_message.get("reasoning_content")),
        )
        span.set_status(Status(StatusCode.OK))

        # Full content and reasoning stay in logs/traces, never metric labels.
        logger.info(
            "TRAFFIC_MIRROR_MATRIX primary_response_id=%s matrix_model=%s "
            "matrix_response_id=%s finish_reason=%s matrix_response=%s "
            "matrix_reasoning=%s",
            primary["response_id"],
            MATRIX_MODEL,
            matrix_response.get("id"),
            matrix_choice.get("finish_reason"),
            _json(matrix_message.get("content")),
            _json(matrix_message.get("reasoning_content")),
        )

        try:
            # Evaluation enqueue is intentionally independent: a failure here
            # does not turn the already successful Matrix request into an error.
            stream_entry_id = await _enqueue_evaluation(
                {
                    "created_at": job["created_at"],
                    "trace_id": trace_id,
                    "prompt": job["prompt"],
                    "primary": primary,
                    "matrix": {
                        "model": MATRIX_MODEL,
                        "response_id": matrix_response.get("id"),
                        "response": matrix_message.get("content"),
                    },
                }
            )
            span.set_attribute("evaluation.enqueued", True)
            span.set_attribute(
                "evaluation.redis_stream_entry_id",
                stream_entry_id,
            )
            logger.info(
                "TRAFFIC_MIRROR_EVALUATION_ENQUEUED "
                "primary_response_id=%s stream=%s entry_id=%s",
                primary["response_id"],
                EVALUATION_STREAM,
                stream_entry_id,
            )
        except Exception as error:
            span.set_attribute("evaluation.enqueued", False)
            logger.exception(
                "TRAFFIC_MIRROR_EVALUATION_ENQUEUE_FAILED "
                "primary_response_id=%s stream=%s error_type=%s error=%r",
                primary["response_id"],
                EVALUATION_STREAM,
                type(error).__name__,
                error,
            )

        # Keep completion logs compact: the full responses have already been
        # recorded in the span and evaluation Stream payload.
        return {
            "trace_id": trace_id,
            "primary_response_id": primary["response_id"],
            "matrix_response_id": matrix_response.get("id"),
        }


async def _handle_entry(entry_id: str, fields: dict[str, str]) -> None:
    """Decode, process, classify, and finally acknowledge one Stream entry.

    Timeout and HTTP status failures receive distinct metrics/log fields. The
    ACK is intentionally outside the exception branches, so both successful and
    terminally failed jobs leave the consumer group's pending list.
    """
    try:
        job = json.loads(fields["payload"])
        result = await _process(job)
        _record_mirror_outcome("success")
        logger.info(
            "MIRROR_JOB_COMPLETED entry_id=%s result=%s",
            entry_id,
            _json(result),
        )
    # Cancellation signals process shutdown and must propagate. Because control
    # leaves before XACK, Redis retains this entry as pending rather than losing it.
    except asyncio.CancelledError:
        raise
    # A timeout is operationally different from a Matrix HTTP rejection and has
    # its own outcome value for alerts and Grafana panels.
    except httpx.TimeoutException as error:
        _record_mirror_outcome("timeout", type(error).__name__)
        logger.exception(
            "TRAFFIC_MIRROR_MATRIX_FAILED entry_id=%s "
            "error_type=%s error=%r",
            entry_id,
            type(error).__name__,
            error,
        )
    # Limit response_body to 2,000 characters to retain the useful API error
    # without flooding Kubernetes/VictoriaLogs with a large upstream response.
    except httpx.HTTPStatusError as error:
        _record_mirror_outcome("failed", type(error).__name__)
        logger.exception(
            "TRAFFIC_MIRROR_MATRIX_FAILED entry_id=%s "
            "error_type=%s status_code=%s response_body=%s error=%r",
            entry_id,
            type(error).__name__,
            error.response.status_code,
            _json(error.response.text[:2_000]),
            error,
        )
    except Exception as error:
        _record_mirror_outcome("failed", type(error).__name__)
        logger.exception(
            "TRAFFIC_MIRROR_MATRIX_FAILED entry_id=%s "
            "error_type=%s error=%r",
            entry_id,
            type(error).__name__,
            error,
        )

    # Acknowledge success and terminal failures alike. Failed Matrix jobs are
    # observable through logs/metrics and are not retried indefinitely.
    await redis.xack(
        MIRROR_STREAM,
        MIRROR_CONSUMER_GROUP,
        entry_id,
    )


async def main() -> None:
    """Expose metrics and continuously consume mirror jobs in small batches."""
    # prometheus_client serves /metrics in a background thread; the asyncio event
    # loop remains dedicated to Redis and concurrent Matrix HTTP requests.
    start_http_server(METRICS_PORT)
    await _create_consumer_group()
    logger.info(
        "MIRROR_WORKER_STARTED stream=%s group=%s consumer=%s "
        "concurrency=%s matrix_model=%s litellm_env=%s metrics_port=%s",
        MIRROR_STREAM,
        MIRROR_CONSUMER_GROUP,
        CONSUMER_NAME,
        MIRROR_CONCURRENCY,
        MATRIX_MODEL,
        LITELLM_ENV,
        METRICS_PORT,
    )

    # XREADGROUP blocks briefly when idle, avoiding a tight polling loop. The
    # special ID ">" requests entries never delivered to this consumer group.
    while True:
        records = await redis.xreadgroup(
            groupname=MIRROR_CONSUMER_GROUP,
            consumername=CONSUMER_NAME,
            streams={MIRROR_STREAM: ">"},
            count=MIRROR_CONCURRENCY,
            block=READ_BLOCK_MS,
        )

        # Redis returns entries grouped by Stream. Only one Stream is requested,
        # so flatten the response and ignore the repeated Stream name.
        entries = [entry for _, stream_entries in records for entry in stream_entries]
        if entries:
            # Run up to MIRROR_CONCURRENCY Matrix calls in parallel.
            await asyncio.gather(
                *(
                    _handle_entry(entry_id, fields)
                    for entry_id, fields in entries
                )
            )


# Do not start the infinite loop when helpers are imported by tests or tooling.
if __name__ == "__main__":
    asyncio.run(main())

Helm templates та values

Файл mirror-worker.yaml – створює один Pod, в який через ConfigMap додає скрипт. Kubernetes Service – тільки для метрик, використовується у VMServiceScrape.

ConfigMap із Python-скриптом:

{{- if .Values.mirrorWorker.enabled }}
apiVersion: v1
kind: ConfigMap
metadata:
  name: {{ .Release.Name }}-mirror-worker
data:
  mirror_worker.py: |
{{ .Files.Get "files/mirror_worker.py" | indent 4 }}

Kubernetes Service для метрик – Service потрібен лише для збору Prometheus-метрик із Mirror Worker:

---
apiVersion: v1
kind: Service
metadata:
  name: {{ .Release.Name }}-mirror-worker
  labels:
    app.kubernetes.io/name: mirror-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  selector:
    app.kubernetes.io/name: mirror-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
  ports:
    - name: metrics
      port: {{ .Values.mirrorWorker.metricsPort }}
      targetPort: metrics

Deployment – використовується готовий LiteLLM image, в якому вже є потрібні Python-бібліотеки, монтує скрипт, передає параметри через змінні:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-mirror-worker
  labels:
    app.kubernetes.io/name: mirror-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: mirror-worker
      app.kubernetes.io/instance: {{ .Release.Name }}
  template:
    metadata:
      annotations:
        checksum/mirror-worker: {{ .Files.Get "files/mirror_worker.py" | sha256sum }}
      labels:
        app.kubernetes.io/name: mirror-worker
        app.kubernetes.io/instance: {{ .Release.Name }}
    spec:
      containers:
        - name: mirror-worker
          image: {{ .Values.mirrorWorker.image | quote }}
          imagePullPolicy: IfNotPresent
          command: ["python", "/app/mirror_worker.py"]
          env:
            - name: PYTHONUNBUFFERED
              value: "1"
            - name: MIRROR_REDIS_URL
              value: {{ .Values.mirrorWorker.redisUrl | quote }}
            - name: MIRROR_REDIS_STREAM
              value: {{ .Values.mirrorWorker.redisStream | quote }}
            - name: MIRROR_REDIS_CONSUMER_GROUP
              value: {{ .Values.mirrorWorker.consumerGroup | quote }}
            - name: MIRROR_CONCURRENCY
              value: {{ .Values.mirrorWorker.concurrency | quote }}
            - name: EVALUATION_REDIS_STREAM
              value: {{ .Values.evaluationWorker.redisStream | quote }}
            - name: EVALUATION_REDIS_STREAM_MAXLEN
              value: {{ .Values.mirrorWorker.evaluationStreamMaxLen | quote }}
            - name: TRAFFIC_MIRROR_URL
              value: {{ .Values.mirrorWorker.matrixUrl | quote }}
            - name: TRAFFIC_MIRROR_MODEL
              value: {{ .Values.mirrorWorker.matrixModel | quote }}
            - name: TRAFFIC_MIRROR_TIMEOUT_SECONDS
              value: {{ .Values.mirrorWorker.matrixTimeoutSeconds | quote }}
            - name: MIRROR_METRICS_PORT
              value: {{ .Values.mirrorWorker.metricsPort | quote }}
            - name: LITELLM_ENV
              value: {{ .Values.env | quote }}
            - name: OTEL_ENDPOINT
              value: {{ .Values.mirrorWorker.otelEndpoint | quote }}
            - name: OTEL_PROJECT_NAME
              value: {{ .Values.mirrorWorker.otelProjectName | quote }}
          ports:
            - name: metrics
              containerPort: {{ .Values.mirrorWorker.metricsPort }}
              protocol: TCP
          resources:
{{ toYaml .Values.mirrorWorker.resources | indent 12 }}
          volumeMounts:
            - name: worker-code
              mountPath: /app/mirror_worker.py
              subPath: mirror_worker.py
      volumes:
        - name: worker-code
          configMap:
            name: {{ .Release.Name }}-mirror-worker
      affinity:
{{ toYaml .Values.mirrorWorker.affinity | indent 8 }}
      tolerations:
{{ toYaml .Values.mirrorWorker.tolerations | indent 8 }}

VMServiceScrape знаходить Service за labels і забирає метрики з /metrics:

---
apiVersion: operator.victoriametrics.com/v1beta1
kind: VMServiceScrape
metadata:
  name: {{ .Release.Name }}-mirror-worker
spec:
  namespaceSelector:
    matchNames:
      - {{ .Release.Namespace }}
  selector:
    matchLabels:
      app.kubernetes.io/name: mirror-worker
      app.kubernetes.io/instance: {{ .Release.Name }}
  endpoints:
    - port: metrics
      path: /metrics
      interval: {{ .Values.mirrorWorker.scrapeInterval | quote }}
{{- end }}

Values:

# Worker consuming queued primary responses and calling Matrix asynchronously.
mirrorWorker:
  enabled: true
  image: ghcr.io/berriai/litellm-database:v1.100.1
  redisUrl: redis://litellm-evaluation-redis:6379/0
  redisStream: traffic_mirror:requests
  consumerGroup: traffic-mirror-workers
  concurrency: 4
  evaluationStreamMaxLen: 200
  matrixUrl: http://matrix.neoc.vpn.ops.example.co:31000/v1/chat/completions
  matrixModel: gemma-4-26b-a4b-it-q8
  matrixTimeoutSeconds: 120
  metricsPort: 9091
  scrapeInterval: 30s
  otelEndpoint: http://atlas-victoriametrics-vt-single-server.ops-monitoring-ns.svc.cluster.local:10428/insert/opentelemetry/v1/traces
  otelProjectName: litellm-test
  resources:
    requests:
      cpu: 50m
      memory: 128Mi
    limits:
      cpu: 250m
      memory: 256Mi
  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
          - matchExpressions:
              - key: component
                operator: In
                values:
                  - llm
  tolerations:
    - key: LLMOnly
      operator: Exists
      effect: NoSchedule

А значення для EVALUATION_REDIS_STREAM (куди записати результат виконання запитів до обох моделей) береться із .Values.evaluationWorker.redisStream (див. нижче в Evaluation Worker).

Evaluation Worker

Читає готові пари відповідей із Redis, передає їх до Judge-моделі та записує результат у логи й метрики.

Скрипт evaluation_worker.py

• постійно слухає Redis Stream traffic_mirror:evaluations через окрему Consumer Group
• отримує оригінальний prompt, primary response і Matrix response
• випадково розміщує відповіді як “Response A” і “Response B” не передаючи Judge назви моделей, аби це не вплинуло на оцінку
• напряму викликає Judge LLM через OpenAI API – робимо не через LiteLLM, щоб не запустити рекурсивний mirroring
• до Judge LLM передає промпт, в якому модель має спочатку сформувати пояснення свого рішенню, а потім виставити оцінки від 0.0 до 1.0 і вибрати “переможця” – primary (модель OpenAI), matrix (модель на нашому сервері), або tie (обидві відповіді рівні, переможця нема)
• при отриманні відповіді – переводить анонімні імена провайдерів і моделей A/B назад в реальні primary/matrix
• записує повний результат, включно з оцінками, winner і reason в логи
• на власному /metrics endpoint повертає метрики оцінок, переможців, тривалості та помилок, які забирає VMAgent

Весь скрипт:

"""Evaluate completed primary/Matrix response pairs with an OpenAI Judge.

Mirror Worker writes self-contained jobs to ``traffic_mirror:evaluations``.
This long-running consumer performs a blind randomized A/B comparison, maps the
anonymous result back to the real response sources, emits low-cardinality
Prometheus metrics, and logs the Judge's human-readable reason.

Successful jobs are ACKed only after all result side effects complete. Failed
jobs remain in the Redis consumer group's pending list for inspection or manual
recovery; this worker currently has no automatic pending-entry retry loop.
"""

import asyncio
import json
import logging
import os
import random
import socket
import time
from typing import Any

from openai import AsyncOpenAI
from prometheus_client import Counter, Histogram, start_http_server
from redis.asyncio import Redis
from redis.exceptions import ResponseError


# Redis connection and Stream name are configured in the environment values and
# injected by helm/templates/evaluation-worker.yaml. The Stream is populated by
# mirror_worker.py only after both the primary and Matrix responses are ready.
REDIS_URL = os.environ["EVALUATION_REDIS_URL"]
REDIS_STREAM = os.environ["EVALUATION_REDIS_STREAM"]

# The consumer group coordinates workers so each Stream entry is delivered to
# one evaluation-worker pod. Jobs are acknowledged only after their evaluation
# and metrics have been produced, leaving failed jobs pending for inspection.
REDIS_CONSUMER_GROUP = os.environ["EVALUATION_REDIS_CONSUMER_GROUP"]

# The Judge model is configured per environment and called directly through the
# OpenAI API. It must not be routed through LiteLLM, because every successful
# LiteLLM request starts traffic mirroring and would recursively enqueue jobs.
JUDGE_MODEL = os.environ["EVALUATION_JUDGE_MODEL"]

# Reasoning effort is optional because non-reasoning models such as GPT-4.1 do
# not accept this request parameter. An empty value makes _evaluate omit it;
# reasoning models can enable it through the environment values without a code
# change.
JUDGE_REASONING_EFFORT = os.environ.get("EVALUATION_JUDGE_REASONING_EFFORT", "")

# This limit caps the complete Judge output. For reasoning models it is shared
# by hidden reasoning and visible JSON; for GPT-4.1 it limits the visible result.
JUDGE_MAX_COMPLETION_TOKENS = int(
    os.environ["EVALUATION_JUDGE_MAX_COMPLETION_TOKENS"]
)

# The HTTP port is scraped by the VMServiceScrape created in the same Helm
# template. LITELLM_ENV separates ops and test series without high-cardinality
# request identifiers in Prometheus labels.
METRICS_PORT = int(os.environ["EVALUATION_METRICS_PORT"])
LITELLM_ENV = os.environ["LITELLM_ENV"]

# Kubernetes assigns each pod a unique hostname, which becomes the consumer
# name inside the shared Redis consumer group and keeps deliveries attributable
# to an individual worker instance.
CONSUMER_NAME = socket.gethostname()

# XREADGROUP blocks for five seconds when the Stream is empty. The finite block
# avoids a tight polling loop while still letting the coroutine regain control
# periodically for connection errors and process shutdown.
READ_BLOCK_MS = 5_000

# Use a timestamped single-line format because Kubernetes collects stdout and
# stderr. The named logger makes evaluation-worker records easy to distinguish
# when the container image includes logs from other Python libraries.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("evaluation-worker")

# Decode Redis responses to strings so the JSON payload can be passed directly
# to json.loads. This client is shared by group creation, reads, and ACKs for the
# lifetime of the worker process.
redis = Redis.from_url(REDIS_URL, decode_responses=True)

# OPENAI_JUDGE_KEY is a dedicated OpenAI project key from the litellm-secrets
# Kubernetes Secret. Keeping it separate from OPENAI_API_KEY makes Judge spend
# independently visible in OpenAI billing. This client bypasses LiteLLM to
# prevent recursive traffic mirrors.
openai = AsyncOpenAI(api_key=os.environ["OPENAI_JUDGE_KEY"])

# Record one score observation for each side of every completed comparison.
# These labels allow provider/model dashboards while deliberately excluding
# trace IDs, prompts, responses, and reasons to control metric cardinality.
EVALUATION_SCORE = Histogram(
    "litellm_evaluation_score",
    "Judge score assigned to an evaluated LLM response.",
    ["litellm_env", "provider", "model", "judge_model"],
    buckets=(0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0),
)

# Record one result per evaluated pair. The winner label contains primary,
# matrix, or tie after anonymous A/B positions are translated back to their
# actual sources by _evaluate.
EVALUATIONS_TOTAL = Counter(
    "litellm_evaluations_total",
    "Completed pairwise LLM response evaluations.",
    [
        "litellm_env",
        "winner",
        "primary_model",
        "matrix_model",
        "judge_model",
    ],
)

# Count processing failures by environment and Judge model. The stage label is
# intentionally coarse and currently fixed to processing so exception details
# stay in logs instead of becoming unbounded Prometheus label values.
EVALUATION_ERRORS_TOTAL = Counter(
    "litellm_evaluation_errors_total",
    "Failed LLM response evaluations.",
    ["litellm_env", "stage", "judge_model"],
)

# Measure wall-clock time from decoding a Stream entry until the Judge result
# has been converted into metrics. Redis blocking time is excluded because the
# timer starts only after a job has been delivered.
EVALUATION_DURATION_SECONDS = Histogram(
    "litellm_evaluation_duration_seconds",
    "Time spent evaluating a response pair with the Judge model.",
    ["litellm_env", "judge_model"],
)

# Pre-create the processing-error series so dashboards can show zero immediately
# after startup instead of treating the metric as absent until the first error.
EVALUATION_ERRORS_TOTAL.labels(
    litellm_env=LITELLM_ENV,
    stage="processing",
    judge_model=JUDGE_MODEL,
).inc(0)

# Define the exact structured output accepted from the Judge. Strict mode and
# additionalProperties=false prevent prose or unknown fields from being treated
# as a valid evaluation, while numeric bounds keep Histogram observations sane.
EVALUATION_SCHEMA = {
    "type": "object",
    "properties": {
        "reason": {"type": "string"},
        "score_a": {"type": "number", "minimum": 0, "maximum": 1},
        "score_b": {"type": "number", "minimum": 0, "maximum": 1},
        "winner": {"type": "string", "enum": ["a", "b", "tie"]},
    },
    "required": ["reason", "score_a", "score_b", "winner"],
    "additionalProperties": False,
}

# Ask for evidence before numeric scores and the winner. This ordering reduces
# post-hoc justification, while anonymous response names prevent known provider
# or model identities from influencing the comparison.
SYSTEM_PROMPT = """You are an impartial evaluator of two LLM responses.
Evaluate how well each response answers the user's request. Consider factual
correctness, relevance, instruction following, clarity, and usefulness.

Do not infer which model produced either response. First compare both responses
against the evaluation criteria and write a concise, evidence-based reason that
does not refer to scores or a winner. Only after that analysis, score each
response from 0.0 to 1.0 and select the winner. Use winner=\"tie\" only when
their quality is effectively equal. Return only the requested structured
result, in the schema-defined field order."""


# Build the user message sent to the Judge from the original request and the two
# randomized responses. json.dumps preserves structured prompts and Unicode;
# the explicit section markers keep the request separate from model outputs.
def _format_prompt(prompt: Any, response_a: str, response_b: str) -> str:
    return (
        f"USER REQUEST:\n{json.dumps(prompt, ensure_ascii=False)}\n\n"
        f"RESPONSE A:\n{response_a}\n\n"
        f"RESPONSE B:\n{response_b}"
    )


# Randomize and evaluate one completed mirror job. The input shape is produced
# by mirror_worker.py and contains trace metadata plus primary and Matrix result
# objects. The returned dictionary uses real source names again so main can emit
# metrics without knowing which source occupied anonymous position A or B.
async def _evaluate(job: dict[str, Any]) -> dict[str, Any]:
    # Keep direct references to both response objects because their text is sent
    # to the Judge and their provider/model metadata is returned for metrics.
    primary = job["primary"]
    matrix = job["matrix"]

    # Assign primary and Matrix independently to A or B for every job to reduce
    # position bias. Only response text is passed onward; provider and model names
    # remain outside the Judge prompt to preserve the blind comparison.
    if random.random() < 0.5:
        response_a = primary
        response_b = matrix
        positions = {"a": "primary", "b": "matrix"}
    else:
        response_a = matrix
        response_b = primary
        positions = {"a": "matrix", "b": "primary"}

    # Use one request-options dictionary so optional model capabilities can be
    # included conditionally. Structured output guarantees parseable scores and
    # allows a malformed or refused response to follow the normal retry path.
    request_options: dict[str, Any] = {
        "model": JUDGE_MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": _format_prompt(
                    job["prompt"],
                    response_a["response"],
                    response_b["response"],
                ),
            },
        ],
        "response_format": {
            "type": "json_schema",
            "json_schema": {
                "name": "llm_response_evaluation",
                "strict": True,
                "schema": EVALUATION_SCHEMA,
            },
        },
        "max_completion_tokens": JUDGE_MAX_COMPLETION_TOKENS,
    }
    # Send reasoning_effort only when configured. OpenAI rejects this parameter
    # for models such as GPT-4.1, while reasoning-capable models may require it.
    if JUDGE_REASONING_EFFORT:
        request_options["reasoning_effort"] = JUDGE_REASONING_EFFORT

    # Await the direct OpenAI Chat Completions call so each worker processes one
    # Stream job at a time and acknowledges it only after the response succeeds.
    response = await openai.chat.completions.create(**request_options)

    # A successful HTTP response may still contain no assistant content, for
    # example after a refusal. Treat it as a failed evaluation and leave the job
    # pending rather than recording incomplete or invented scores.
    content = response.choices[0].message.content
    if content is None:
        raise ValueError("Judge returned an empty response")

    # Parse the schema-constrained JSON. The API validates its shape; json.loads
    # converts it to the local dictionary used for source-position translation.
    result = json.loads(content)

    # Translate anonymous A/B scores back to primary and Matrix so metric labels
    # describe the real response sources regardless of randomized ordering.
    scores = {
        positions["a"]: result["score_a"],
        positions["b"]: result["score_b"],
    }
    # Preserve tie as a source-independent result; otherwise map the winning
    # anonymous position back to primary or matrix for the counter label.
    winner = (
        "tie"
        if result["winner"] == "tie"
        else positions[result["winner"]]
    )

    # Return the normalized result consumed by main for metrics and structured
    # completion logs. The reason and trace ID stay in logs, never metric labels.
    return {
        "trace_id": job["trace_id"],
        "judge_model": JUDGE_MODEL,
        "primary_provider": primary.get("provider", "unknown"),
        "primary_model": primary["model"],
        "primary_score": scores["primary"],
        "matrix_model": matrix["model"],
        "matrix_score": scores["matrix"],
        "winner": winner,
        "reason": result["reason"],
    }


# Create the Redis Stream and consumer group during worker startup. mkstream
# handles a fresh ephemeral Redis instance; BUSYGROUP means another pod or a
# previous run already created the group and is therefore safe to ignore.
async def _create_consumer_group() -> None:
    try:
        await redis.xgroup_create(
            REDIS_STREAM,
            REDIS_CONSUMER_GROUP,
            id="0",
            mkstream=True,
        )
        logger.info(
            "EVALUATION_CONSUMER_GROUP_CREATED stream=%s group=%s",
            REDIS_STREAM,
            REDIS_CONSUMER_GROUP,
        )
    except ResponseError as error:
        if "BUSYGROUP" not in str(error):
            raise


# Start the metrics endpoint, initialize Redis coordination, and continuously
# process new Stream entries. An entry is acknowledged only after Judge output,
# metrics, and the completion log succeed; any exception increments the bounded
# error series and leaves the entry pending for inspection or later recovery.
async def main() -> None:
    # prometheus_client runs the scrape endpoint in a background thread, leaving
    # this asyncio event loop dedicated to Redis and OpenAI network operations.
    start_http_server(METRICS_PORT)

    # Group creation must finish before XREADGROUP can consume entries. Startup
    # fails on Redis errors other than BUSYGROUP so Kubernetes can restart the pod.
    await _create_consumer_group()
    logger.info(
        "EVALUATION_WORKER_STARTED stream=%s group=%s consumer=%s "
        "judge=%s reasoning_effort=%s max_completion_tokens=%s "
        "litellm_env=%s metrics_port=%s",
        REDIS_STREAM,
        REDIS_CONSUMER_GROUP,
        CONSUMER_NAME,
        JUDGE_MODEL,
        JUDGE_REASONING_EFFORT,
        JUDGE_MAX_COMPLETION_TOKENS,
        LITELLM_ENV,
        METRICS_PORT,
    )

    # The worker is a long-running Kubernetes process. Redis blocks briefly while
    # idle, then returns zero or more Stream entries assigned to this consumer.
    while True:
        records = await redis.xreadgroup(
            groupname=REDIS_CONSUMER_GROUP,
            consumername=CONSUMER_NAME,
            streams={REDIS_STREAM: ">"},
            count=1,
            block=READ_BLOCK_MS,
        )

        # Redis returns a list per requested Stream. Only one Stream is requested
        # today, so its name is intentionally ignored and each entry is processed.
        for _, entries in records:
            for entry_id, fields in entries:
                # Start timing after Redis delivery so queue wait time does not
                # inflate the duration of the actual Judge evaluation.
                started_at = time.perf_counter()
                try:
                    # mirror_worker.py stores the complete job as JSON under the
                    # payload field; _evaluate performs blind A/B normalization.
                    job = json.loads(fields["payload"])
                    result = await _evaluate(job)

                    # Emit one score for the primary response, one for Matrix, one
                    # pair-level winner, and one end-to-end processing duration.
                    EVALUATION_SCORE.labels(
                        litellm_env=LITELLM_ENV,
                        provider=result["primary_provider"],
                        model=result["primary_model"],
                        judge_model=JUDGE_MODEL,
                    ).observe(result["primary_score"])
                    EVALUATION_SCORE.labels(
                        litellm_env=LITELLM_ENV,
                        provider="matrix",
                        model=result["matrix_model"],
                        judge_model=JUDGE_MODEL,
                    ).observe(result["matrix_score"])
                    EVALUATIONS_TOTAL.labels(
                        litellm_env=LITELLM_ENV,
                        winner=result["winner"],
                        primary_model=result["primary_model"],
                        matrix_model=result["matrix_model"],
                        judge_model=JUDGE_MODEL,
                    ).inc()
                    EVALUATION_DURATION_SECONDS.labels(
                        litellm_env=LITELLM_ENV,
                        judge_model=JUDGE_MODEL,
                    ).observe(time.perf_counter() - started_at)
                    logger.info(
                        "EVALUATION_COMPLETED entry_id=%s result=%s",
                        entry_id,
                        json.dumps(result, ensure_ascii=False),
                    )
                    # ACK only after all success-side effects complete. If anything
                    # above raises, Redis keeps the entry in the group's pending
                    # list rather than silently losing an unevaluated response pair.
                    await redis.xack(
                        REDIS_STREAM,
                        REDIS_CONSUMER_GROUP,
                        entry_id,
                    )
                # Keep the worker alive after malformed jobs, OpenAI failures, or
                # metric errors. Details go to the traceback log; the Prometheus
                # counter remains low-cardinality and suitable for alerting.
                except Exception:
                    EVALUATION_ERRORS_TOTAL.labels(
                        litellm_env=LITELLM_ENV,
                        stage="processing",
                        judge_model=JUDGE_MODEL,
                    ).inc()
                    logger.exception("EVALUATION_FAILED entry_id=%s", entry_id)


# Run the async worker only when this mounted file is executed as the container
# entrypoint. This guard also allows importing helpers in tests without starting
# the infinite Redis consumption loop.
if __name__ == "__main__":
    asyncio.run(main())

Helm templates та values

Тут ConfigMap зі скриптом, Kubernetes Service для метрик, Deployment та VMServiceScrape.

Вже запишу все одним файлом, як він і є:

{{- if .Values.evaluationWorker.enabled }}
apiVersion: v1
kind: ConfigMap
metadata:
  name: {{ .Release.Name }}-evaluation-worker
data:
  evaluation_worker.py: |
{{ .Files.Get "files/evaluation_worker.py" | indent 4 }}
---
apiVersion: v1
kind: Service
metadata:
  name: {{ .Release.Name }}-evaluation-worker
  labels:
    app.kubernetes.io/name: evaluation-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  selector:
    app.kubernetes.io/name: evaluation-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
  ports:
    - name: metrics
      port: {{ .Values.evaluationWorker.metricsPort }}
      targetPort: metrics
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-evaluation-worker
  labels:
    app.kubernetes.io/name: evaluation-worker
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: evaluation-worker
      app.kubernetes.io/instance: {{ .Release.Name }}
  template:
    metadata:
      annotations:
        checksum/evaluation-worker: {{ .Files.Get "files/evaluation_worker.py" | sha256sum }}
      labels:
        app.kubernetes.io/name: evaluation-worker
        app.kubernetes.io/instance: {{ .Release.Name }}
    spec:
      containers:
        - name: evaluation-worker
          image: {{ .Values.evaluationWorker.image | quote }}
          imagePullPolicy: IfNotPresent
          command: ["python", "/app/evaluation_worker.py"]
          env:
            - name: PYTHONUNBUFFERED
              value: "1"
            - name: EVALUATION_REDIS_URL
              value: {{ .Values.evaluationWorker.redisUrl | quote }}
            - name: EVALUATION_REDIS_STREAM
              value: {{ .Values.evaluationWorker.redisStream | quote }}
            - name: EVALUATION_REDIS_CONSUMER_GROUP
              value: {{ .Values.evaluationWorker.consumerGroup | quote }}
            - name: EVALUATION_JUDGE_MODEL
              value: {{ .Values.evaluationWorker.judgeModel | quote }}
            - name: EVALUATION_JUDGE_REASONING_EFFORT
              value: {{ .Values.evaluationWorker.judgeReasoningEffort | quote }}
            - name: EVALUATION_JUDGE_MAX_COMPLETION_TOKENS
              value: {{ .Values.evaluationWorker.judgeMaxCompletionTokens | quote }}
            - name: EVALUATION_METRICS_PORT
              value: {{ .Values.evaluationWorker.metricsPort | quote }}
            - name: LITELLM_ENV
              value: {{ .Values.env | quote }}
            - name: OPENAI_JUDGE_KEY
              valueFrom:
                secretKeyRef:
                  name: litellm-secrets
                  key: OPENAI_JUDGE_KEY
          ports:
            - name: metrics
              containerPort: {{ .Values.evaluationWorker.metricsPort }}
              protocol: TCP
          resources:
{{ toYaml .Values.evaluationWorker.resources | indent 12 }}
          volumeMounts:
            - name: worker-code
              mountPath: /app/evaluation_worker.py
              subPath: evaluation_worker.py
      volumes:
        - name: worker-code
          configMap:
            name: {{ .Release.Name }}-evaluation-worker
      affinity:
{{ toYaml .Values.evaluationWorker.affinity | indent 8 }}
      tolerations:
{{ toYaml .Values.evaluationWorker.tolerations | indent 8 }}
---
apiVersion: operator.victoriametrics.com/v1beta1
kind: VMServiceScrape
metadata:
  name: {{ .Release.Name }}-evaluation-worker
spec:
  namespaceSelector:
    matchNames:
      - {{ .Release.Namespace }}
  selector:
    matchLabels:
      app.kubernetes.io/name: evaluation-worker
      app.kubernetes.io/instance: {{ .Release.Name }}
  endpoints:
    - port: metrics
      path: /metrics
      interval: {{ .Values.evaluationWorker.scrapeInterval | quote }}
{{- end }}

Values – тут вже 5.6 Luna, хоча починав робити на 5.6 Terra – але це по грошам вийшло дико 🙂 Далі про це коротко допишу.

Значення для Evaluation Worker:

evaluationWorker:
  enabled: true
  image: ghcr.io/berriai/litellm-database:v1.100.1

  redisUrl: redis://litellm-evaluation-redis:6379/0
  redisStream: traffic_mirror:evaluations
  consumerGroup: evaluation-workers

  judgeModel: gpt-5.6-luna
  judgeReasoningEffort: low
  judgeMaxCompletionTokens: 4000

  metricsPort: 9090
  scrapeInterval: 30s

  resources:
    requests:
      cpu: 50m
      memory: 128Mi
    limits:
      cpu: 250m
      memory: 256Mi

  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
          - matchExpressions:
              - key: component
                operator: In
                values:
                  - llm

  tolerations:
    - key: LLMOnly
      operator: Exists
      effect: NoSchedule

Значення для OPENAI_JUDGE_KEY отримується із AWS Secrets Manager з External Secrets Operator для AWS Secrets Manager : це окремий API Key в OpenAI, аби мати можливість відрізнити витрати самого LiteLLM і його клієнтів від витрат на LLM Evaluations.

Результат роботи та Grafana dashboard

Тут вже теж не буду детально описувати всю дашборду – просто для прикладу, як це зараз виглядає у нас:

Нюанси, оптимізація вартості та подальші плани

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

Traffic Mirroring окремо від LiteLLM

У першій версії запит до Matrix виконувався безпосередньо з LiteLLM Custom Callback.

Це виявилось проблемою – бо відповідь self-hosted моделі може займати кілька секунд або навіть хвилин, а внутрішній LiteLLM LoggingWorker дропав callback tasks, які виконувались довше двох секунд.

Тому зараз запит до Matrix винес в окремий Mirror Worker:

• Callback виконує тільки короткий запис job у Redis
• Mirror Worker виконує повільний запит до Matrix незалежно від LiteLLM
• недоступність Matrix не впливає на primary request та відповідь клієнту

Callback також має Redis timeout 500 мілісекунд і працює за принципом fail-open: якщо Redis недоступний – то mirroring буде втрачено, але клієнт усе одно отримає primary response.

Chat Completions API та Responses API

Постійний біль з форматами…

Наші клієнти використовують обидва OpenAI API, а Matrix з llama.cpp приймає Chat Completions API – тому Responses API input не можна передати туди без змін. Особливо це стосується:

• structured content та input_text
• developer messages
• function_call
• function_call_output
• tool calls та tool results

Через це в TrafficMirrorCallback додана нормалізація input до звичайного списку messages з ролями system, user та assistant.

Крім того, стандартний LiteLLM OTel exporter не додавав response від Responses API до primary span. Для таких запитів Callback створює окремий response_capture span у тому самому trace.

Оптимізація вартості Judge LLM

Кожен evaluation створює окремий платний запит до Judge LLM, причому в його input входять:

• оригінальний prompt клієнта
• повна відповідь primary-моделі
• повна відповідь Matrix-моделі

Для довгих запитів саме input tokens можуть складати основну частину вартості evaluation.

В якийсь момент на 5.6 Terra у нас за день було 113 долларів витрат загалом на весь проект і клієнтів, але з цих 113 витратили 90 чисто на Evaluation Worker.

Для контролю витрат маємо кілька механізмів:

• TRAFFIC_MIRROR_SAMPLE_RATE дозволяє оцінювати лише частину production-запитів
• у production зараз mirror виконується для 5% успішних запитів
• Judge був змінений з дорожчої GPT 5.6 Terra на GPT 5.6 Luna
• для Judge використовується reasoning_effort=low
• Judge має окремий OPENAI_JUDGE_KEY, тому його витрати можна бачити окремо від клієнтського трафіку LiteLLM
• max_completion_tokens обмежує максимальний бюджет відповіді Judge

Що можна буде зробити ще:

• зменшити judgeMaxCompletionTokens: обмежуємо кількість reasoning та output tokens у відповіді Judge
• обрізати великі prompts і responses: зменшуємо кількість input tokens (але тут треба симетрично зберігати початок і кінець тексту)
• використати OpenAI Prompt Caching: повторно використовуємо стабільний префікс із Judge instructions і зменшуємо вартість cached input
• використати Batch API: накопичуємо evaluation jobs і обробляємо їх асинхронно зі знижкою, якщо допустима затримка результатів
• використати Flex Processing: отримуємо нижчу вартість в обмін на більшу latency та можливу тимчасову недоступність ресурсів – для задачі LLM Evaluation це не критично

Неупередженість Judge

Коли показував мій код нашим девелоперам, то вказали один цікавий момент: в system prompt до Judge LLM та шаблоні його відповіді спершу йшли “score_a“, “score_b” – а потім “reasoning“.

Це могло привести до того, що LLM спочатку ставить оцінку – а потім під вже визначені оцінки “підтягує” обґрунтування.

Тому зараз робиться навпаки: спочатку LLM “думає”, а вже по результату виставляє оцінки.

Дуже неочевидний і цікавий нюанс.

Ну вже те, що вище наче описував – але зібране списком:

• відповіді моделей перед передачею до “судді” випадково розміщуються як Response A та Response B
• назви моделей і провайдерів до Judge LLM не передаються
• Judge LLM спочатку має сформувати evidence-based reason
• лише після пояснення Judge виставляє scores та вибирає winner
• відповідь перевіряється через strict JSON Schema

Це не робить оцінку абсолютно об’єктивною – LLM Judge теж може мати власні упередження, нестабільність і переваги щодо стилю відповіді. Власне все те, за що ми “любимо” LLM 🙂

Performance

На відміну від першого варіанту з Mirroring script і задачі самого Evaluation Worker – повільні частини системи винесені з процесу LiteLLM у фонові workers.

Mirror Worker може виконувати кілька Matrix requests паралельно через MIRROR_CONCURRENCY – бо ми впирались в те, що воркер не встигав обробити все.

При цьому ще є питання до конфігурації самого llama.cpp:

• реальної пропускної здатності Matrix
• кількості одночасних inference requests, які підтримує llama.cpp
• розміру prompts
• довжини відповідей
• доступної GPU memory

Але це вже окрема тема.

Matrix request має timeout 120 секунд. Timeout та HTTP errors записуються окремо в логи й метрики.

Redis та надійність

Redis використовується як тимчасова черга, а не як постійне сховище результатів.

Поточний Redis:

• працює в одному pod
• не має authentication
• не має persistence
• не має replication або HA
• використовує emptyDir
• має maxmemory 96 MB та політику noeviction

Це свідомо прийняте обмеження для поточного рішення: після перезапуску поду з Redis його jobs можуть бути втрачені – але вже записані історичні метрики залишаться у VictoriaMetrics, а трейси у VictoriaTraces – тому не критично.

Власне, на цьому все.

Зараз додатково готуємо LLM Evaluations з Opik і потроху тестуємо різні моделі.