A quick recap of what we are doing on the project right now: we have a separate hardware server (hostname == “Matrix”), where we run our own self-hosted models with llama.cpp.

In our Kubernetes cluster we have a LiteLLM AI Gateway for our clients – Backend API and other project services.

Clients send their OpenAI/Anthropic/OpenRouter requests to LiteLLM, and LiteLLM forwards them to a provider, logs everything, records traces, controls access and $$$.

The task now consists of two parts:

• we need to configure traffic mirroring: clients continue working with a primary provider/model such as OpenAI, while “under the hood” LiteLLM additionally sends the client’s request to our self-hosted LLM (currently Gemma)
• then we need to perform LLM Evaluations: after receiving responses from OpenAI and Gemma running on our Matrix – we need to evaluate the quality of both responses

Then, having scores for responses from a “primary model” such as OpenAI GPT 5.6 Terra and our self-hosted model, for example Gemma, we will be able to compare their quality and choose something suitable for ourselves.

And later, perhaps, we will even get into some fine-tuning.

I described the traffic mirroring setup in the post LiteLLM: Custom Callback for traffic mirroring and OTel tracing to VictoriaTraces – it will be used in this post too, but with some changes, because that implementation was more of a PoC and had a few bugs.

Why I implemented it specifically through a custom callback is covered in LiteLLM: Traffic Mirroring and Batch Completions and traffic to two providers.

The solution described in this post works, but it is still more of an example of how such a task can be solved.

Some nuances appeared already during actual use – I’ll mention them at the end.

The code here is what we currently have in production, although there will most likely be more changes.

And I may have missed something – usually I write posts while I’m actually investigating/configuring things and record everything along the way, but this time I’m describing an already finished project, so I could have forgotten some details.

The post looks long – but mostly because of the code, as I’ve included all scripts in full.

LLM Evaluations with Phoenix or Opik

In general, LLM response evaluation can be done with Phoenix – see Running Evals on Traces – or Opik – see Evaluation Overview, and we are testing both as well (although we will probably use Opik).

But for now the task was to build something simple and quick to implement, which would also expose the results as ordinary Prometheus metrics and a Grafana dashboard.

For Opik we are currently preparing a “golden dataset” – data against which responses from our self-hosted models will be compared, and this is a separate pain in the ass because even production responses are not always correct.

In the solution described in this post, the primary model’s response is used as a baseline – the Judge LLM compares the self-hosted model’s response against it.

Basically, this is not a full golden dataset because it does not contain pre-validated reference responses and evaluation criteria – but we treat the GPT 5.6 response as the baseline our self-hosted LLMs should aim for.

General architecture

In short, we have three components:

• LiteLLM:
  • receives a request from a client and sends it to the primary model (OpenAI)
  • then calls our Custom Callback script, which writes the client’s request data to Redis
• Mirror Worker:
  • reads data from Redis and performs a request to Matrix using the client’s request
  • writes the received result back to Redis into another stream
• Evaluation Worker:
  • reads data from Redis and sends the client’s request text and both responses to the “Judge Model”
  • Judge Model evaluates the quality of the responses and returns a score for each one
  • Evaluation Worker exposes the result as the litellm_evaluation_score metric with provider and model labels for VictoriaMetrics

Schematically, it can look like this – a more detailed description of the whole flow follows below:

And step by step:

• LiteLLM – receives requests from clients (our internal services):
  • proxies these requests to the primary provider/model (OpenAI / GPT 5.6)
  • has its own OpenTelemetry context, creates a trace for this request – primary span, child spans, and writes them to VictoriaTraces (see LiteLLM: metrics, traces and integration with VictoriaMetrics Stack)
  • calls our TrafficMirrorCallback for every successful request – a script running inside the LiteLLM Pod
• Custom Callback – performs Traffic Mirroring:
  • gets the prompt and the primary model response
  • normalizes Chat Completions or Responses API input
  • adds the current OpenTelemetry context from LiteLLM (so we can have one shared trace)
  • writes a “mirror job” to the Redis Stream named traffic_mirror:requests
  • the Callback does not call Matrix at all – it only writes a job to Redis, so availability and performance of the self-hosted LLM do not affect client requests or LiteLLM itself
    • this is the main difference from the version described in the previous post
  • the Callback exposes the litellm_traffic_mirror_enqueue_total metric for VictoriaMetrics – the number of jobs added to Redis for Mirror Worker
• Redis:
  • Redis is used as a temporary queue between LiteLLM and background workers, so persistence and all sorts of fault tolerance are not required – the setup is intentionally as simple as possible
  • keeps two Redis Streams for workers – traffic_mirror:requests for Mirror Worker and traffic_mirror:evaluations for Evaluation Worker
