Eval & Observability Skill

Role

You are an elite AI quality engineer. You build evaluation and observability systems

that catch quality regressions before they reach users. You know LLM-as-judge,

reference-based evals, behavioural testing, and production trace analysis.

The Eval Pyramid

                    ┌───────────────────┐
                    │  LLM-as-Judge     │  (automated, scalable)
                   ┌┤  (no reference)   │
                  / └───────────────────┘
                 /  ┌───────────────────┐
                /   │  Reference-Based  │  (compare to gold standard)
               ┌┤   │  Evals            │
              / └───────────────────────┘
             /  ┌───────────────────────┐
            /   │  Unit Evals           │  (deterministic checks)
           ┌┤   │  (exact match, regex) │
          / └───────────────────────────┘

Part 1: Custom Eval Harness

from dataclasses import dataclass, field
from typing import Callable, Any
import json, time, logging
from pathlib import Path

logger = logging.getLogger(__name__)


@dataclass
class EvalCase:
    """A single evaluation test case."""
    id: str
    input: str
    expected: str | dict | None = None  # None for LLM-as-judge
    metadata: dict = field(default_factory=dict)
    tags: list[str] = field(default_factory=list)


@dataclass
class EvalResult:
    """Result of running an eval case."""
    case_id: str
    output: str
    score: float  # 0.0 to 1.0
    passed: bool
    latency_ms: float
    reasoning: str = ""
    error: str = ""


class EvalHarness:
    """
    Run evaluation suites against an LLM pipeline.
    Tracks scores, latency, and regressions over time.
    """

    def __init__(self, name: str, llm_fn: Callable, scorer: Callable):
        self.name = name
        self.llm_fn = llm_fn
        self.scorer = scorer  # fn(output, case) -> float
        self.results: list[EvalResult] = []

    def run(self, cases: list[EvalCase],
             show_progress: bool = True) -> dict:
        """Run all eval cases and return summary."""
        self.results = []

        for case in cases:
            result = self._run_case(case)
            self.results.append(result)

            if show_progress:
                status = "✅" if result.passed else "❌"
                print(f"{status} [{case.id}] score={result.score:.2f} "
                      f"latency={result.latency_ms:.0f}ms")

        return self._summarise()

    def _run_case(self, case: EvalCase) -> EvalResult:
        start = time.time()
        error = ""
        output = ""

        try:
            output = self.llm_fn(case.input)
        except Exception as e:
            error = str(e)
            output = ""

        latency = (time.time() - start) * 1000

        if error:
            return EvalResult(case.id, output, 0.0, False, latency, error=error)

        try:
            score = self.scorer(output, case)
        except Exception as e:
            score = 0.0
            error = f"Scorer failed: {e}"

        return EvalResult(
            case_id=case.id,
            output=output,
            score=score,
            passed=score >= 0.7,
            latency_ms=latency,
            error=error
        )

    def _summarise(self) -> dict:
        if not self.results:
            return {}

        scores = [r.score for r in self.results]
        latencies = [r.latency_ms for r in self.results]
        passed = sum(1 for r in self.results if r.passed)

        return {
            "eval_name": self.name,
            "n_cases": len(self.results),
            "pass_rate": passed / len(self.results),
            "mean_score": sum(scores) / len(scores),
            "min_score": min(scores),
            "max_score": max(scores),
            "mean_latency_ms": sum(latencies) / len(latencies),
            "failed_cases": [r.case_id for r in self.results if not r.passed]
        }

    def save_results(self, path: str) -> None:
        """Save results to JSON for regression tracking."""
        data = {
            "eval_name": self.name,
            "timestamp": time.time(),
            "summary": self._summarise(),
            "results": [
                {
                    "case_id": r.case_id,
                    "score": r.score,
                    "passed": r.passed,
                    "latency_ms": r.latency_ms,
                    "output": r.output[:200],
                    "error": r.error
                }
                for r in self.results
            ]
        }
        Path(path).write_text(json.dumps(data, indent=2))

Part 2: LLM-as-Judge (The Perplexity/Anthropic Standard)

def build_judge_prompt(
    question: str,
    answer: str,
    criteria: list[str],
    reference: str = None
) -> str:
    """Build an evaluation prompt for LLM-as-judge."""
    criteria_str = "\n".join(f"- {c}" for c in criteria)
    ref_section = f"\nReference Answer:\n{reference}\n" if reference else ""

    return f"""You are an expert evaluator. Score the following answer.

Question: {question}
{ref_section}
Answer to Evaluate:
{answer}

Evaluation Criteria:
{criteria_str}

Instructions:
1. Score each criterion from 1-5
2. Provide brief reasoning for each score
3. Give an overall score from 0.0 to 1.0

Respond with JSON:
{{
  "criteria_scores": {{"criterion_name": score, ...}},
  "overall_score": 0.0-1.0,
  "reasoning": "brief explanation",
  "passed": true/false
}}"""


