6.5 The Evaluation Lifecycle & Online Production Monitoring

Canonical Curriculum Reference: docs/plan/06_phase6_stateful_graphs_hitl.md
Textbook Reference: AI Agents and Applications (Manning Publications), Chapter 14, Section 14.3.4 ("Evaluation of AI agents and applications", pp. 375–376).

---

🔄 1. The Evaluation Lifecycle: Offline Testing vs. Online Sampling

In enterprise software engineering, code quality is assured through unit tests, integration tests, and staging environments. In Generative AI systems, offline testing alone is insufficient:

• Users submit novel, unanticipated queries that are never represented in static golden test sets.
• Real-world document stores and external APIs undergo schema drift and semantic updates.
• Prompt injection and jailbreak patterns continuously evolve in the wild.

A complete evaluation strategy requires a dual-phase Evaluation Lifecycle:

+-----------------------------------------------------------------------+
|                       THE EVALUATION LIFECYCLE                        |
+-----------------------------------------------------------------------+
|                                                                       |
|   +---------------------------------------------------------------+   |
|   |         PHASE 1: OFFLINE EVALUATION (Pre-Deployment)          |   |
|   |  - Static, versioned golden datasets (20-50 curated cases)   |   |
|   |  - Automated Pytest suites executed in CI/CD pipeline         |   |
|   |  - Hard blocker: PR rejected if Faithfulness < 90%            |   |
|   +---------------------------------------------------------------+   |
|                                  |                                    |
|                       [ Deploy to Production ]                        |
|                                  |                                    |
|                                  v                                    |
|   +---------------------------------------------------------------+   |
|   |          PHASE 2: ONLINE EVALUATION (Post-Deployment)         |   |
|   |  - Live production traffic sampling (5-10% of all traces)     |   |
|   |  - Asynchronous background LLM-as-a-Judge scoring             |   |
|   |  - LangFuse metric monitors & PagerDuty alerts on drift       |   |
|   +---------------------------------------------------------------+   |
|                                                                       |
+-----------------------------------------------------------------------+

---

⚖️ 2. Online Production Sampling & LLM-as-a-Judge Mechanics

Running an LLM judge on $100\mathrm{\%}$ of production traffic would double infrastructure costs and add latency. Instead, production systems employ stochastic sampling:

1. Sampling Gate: The harness evaluates each completed production trace against a sampling probability (e.g. $p=0.05$).
2. Asynchronous Decoupling: Sampled traces are placed on an asynchronous queue (Celery, AWS SQS, or background worker pool). The user response is returned immediately with zero added latency.
3. Judge Execution: An independent, highly aligned evaluator model (e.g. Claude 3.5 Sonnet or GPT-4o) evaluates the trace against standard criteria:
   • Hallucination Detection: Did the model make claims not grounded in retrieved tools?
   • Goal Alignment: Did the agent fulfill the user's explicit objective?
   • Tone & Safety Compliance: Did the completion adhere to enterprise brand guidelines?
4. Metric Telemetry: Scores are committed directly to the trace in LangFuse via langfuse_context.score_current_trace().

---

💻 3. Production Implementation: Asynchronous Live Traffic Evaluator

import asyncio
import random
from typing import Final
from pydantic import BaseModel, ConfigDict, Field


class JudgeEvaluationScore(BaseModel):
    """Immutable scoring payload generated by online LLM judge."""
    model_config = ConfigDict(frozen=True, extra="forbid")
    
    trace_id: str
    faithfulness_score: float = Field(ge=0.0, le=1.0)
    hallucination_detected: bool
    evaluation_rationale: str


class OnlineProductionEvaluator:
    """Samples live production traffic and applies asynchronous quality evaluation."""

    def __init__(self, sample_rate: float = 0.10) -> None:
        if not (0.0 < sample_rate <= 1.0):
            raise ValueError("Sample rate must be in range (0.0, 1.0]")
        self._sample_rate: Final[float] = sample_rate

    def should_sample(self) -> bool:
        """Determines if current production trace should be evaluated."""
        return random.random() < self._sample_rate

    async def evaluate_trace_async(
        self,
        trace_id: str,
        user_prompt: str,
        retrieved_context: str,
        agent_answer: str
    ) -> JudgeEvaluationScore:
        """Executes LLM-as-a-judge scoring out-of-band."""
        # Simulate asynchronous judge evaluation prompt
        await asyncio.sleep(0.05)  # Simulates model network latency
        
        # In production: call Claude-3.5-Sonnet with strict evaluation schema
        # For demonstration: assert grounding
        is_grounded = all(term in retrieved_context for term in ["Tokyo", "culinary"])
        score = 0.95 if is_grounded else 0.40
        
        return JudgeEvaluationScore(
            trace_id=trace_id,
            faithfulness_score=score,
            hallucination_detected=(score < 0.80),
            evaluation_rationale="Claims were strictly verified against grounding context." if is_grounded else "Unsubstantiated claims detected."
        )

---

🚨 4. LangFuse Metric Monitors & Automated Threshold Alerting

In LangFuse, sampled online scores aggregate into rolling metrics. Senior architects configure Automated Metric Monitors:

Metric	Target SLA	Alert Threshold	Automated Action
Hallucination Rate	$<2.0\mathrm{\%}$	$>5.0\mathrm{\%}$ over 15 mins	PagerDuty Alert $\to$ Trip circuit breaker to fallback model
Context Precision	$\ge 0.85$	$<0.70$ over 1 hour	Slack notification to retrieval team $\to$ Inspect vector index
User Escalation / Rejection	$<1.0\mathrm{\%}$	$>3.0\mathrm{\%}$ over 30 mins	Trigger human review queue for all ongoing threads

---

🧪 5. Automated Verification with Pytest

import pytest

@pytest.mark.asyncio
async def test_online_evaluator_sampling_and_scoring():
    evaluator = OnlineProductionEvaluator(sample_rate=1.0)  # 100% for test determinism
    
    assert evaluator.should_sample() is True
    
    score = await evaluator.evaluate_trace_async(
        trace_id="prod-trace-987",
        user_prompt="Find vegan tours in Tokyo.",
        retrieved_context="Tokyo vegan culinary tours operate daily in Shibuya.",
        agent_answer="We offer daily Tokyo vegan culinary tours in Shibuya."
    )
    
    assert score.faithfulness_score >= 0.90
    assert score.hallucination_detected is False
    assert "Tokyo" in score.evaluation_rationale