production-eval-strategy

Production Evaluation Strategy

How to evaluate agents in production without breaking the bank or slowing things down.

Core Principle

Production eval is fundamentally different from offline eval:

• Offline eval: Run on test set, comprehensive, blocking
• Production eval: Sample real traffic, async, non-blocking

You need both. This skill focuses on production.

The Production Eval Stack

┌─────────────────────────────────────────────────────────────┐
│                    Production Traffic                        │
└─────────────────┬───────────────────────────────────────────┘
                  │
        ┌─────────▼─────────┐
        │   Sampling Layer   │  ← What % to evaluate?
        └─────────┬─────────┘
                  │
    ┌─────────────▼─────────────┐
    │   Async Evaluation Queue   │  ← Non-blocking
    └─────────────┬─────────────┘
                  │
    ┌─────────────▼─────────────┐
    │   Evaluation Workers       │  ← LLM-as-judge, heuristics
    └─────────────┬─────────────┘
                  │
    ┌─────────────▼─────────────┐
    │   Scores → Observability   │  ← Langfuse, etc.
    └─────────────┬─────────────┘
                  │
    ┌─────────────▼─────────────┐
    │   Dashboards & Alerts      │  ← Regression detection
    └─────────────────────────────┘

Sampling Strategies

Random Sampling

import random
from langfuse.decorators import observe

EVAL_SAMPLE_RATE = 0.1  # Evaluate 10% of traffic

@observe(name="agent.run")
def run_agent(task: str) -> str:
    result = agent.invoke(task)

    # Random sampling
    should_eval = random.random() < EVAL_SAMPLE_RATE

    if should_eval:
        # Queue for async evaluation
        eval_queue.send({
            "trace_id": langfuse_context.get_current_trace_id(),
            "task": task,
            "output": result,
            "timestamp": datetime.utcnow().isoformat(),
        })

    return result

Stratified Sampling

# Different sample rates by segment
SAMPLE_RATES = {
    "new_user": 0.5,      # 50% - more signal needed
    "power_user": 0.05,   # 5% - already know behavior
    "enterprise": 1.0,    # 100% - high stakes
    "default": 0.1,
}

def get_sample_rate(user_id: str, user_tier: str) -> float:
    return SAMPLE_RATES.get(user_tier, SAMPLE_RATES["default"])

@observe(name="agent.run")
def run_agent(task: str, user_id: str, user_tier: str) -> str:
    result = agent.invoke(task)

    sample_rate = get_sample_rate(user_id, user_tier)
    should_eval = random.random() < sample_rate

    langfuse_context.update_current_observation(
        metadata={
            "eval_sampled": should_eval,
            "sample_rate": sample_rate,
            "user_tier": user_tier,
        }
    )

    if should_eval:
        queue_for_eval(task, result)

    return result