class LLMJudge:
    """
    LLM-as-judge scorer. Uses a strong model to evaluate outputs.
    This is the technique used by Anthropic's Constitutional AI team.
    """

    DEFAULT_CRITERIA = [
        "Factual accuracy — are the claims correct?",
        "Completeness — does it address the full question?",
        "Clarity — is it well-structured and unambiguous?",
        "Actionability — does it enable a decision or action?"
    ]

    def __init__(self, judge_llm_fn: callable,
                  criteria: list[str] = None,
                  pass_threshold: float = 0.7):
        self.judge_fn = judge_llm_fn
        self.criteria = criteria or self.DEFAULT_CRITERIA
        self.pass_threshold = pass_threshold

    def score(self, output: str, case: EvalCase) -> float:
        """Score an output using LLM-as-judge."""
        import json

        prompt = build_judge_prompt(
            question=case.input,
            answer=output,
            criteria=self.criteria,
            reference=case.expected if isinstance(case.expected, str) else None
        )

        try:
            response = self.judge_fn(prompt)
            data = json.loads(response)
            return float(data.get("overall_score", 0.0))
        except Exception as e:
            logger.error(f"Judge failed: {e}")
            return 0.0

Part 3: Deterministic Scorers

from difflib import SequenceMatcher
import re

def exact_match_scorer(output: str, case: EvalCase) -> float:
    """1.0 if exact match, 0.0 otherwise."""
    return 1.0 if output.strip() == str(case.expected).strip() else 0.0


def fuzzy_match_scorer(output: str, case: EvalCase,
                        threshold: float = 0.8) -> float:
    """Fuzzy string similarity score."""
    ratio = SequenceMatcher(
        None,
        output.strip().lower(),
        str(case.expected).strip().lower()
    ).ratio()
    return ratio


def contains_required_elements(output: str, case: EvalCase) -> float:
    """Score based on how many required elements are present."""
    required = case.metadata.get("required_elements", [])
    if not required:
        return 1.0

    output_lower = output.lower()
    found = sum(1 for elem in required if elem.lower() in output_lower)
    return found / len(required)


def json_validity_scorer(output: str, case: EvalCase) -> float:
    """Check if output is valid JSON matching expected schema keys."""
    try:
        data = json.loads(output)
        expected_keys = set(case.expected.keys()) if isinstance(case.expected, dict) else set()
        if expected_keys:
            actual_keys = set(data.keys())
            coverage = len(expected_keys & actual_keys) / len(expected_keys)
            return coverage
        return 1.0
    except (json.JSONDecodeError, AttributeError):
        return 0.0

Part 4: Lightweight Tracing (No External Dependencies)

import functools, uuid
from datetime import datetime

class LLMTracer:
    """
    Lightweight trace logging for LLM calls.
    Captures inputs, outputs, latency, and costs.
    No external dependencies required.
    """

    def __init__(self, log_path: str = "./llm_traces.jsonl"):
        self.log_path = log_path
        self._traces: list[dict] = []

    def trace(self, fn: callable) -> callable:
        """Decorator to trace an LLM function."""
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            trace_id = str(uuid.uuid4())[:8]
            start = time.time()

            # Capture input
            prompt = args[0] if args else kwargs.get("prompt", "")

            error = None
            result = None
            try:
                result = fn(*args, **kwargs)
                return result
            except Exception as e:
                error = str(e)
                raise
            finally:
                latency = (time.time() - start) * 1000
                entry = {
                    "trace_id": trace_id,
                    "timestamp": datetime.utcnow().isoformat(),
                    "function": fn.__name__,
                    "prompt_preview": str(prompt)[:200],
                    "output_preview": str(result)[:200] if result else None,
                    "latency_ms": round(latency, 1),
                    "error": error
                }
                self._traces.append(entry)
                self._write(entry)

        return wrapper

    def _write(self, entry: dict) -> None:
        with open(self.log_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def stats(self, last_n: int = 100) -> dict:
        """Summarise recent traces."""
        recent = self._traces[-last_n:]
        if not recent:
            return {}

        latencies = [t["latency_ms"] for t in recent if not t.get("error")]
        errors = [t for t in recent if t.get("error")]

        return {
            "total_calls": len(recent),
            "error_rate": len(errors) / len(recent),
            "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
        }

Part 5: IMI Eval Suite

# Standard IMI eval cases for fan segment classification
IMI_SEGMENT_EVAL_CASES = [
    EvalCase(
        id="tribal_001",
        input="30-year season ticket holder, cries when team loses, names child after club legend.",
        expected="tribal",
        metadata={"required_elements": ["tribal"], "segment": "tribal"},
        tags=["segment", "edge-case"]
    ),
    EvalCase(
        id="casual_001",
        input="Watches big games on TV, went to one game last year, supports because of family tradition.",
        expected="casual",
        metadata={"required_elements": ["casual"]},
        tags=["segment"]
    ),
    EvalCase(
        id="corporate_001",
        input="Attends via corporate box, primarily for client entertainment, doesn't follow results.",
        expected="corporate",
        metadata={"required_elements": ["corporate"]},
        tags=["segment"]
    ),
]

def run_imi_segment_eval(llm_fn: callable) -> dict:
    """Run the standard IMI segment classification eval."""
    def scorer(output: str, case: EvalCase) -> float:
        return 1.0 if case.expected in output.lower() else 0.0

    harness = EvalHarness("imi-segment-classification", llm_fn, scorer)
    return harness.run(IMI_SEGMENT_EVAL_CASES)

Output Standards

• Run evals before AND after every prompt change
• Target pass rate: >85% on all standard eval suites
• Use LLM-as-judge for open-ended outputs; deterministic scorers for extractions
• Save all eval results with timestamps for regression detection
• Log 100% of production LLM calls — latency and errors at minimum
• Alert if error rate exceeds 2% or p95 latency exceeds 5 seconds

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)