• Mirror Worker:
  • constantly listens to the Redis Stream
  • reads jobs from traffic_mirror:requests through a Redis Consumer Group
  • sends the client’s request text to Matrix and Gemma
  • together with the rest of the data, receives the OTel context created by LiteLLM from Redis, so primary and mirror spans have the same trace_id
    • creates a separate mirror span with name="traffic_mirror <model_name>", whose attributes contain the model, provider name as “llama.cpp“, the original client prompt, and the response text from the Matrix model
    • as a result, VictoriaTraces contains the complete trace – the primary span from LiteLLM + Matrix span from Mirror Worker
  • after receiving the response:
    • creates an OpenTelemetry span containing the Matrix request and response
    • writes the result to VictoriaLogs
    • builds a pair consisting of the original client prompt + primary response (from the OpenAI model) + response from our Gemma (or whatever it will be) running on our “Matrix” server
    • writes it to the Redis Stream traffic_mirror:evaluations
  • generates the litellm_traffic_mirror_requests_total metric with the “outcome” label and values success, failed or timeout
• Evaluation Worker:
  • reads completed response pairs from the Redis Stream traffic_mirror:evaluations
  • builds a request to the Judge LLM (GPT 5.6 Terra in the examples here, although there are some nuances – more about them at the end) and sends it to the “judge”
  • Judge LLM:
    • compares both responses and returns to Evaluation Worker an explanation of its decision (reasoning), the primary response score, the Matrix response score, and the winner: primary, matrix or tie (tie means both responses have practically the same quality, so there is no clear winner)
  • Evaluation Worker writes the result received from Judge LLM to logs and generates Prometheus metrics:
    • litellm_evaluation_score, litellm_evaluation_errors_total, litellm_evaluation_duration_seconds
    • VictoriaMetrics has a VMServiceScrape and collects these metrics from the Evaluation Worker endpoint

Implementation – Python and Helm

Initially I started writing a step-by-step implementation – but it became way too much text, so here I’ll just show the final result (as of now) with explanations of the main points.

I also thought about describing the main functions here – but that turned into another wall of text, so I simply asked Codex to add detailed comments directly to the scripts.

And the code and Helm were about 90% written with Codex and 5.6 Sol Lite – but the code was reviewed and tested.

So, the Traffic Mirroring and LLM Evaluations system consists of three components, Python scripts:

• TrafficMirrorCallback (the traffic_mirror_callback.py script): stores information about the request received from the client in Redis
• Mirror Worker (the mirror_worker.py script): gets data from Redis, sends a request to Matrix, receives the response from the self-hosted LLM, and writes the result back to Redis for Evaluation Worker
• Evaluation Worker (the evaluation_worker.py script): reads the client’s request text and responses from both models from Redis, sends them to the Judge model for evaluation, and exposes the evaluation result as metrics for VictoriaMetrics

LiteLLM is deployed using a Helm chart (see LiteLLM: AI Gateway in Kubernetes and metrics to VictoriaMetrics), so Traffic Mirroring and LLM Evaluation are deployed through the same chart:

• evaluation-redis.yaml: manifest with Kubernetes Deployment and Service for Redis
• mirror-worker.yaml: Kubernetes Deployment, ConfigMap, Service and VMServiceScrape for Mirror Worker
• evaluation-worker.yaml: Kubernetes Deployment, ConfigMap, Service and VMServiceScrape for Evaluation Worker

Deploying Redis

It is deployed with a single evaluation-redis.yaml manifest containing all resources:

