LLM-as-judge & automated evaluation

from openai import OpenAI
import json

JUDGE_MODEL = "gpt-4o-mini"
RUBRIC = """You grade support answers for regression eval — not live customers.
Score each dimension 1-5 using the anchors in the rubric doc.
Return ONLY JSON matching the schema."""

JUDGE_SCHEMA = {
    "type": "json_schema",
    "json_schema": {
        "name": "judge_scores",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "correctness": {"type": "integer", "minimum": 1, "maximum": 5},
                "completeness": {"type": "integer", "minimum": 1, "maximum": 5},
                "policy_ok": {"type": "boolean"},
                "reason": {"type": "string", "maxLength": 280},
            },
            "required": ["correctness", "completeness", "policy_ok", "reason"],
            "additionalProperties": False,
        },
    },
}

def judge_pointwise(question: str, answer: str, context: str | None = None) -> dict:
    client = OpenAI()
    user = f"QUESTION:\n{question}\n\nANSWER:\n{answer}"
    if context:
        user += f"\n\nCONTEXT:\n{context}"
    resp = client.chat.completions.create(
        model=JUDGE_MODEL,
        temperature=0,
        messages=[
            {"role": "system", "content": RUBRIC},
            {"role": "user", "content": user},
        ],
        response_format=JUDGE_SCHEMA,
    )
    return json.loads(resp.choices[0].message.content)

def passes_gate(scores: dict, min_correctness: int = 4) -> bool:
    return scores["policy_ok"] and scores["correctness"] >= min_correctness

from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)

# Each row: question, answer, contexts (list[str]), ground_truth (optional)
rows = [{
    "question": "What is the refund window?",
    "answer": "Refunds are available within 30 days of purchase.",
    "contexts": [
        "Section 4.2: Refunds within 30 days of delivery with receipt.",
        "Shipping times vary by region.",
    ],
    "ground_truth": "30 days with receipt.",
}]
ds = Dataset.from_list(rows)

result = evaluate(
    ds,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
print(result)  # per-metric scores 0.0–1.0

# Gate example
assert result["faithfulness"] >= 0.85, "hallucination regression"

import re
from openai import OpenAI

GEVAL_RUBRIC = """Evaluate the ANSWER for correctness and completeness.
Steps:
1. Identify key points the answer should cover.
2. Note covered points and missing/incorrect points.
3. Assign a score from 1.0 to 10.0 (decimals allowed).

End with a JSON line: {"score": <float>, "summary": "<one sentence>"}"""

def g_eval(question: str, answer: str, reference: str | None = None) -> dict:
    client = OpenAI()
    user = f"QUESTION:\n{question}\n\nANSWER:\n{answer}"
    if reference:
        user += f"\n\nREFERENCE:\n{reference}"
    resp = client.chat.completions.create(
        model="gpt-4o",
        temperature=0,
        messages=[
            {"role": "system", "content": GEVAL_RUBRIC},
            {"role": "user", "content": user},
        ],
    )
    text = resp.choices[0].message.content or ""
    # CoT preserved in text; extract structured tail
    match = re.search(r'\{"score":\s*([\d.]+).*?"summary":\s*"([^"]*)"', text, re.S)
    if match:
        return {"score": float(match.group(1)), "summary": match.group(2), "cot": text}
    raise ValueError("failed to parse G-Eval output")

def normalize_to_unit(score_1_10: float) -> float:
    return max(0.0, min(1.0, (score_1_10 - 1) / 9))

import statistics

def g_eval_robust(question: str, answer: str, n: int = 3) -> dict:
    scores = []
    cot_blocks = []
    for _ in range(n):
        r = g_eval(question, answer)  # temperature 0.7 inside
        scores.append(r["score"])
        cot_blocks.append(r["cot"])
    return {
        "score": statistics.median(scores),
        "score_std": statistics.pstdev(scores) if len(scores) > 1 else 0,
        "cot_samples": cot_blocks,
    }

from openai import OpenAI
import json

PAIRWISE_RUBRIC = """Compare two assistant answers to the same user question.
Pick the better answer for correctness, helpfulness, and policy compliance.
Return JSON: {"winner": "A"|"B"|"tie", "reason": "..."}"""

def pairwise_once(client, question: str, first: str, second: str) -> str:
    user = (
        f"QUESTION:\n{question}\n\n"
        f"ANSWER A:\n{first}\n\n"
        f"ANSWER B:\n{second}"
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        messages=[
            {"role": "system", "content": PAIRWISE_RUBRIC},
            {"role": "user", "content": user},
        ],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    return data["winner"], data["reason"]

def pairwise_swap(question: str, answer_a: str, answer_b: str) -> dict:
    client = OpenAI()
    w1, r1 = pairwise_once(client, question, answer_a, answer_b)
    w2, r2 = pairwise_once(client, question, answer_b, answer_a)
    # Map second call back to original labels
    if w2 == "A":
        w2_mapped = "B"
    elif w2 == "B":
        w2_mapped = "A"
    else:
        w2_mapped = "tie"
    if w1 == w2_mapped and w1 != "tie":
        return {"winner": w1, "consistent": True, "reasons": [r1, r2]}
    return {"winner": "tie", "consistent": False, "reasons": [r1, r2]}

import json
from pathlib import Path

def human_pass(row: dict) -> bool:
    return row["human_score"] >= 4 and row.get("human_policy_ok", True)

def judge_pass(row: dict) -> bool:
    j = row["judge_scores"]
    return j["correctness"] >= 4 and j["policy_ok"]

def calibration_report(path: Path) -> dict:
    rows = [json.loads(l) for l in path.read_text().splitlines() if l.strip()]
    agree = sum(human_pass(r) == judge_pass(r) for r in rows)
    n = len(rows)
    # Simple pass/fail agreement
    return {
        "n": n,
        "agreement_pct": round(100 * agree / n, 1),
        "false_pass": sum(not human_pass(r) and judge_pass(r) for r in rows),
        "false_fail": sum(human_pass(r) and not judge_pass(r) for r in rows),
    }

# python calibrate.py eval/gold/human_labeled.jsonl
# → agreement_pct: 87.2, false_pass: 3 (dangerous), false_fail: 8

# eval/gold/[email protected]/cases.jsonl (one JSON object per line)
{
  "id": "refund-014",
  "question": "Can I get a refund after 45 days?",
  "tags": ["refund", "policy", "negative"],
  "context_doc_ids": ["policy-refund-v3"],
  "must_contain": ["30 day"],
  "must_not_contain": ["guaranteed", "full refund"],
  "human_score": 5,
  "human_policy_ok": true,
  "reference_answer": "Refunds are available within 30 days of delivery with receipt.",
  "dataset_version": "[email protected]"
}

from openai import OpenAI
import json

SYNTH_PROMPT = """Given this policy chunk, generate 3 user questions a customer might ask.
Include one edge case near policy boundary. Return JSON array of strings."""

def synth_questions(chunk: str, n_chunks: int = 1) -> list[str]:
    client = OpenAI()
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.7,
        messages=[
            {"role": "system", "content": SYNTH_PROMPT},
            {"role": "user", "content": chunk},
        ],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    return data.get("questions", [])

# Human-review sample before promoting synth rows to gold

import os
from eval.scorers import score_case_rules

def judge_mock(case: dict, answer: str) -> dict:
    rules = score_case_rules(case, answer)
    return {
        "correctness": 5 if rules["pass"] else 2,
        "completeness": 5 if rules["pass"] else 2,
        "policy_ok": rules["pass"],
        "reason": "mock_derived_from_rules",
        "mode": "mock",
    }

def judge_live(case: dict, answer: str) -> dict:
    from eval.judges.pointwise import judge_pointwise
    out = judge_pointwise(case["question"], answer)
    out["mode"] = "live"
    return out

def run_judge(case: dict, answer: str) -> dict:
    mode = os.environ.get("JUDGE_MODE", "mock").lower()
    if mode == "off":
        return {"skipped": True}
    if mode == "mock":
        return judge_mock(case, answer)
    if mode == "live":
        return judge_live(case, answer)
    raise ValueError(f"unknown JUDGE_MODE: {mode}")

# .github/workflows/eval-smoke.yml
name: eval-smoke
on: [pull_request]
jobs:
  smoke:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r eval/requirements.txt
      - run: python eval/run_suite_judged.py --smoke
        env:
          JUDGE_MODE: mock
          DATASET_VERSION: [email protected]
      - uses: actions/upload-artifact@v4
        with:
          name: eval-report
          path: reports/latest.json