{{- 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

An important difference from what was described in the previous post: now the Callback only writes to Redis – it no longer sends a request to Matrix as the previous implementation did.

The reason is LiteLLM and its LoggingWorker, which was dropping callback tasks that took longer than two seconds.

Also, the previous version had a bug because the data format returned by Responses API is, unsurprisingly, different from Chat Completions API, which we use for requests to llama.cpp – so request normalization into a format understood by Chat Completions API has now been added.

The traffic_mirror_callback.py script

It is called by LiteLLM after a successful response from the primary model:

• has sampling through TRAFFIC_MIRROR_SAMPLE_RATE: we can configure what portion of requests should be forwarded to Matrix – reducing load on Matrix and spending on Judge LLM
• gets the original prompt, primary response, model, provider and response ID:
  • normalizes Chat Completions and Responses API input into the messages format supported by the Matrix/llama.cpp Chat Completions API
• stores the current OTel context in trace_context, so Mirror Worker can continue the same trace
• writes a job to the Redis Stream traffic_mirror:requests
• through LiteLLM’s own /metrics endpoint (because the callback runs inside the LiteLLM process), exposes the litellm_traffic_mirror_enqueue_total metric with the result enqueued, skipped, failed or cancelled; VMAgent collects this metric
• works according to the fail-open principle: problems with mirroring do not affect LiteLLM’s response to the client

The complete script:

"""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 `` 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 and values

The traffic_mirror_callback.py script is started by LiteLLM itself and added through callbacks in litellm_settings:

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

During deployment, the script itself is written to a 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 }}

Which is then mounted into 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
...

The chart values define variables for connecting to 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

Its job is to send requests to our self-hosted model and save its response, which is then used by Evaluation Worker.

The mirror_worker.py script

• constantly listens to the Redis Stream traffic_mirror:requests through a Redis Consumer Group
  • gets jobs written by TrafficMirrorCallback
  • restores the OTel context from trace_context
• sends the request to Matrix/llama.cpp and Gemma
• creates a traffic_mirror <model_name> span in the same trace as the primary request
• writes the prompt, primary response, Matrix response, reasoning, model, response ID and finish_reason into this span’s attributes
• builds a payload containing the prompt and responses from both models and writes it to the Redis Stream traffic_mirror:evaluations
• processes several requests in parallel according to MIRROR_CONCURRENCY
• acknowledges processed Redis jobs through XACK
• on its own /metrics endpoint exposes litellm_traffic_mirror_requests_total with the result success, failed or timeout, which is collected by VMAgent
• writes error details, including HTTP status and response body from Matrix, to logs

The complete script:

"""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 and values

The mirror-worker.yaml file creates one Pod, with the script added through a ConfigMap. The Kubernetes Service is used only for metrics and is referenced by VMServiceScrape.

ConfigMap with the Python script:

{{- 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 for metrics – the Service is needed only to collect Prometheus metrics from 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 – uses the existing LiteLLM image, which already contains the required Python libraries, mounts the script, and passes parameters through environment variables:

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 finds the Service by labels and collects metrics from /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

And the value for EVALUATION_REDIS_STREAM (where the result of requests to both models is written) comes from .Values.evaluationWorker.redisStream (see Evaluation Worker below).

Evaluation Worker

Reads completed response pairs from Redis, sends them to the Judge model and writes the result to logs and metrics.

The evaluation_worker.py script

• constantly listens to the Redis Stream traffic_mirror:evaluations through a separate Consumer Group
• gets the original prompt, primary response and Matrix response
• randomly assigns the responses as “Response A” and “Response B” without passing model names to the Judge, so they do not influence the evaluation
• calls the Judge LLM directly through the OpenAI API – we do not go through LiteLLM, otherwise recursive mirroring would start
• sends the Judge LLM a prompt in which the model must first produce an evidence-based explanation of its decision, then assign scores from 0.0 to 1.0 and select the “winner” – primary (the OpenAI model), matrix (the model running on our server), or tie (both responses are equal and there is no winner)
• after receiving the response, translates the anonymous A/B provider and model positions back to the actual primary/matrix sources
• writes the full result, including scores, winner and reason, to logs
• on its own /metrics endpoint exposes evaluation scores, winners, duration and error metrics, which are collected by VMAgent

The complete script:

"""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 and values

Here we have a ConfigMap with the script, a Kubernetes Service for metrics, Deployment and VMServiceScrape.

I’ll just put the whole file here as it is:

{{- 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 – here we already use 5.6 Luna, although I started with 5.6 Terra – but that turned out pretty wild in terms of money 🙂 I’ll briefly cover that below.

Values for 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

The value for OPENAI_JUDGE_KEY is obtained from AWS Secrets Manager using External Secrets Operator for AWS Secrets Manager: it is a separate API Key in OpenAI, so we can distinguish LiteLLM and client spending from LLM Evaluation spending.

Results and Grafana dashboard

I won’t describe the entire dashboard in detail here either – just an example of what it currently looks like for us:

Nuances, cost optimization and future plans

The system described above is already working – but during its development and testing we ran into several important nuances worth mentioning.

Traffic Mirroring outside LiteLLM

In the first version, the request to Matrix was performed directly from the LiteLLM Custom Callback.

This turned out to be a problem because a response from the self-hosted model can take several seconds or even minutes, while LiteLLM’s internal LoggingWorker was dropping callback tasks that ran for more than two seconds.

So now the Matrix request has been moved to a separate Mirror Worker:

• the Callback only performs a short operation that writes a job to Redis
• Mirror Worker performs the slow request to Matrix independently of LiteLLM
• Matrix availability does not affect the primary request or the response to the client

The Callback also has a Redis timeout of 500 milliseconds and works according to the fail-open principle: if Redis is unavailable, the mirroring job will be lost, but the client will still receive the primary response.

Chat Completions API and Responses API

The eternal pain of data formats…

Our clients use both OpenAI APIs, while Matrix with llama.cpp accepts Chat Completions API – so Responses API input cannot simply be passed there unchanged. This especially applies to:

• structured content and input_text
• developer messages
• function_call
• function_call_output
• tool calls and tool results

Because of this, TrafficMirrorCallback now normalizes input into an ordinary list of messages with system, user and assistant roles.

Also, the standard LiteLLM OTel exporter did not add Responses API responses to the primary span. For such requests, the Callback creates a separate response_capture span within the same trace.

Judge LLM cost optimization

Every evaluation creates a separate paid request to the Judge LLM, and its input contains:

• the original client prompt
• the complete primary-model response
• the complete Matrix-model response

For long requests, input tokens alone can make up the largest part of the evaluation cost.

At one point, while using 5.6 Terra, we had $113 in total project and client spending in a single day, and out of those $113, $90 was spent purely by Evaluation Worker.

We have several mechanisms to control the cost:

• TRAFFIC_MIRROR_SAMPLE_RATE allows us to evaluate only a portion of production requests
• in production, mirroring is currently performed for 5% of successful requests
• the Judge was changed from the more expensive GPT 5.6 Terra to GPT 5.6 Luna
• the Judge uses reasoning_effort=low
• the Judge has a separate OPENAI_JUDGE_KEY, so its spending can be seen separately from LiteLLM client traffic
• max_completion_tokens limits the maximum budget for the Judge response

What else can be done:

• reduce judgeMaxCompletionTokens: limit the number of reasoning and output tokens in the Judge response
• truncate large prompts and responses: reduce the number of input tokens (although here we need to preserve the beginning and end of the text symmetrically)
• use OpenAI Prompt Caching: reuse the stable prefix containing Judge instructions and reduce the cost of cached input
• use Batch API: accumulate evaluation jobs and process them asynchronously at a discount if delayed results are acceptable
• use Flex Processing: get a lower price in exchange for higher latency and possible temporary resource unavailability – which is not critical for LLM Evaluation

Judge impartiality

When I showed my code to our developers, they pointed out one interesting thing: in the system prompt for the Judge LLM and in its response schema, “score_a” and “score_b” originally came first, followed by “reasoning“.

This could lead to a situation where the LLM first assigns scores and then “pulls” the explanation toward the scores it has already chosen.

So now it works the other way around: first the LLM forms an evidence-based explanation of its decision, and only after that assigns the scores.

A very non-obvious and interesting nuance.

Basically, I’ve already described all this above, but here it is collected into one list:

• before being passed to the “judge”, model responses are randomly assigned as Response A and Response B
• model and provider names are not passed to the Judge LLM
• the Judge LLM must first produce an evidence-based reason
• only after the explanation does the Judge assign scores and select a winner
• the response is validated using a strict JSON Schema

This does not make the evaluation completely objective – an LLM Judge can still have its own biases, instability and preferences regarding response style. Basically, all those things we “love” LLMs for 🙂

Performance

Unlike the first version with the Mirroring script and the Evaluation Worker itself, the slow parts of the system are now moved out of the LiteLLM process into background workers.

Mirror Worker can perform several Matrix requests in parallel through MIRROR_CONCURRENCY – because we ran into a situation where the worker simply couldn’t keep up with all requests.

There are still questions around the llama.cpp configuration itself:

• actual Matrix throughput
• the number of concurrent inference requests supported by llama.cpp
• prompt sizes
• response lengths
• available GPU memory

But that’s a separate topic.

A Matrix request has a timeout of 120 seconds. Timeouts and HTTP errors are recorded separately in logs and metrics.

Redis and reliability

Redis is used as a temporary queue, not as permanent result storage.

The current Redis:

• runs in a single Pod
• has no authentication
• has no persistence
• has no replication or HA
• uses emptyDir
• has maxmemory 96 MB and the noeviction policy

This is an intentionally accepted limitation of the current solution: after restarting the Redis Pod, its jobs may be lost – but already recorded historical metrics remain in VictoriaMetrics, and traces remain in VictoriaTraces, so it is not critical.

And that’s basically it.

We are now additionally preparing LLM Evaluations with Opik and slowly testing different models.