Evaluating RAG Answers

question   What are the default values of lease_seconds and heartbeat_seconds in orbit.toml?
reference  In the [worker] section, lease_seconds defaults to 300 and heartbeat_seconds defaults to 30. ...
context    configuration-reference ✓  leases-and-heartbeats ✓  deployment-kubernetes
answer     The default lease_seconds is 600 and heartbeat_seconds is 30 [deployment-kubernetes].
D29        recall@5 = 1.0 · bucket = ok

# script_id: day_030_answer_evaluation/hand_grades
# The hand-graded grid from this section, as data. No install, no API key.
CRITERIA = ("faithful", "relevant", "context", "correct")
GRADES = {
    "A (good)":      (True, True, True, True),
    "B (invented)":  (False, True, True, False),
    "C (off-topic)": (True, False, True, False),
}
MEANING = {
    "faithful": "made it up",
    "relevant": "didn't answer",
    "context": "wrong pages",
    "correct": "disagrees with the answer key",
}

print(f"{'answer':<15}" + "".join(f"{c:>10}" for c in CRITERIA))
for name, row in GRADES.items():
    print(f"{name:<15}" + "".join(f"{'PASS' if ok else 'FAIL':>10}" for ok in row))
for name, row in GRADES.items():
    fails = [MEANING[c] for c, ok in zip(CRITERIA, row) if not ok]
    print(f"{name}: {', '.join(fails) or 'nothing to fix'}")

answer           faithful  relevant   context   correct
A (good)             PASS      PASS      PASS      PASS
B (invented)         FAIL      PASS      PASS      FAIL
C (off-topic)        PASS      FAIL      PASS      FAIL
A (good): nothing to fix
B (invented): made it up, disagrees with the answer key
C (off-topic): didn't answer, disagrees with the answer key

# script_id: day_030_answer_evaluation/faithfulness_judge
# Minimal binary faithfulness judge. Install: uv add "openai>=3.46,<4" pydantic
# Env: OPENAI_API_KEY; ATLAS_MODEL (optional). Cost: one call, well under $0.01 (as of 2026-11, verify)
import os
from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

class Verdict(BaseModel):
    reason: str   # first: the model writes its evidence before it decides
    passed: bool

CONTEXT = "[worker]\nlease_seconds = 300       # deadline if no heartbeat arrives\nheartbeat_seconds = 30"
ANSWER = "It defaults to 600 s. Expired jobs move to the dead-letter queue."
resp = client.responses.parse(
    model=MODEL,
    instructions="You check answers for faithfulness. PASS only if every claim in <answer> is "
                 "supported by <context>. Give a one-sentence reason, then the verdict.",
    input=f"<context>\n{CONTEXT}\n</context>\n<answer>\n{ANSWER}\n</answer>",
    text_format=Verdict,
)
print(resp.output_parsed)

reason='The answer says the default is 600 s; the context sets lease_seconds = 300, and it never mentions the dead-letter queue.' passed=False

# script_id: day_030_answer_evaluation/break_it
# Planted errors, judged with and without context; then the same answers on a 1-5 scale.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA (optional)
# Cost: 25 short calls, ≈ $0.01 (as of 2026-11, verify)
import os
from pathlib import Path
from openai import OpenAI
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
client = OpenAI()
PAGE = (DATA_DIR / "docs" / "leases-and-heartbeats.md").read_text(encoding="utf-8")

class Verdict(BaseModel):
    reason: str
    passed: bool

PAIRS = [  # (correct, same sentence with one planted error)
    ("lease_seconds defaults to 300.", "lease_seconds defaults to 600."),
    ("When a lease expires, the job becomes claimable again.", "When a lease expires, the job is deleted."),
    ("Orbit prefers running a job twice to never running it.", "Orbit 3.2 guarantees a job never runs twice."),
    ("orbit doctor warns when heartbeat_seconds exceeds a third of lease_seconds.",
     "orbit doctor refuses to start a worker when heartbeat_seconds exceeds a third of lease_seconds."),
    ("2.x workers heartbeated at one tenth of the lease.", "2.x workers heartbeated every 10 seconds."),
]
RULE = "PASS only if every claim in <answer> is supported by <context>. Reason first, then verdict."

def passes(answer: str, context: str | None) -> bool:
    ctx = f"<context>\n{context}\n</context>\n" if context else ""
    r = client.responses.parse(model=MODEL, instructions=RULE, text_format=Verdict,
                               input=f"{ctx}<answer>\n{answer}\n</answer>")
    return bool(r.output_parsed and r.output_parsed.passed)

for label, ctx in (("with context", PAGE), ("without context", None)):
    clean = sum(passes(good, ctx) for good, _ in PAIRS)
    caught = sum(not passes(bad, ctx) for _, bad in PAIRS)
    print(f"{label:<16} clean passed {clean}/5 · planted caught {caught}/5")

with context     clean passed 5/5 · planted caught 5/5
without context  clean passed 1/5 · planted caught 4/5

# script_id: day_030_answer_evaluation/break_it
class Score(BaseModel):
    reason: str
    score: int  # 1 (bad) .. 5 (excellent)

def rate(answer: str) -> int:
    r = client.responses.parse(model=MODEL, text_format=Score,
                               instructions="Rate this answer from 1 (bad) to 5 (excellent) for "
                                            "faithfulness to <context>. Reason first.",
                               input=f"<context>\n{PAGE}\n</context>\n<answer>\n{answer}\n</answer>")
    return r.output_parsed.score if r.output_parsed else 0

print("1-5 scores for the planted answers:", [rate(bad) for _, bad in PAIRS])

1-5 scores for the planted answers: [2, 2, 4, 3, 4]

# script_id: day_030_answer_evaluation/rag
# rag.py: compact stand-in for Days 28-29 (ask(), RagRun, diagnose()). Swap in your real atlas.rag.
# Install: uv add "openai>=3.46,<4" pydantic rank-bm25
# Env: OPENAI_API_KEY; ATLAS_MODEL, ORBIT_DATA (optional). No database: the index is in memory.
import json
import os
import re
import time
from dataclasses import dataclass
from pathlib import Path

from openai import AsyncOpenAI
from pydantic import BaseModel
from rank_bm25 import BM25Okapi

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
PRICES = {"gpt-6-luna": (0.10, 0.50)}  # USD per 1M input/output tokens; as of 2026-11, verify
aclient = AsyncOpenAI()

def cost_usd(usage, model: str = MODEL) -> float:
    """Rough cost of one call (compact Day 12 tracker; ignores cache discounts)."""
    price_in, price_out = PRICES.get(model, (0.0, 0.0))
    return (usage.input_tokens * price_in + usage.output_tokens * price_out) / 1e6

class Citation(BaseModel):
    doc_id: str
    quote: str   # exact text copied from the source

class Answer(BaseModel):
    text: str
    citations: list[Citation]

# script_id: day_030_answer_evaluation/rag
@dataclass
class Hit:
    doc_id: str
    text: str
    score: float = 0.0

@dataclass
class RagRun:                 # Day 29's record of one pipeline run
    question: str
    answer: Answer
    hits: list[Hit]           # the whole candidate pool, best first
    context_ids: list[str]    # doc_ids of the top-k hits that reached the prompt
    latency_s: float
    cost_usd: float
    context: str = ""         # new today: the exact text the generator saw (judges need it)

def tokenize(text: str) -> list[str]:
    return re.findall(r"[a-z0-9_]+", text.lower())

def load_chunks(data_dir: Path = DATA_DIR) -> list[Hit]:
    chunks = []
    for path in sorted((data_dir / "docs").glob("*.md")):
        body = re.sub(r"\A---\n.*?\n---\n", "", path.read_text(encoding="utf-8"), flags=re.S)
        chunks += [Hit(path.stem, s.strip()) for s in re.split(r"\n(?=## )", body) if s.strip()]
    return chunks

class KeywordIndex:
    """In-memory BM25: a stand-in for Day 27's Index.hybrid_search()."""
    def __init__(self, chunks: list[Hit]) -> None:
        self.chunks = chunks
        self.bm25 = BM25Okapi([tokenize(c.text) for c in chunks])

    def search(self, query: str, k: int = 20) -> list[Hit]:
        scores = self.bm25.get_scores(tokenize(query))
        best = sorted(range(len(scores)), key=lambda i: -scores[i])[:k]
        return [Hit(self.chunks[i].doc_id, self.chunks[i].text, float(scores[i])) for i in best]

# script_id: day_030_answer_evaluation/rag
IDK_TEXT = "I can't find that in the Orbit docs."
RAG_INSTRUCTIONS = (
    "You are Atlas, a support assistant for the Orbit job library. Answer ONLY from the sources "
    "in <context>. Cite each claim with its doc_id and an exact quote of at most 25 words. "
    f"If the sources don't contain the answer, reply exactly '{IDK_TEXT}' with no citations.")

def verify_citations(answer: Answer, hits: list[Hit]) -> list[Citation]:
    """Day 28: return the citations whose quote is NOT in a context chunk of that doc."""
    norm = lambda s: " ".join(s.split()).lower()
    return [c for c in answer.citations
            if not any(c.doc_id == h.doc_id and norm(c.quote) in norm(h.text) for h in hits)]

async def arun_rag(question: str, *, index: KeywordIndex, k: int = 6, pool: int = 20,
                   model: str = MODEL) -> RagRun:
    start = time.perf_counter()
    hits = index.search(question, k=pool)
    context = "\n\n".join(f'<source doc_id="{h.doc_id}">\n{h.text}\n</source>' for h in hits[:k])
    resp = await aclient.responses.parse(
        model=model, instructions=RAG_INSTRUCTIONS, text_format=Answer, reasoning={"effort": "low"},
        input=f"<context>\n{context}\n</context>\n\nQuestion: {question}")
    answer = resp.output_parsed or Answer(text=IDK_TEXT, citations=[])  # refusal/incomplete → abstain
    return RagRun(question, answer, hits, [h.doc_id for h in hits[:k]],
                  time.perf_counter() - start, cost_usd(resp.usage, model), context)

def diagnose(case: dict, run: RagRun) -> str:
    """Day 29: which stage lost the gold page? Cases with no gold citations return 'ok'."""
    gold = set(case.get("citations", []))
    if not gold:
        return "ok"
    if not gold & {h.doc_id for h in run.hits}:
        return "retrieval"
    if not gold & set(run.context_ids):
        return "ranking"
    return "generation" if not gold & {c.doc_id for c in run.answer.citations} else "ok"

def load_qa_cases(path: Path | None = None) -> list[dict]:
    """qa_gold rows as plain dicts (Day 20's case shape); answerable unless the row says otherwise."""
    path = path or DATA_DIR / "eval" / "qa_gold.jsonl"
    lines = path.read_text(encoding="utf-8").splitlines()
    return [{"answerable": True, **json.loads(line)} for line in lines if line.strip()]

# script_id: day_030_answer_evaluation/answers
# answers.py: atlas/evals/answers.py, four binary judges + two code checks.
# Install: uv add "openai>=3.46,<4" pydantic rank-bm25 · Env: OPENAI_API_KEY; ATLAS_JUDGE_MODEL (optional)
import asyncio
import inspect
import os
from collections import Counter
from dataclasses import dataclass
from typing import Awaitable, Callable, Literal

from pydantic import BaseModel
from rag import IDK_TEXT, MODEL, RagRun, aclient, arun_rag, cost_usd, diagnose, verify_citations

JUDGE_MODEL = os.environ.get("ATLAS_JUDGE_MODEL", MODEL)
Criterion = Literal["faithfulness", "answer_relevance", "context_relevance", "correctness"]

class Verdict(BaseModel):
    reason: str    # evidence first (schema key order), so the verdict follows it
    passed: bool

JUDGE_SYSTEM = ("You grade one output of a support assistant on ONE criterion. In `reason`, quote "
                "the evidence in one or two sentences; then set `passed`. When unsure, FAIL.")
JUDGE_PROMPTS: dict[Criterion, str] = {
    "faithfulness": "Is every factual claim in <answer> stated in, or directly implied by, "
                    "<context>? A plausible claim missing from <context> is a FAIL. An answer that "
                    "says it can't find the information is a PASS.",
    "answer_relevance": "Does <answer> directly answer what <question> asks? Ignore whether it is "
                        "true. A refusal, a deflection or an answer to another question is a FAIL.",
    "context_relevance": "Does <context> contain the information needed to answer <question>? "
                         "Ignore any answer.",
    "correctness": "Does <answer> agree with <reference> on every fact the question asks about? "
                   "Extra correct detail is fine. A wrong or missing number, version, name or "
                   "yes/no is a FAIL.",
}
REQUIRES: dict[Criterion, tuple[str, ...]] = {
    "faithfulness": ("answer", "context"), "answer_relevance": ("answer",),
    "context_relevance": ("context",), "correctness": ("answer", "reference")}
COSTS: Counter[str] = Counter()   # judge $ per criterion (Day 12's CostTracker does this properly)

# script_id: day_030_answer_evaluation/answers
async def ajudge(criterion: Criterion, *, question: str, answer: str = "",
                 context: str | None = None, reference: str | None = None,
                 model: str = JUDGE_MODEL) -> Verdict:
    given = {"answer": answer, "context": context, "reference": reference}
    missing = [name for name in REQUIRES[criterion] if not given[name]]
    if missing:
        raise ValueError(f"{criterion} judge needs {missing}; without them it can only guess")
    parts = [f"<question>\n{question}\n</question>"]
    parts += [f"<{name}>\n{given[name]}\n</{name}>" for name in REQUIRES[criterion]]
    resp = await aclient.responses.parse(
        model=model, instructions=JUDGE_SYSTEM, text_format=Verdict, reasoning={"effort": "low"},
        input=f"Criterion: {JUDGE_PROMPTS[criterion]}\n\n" + "\n".join(parts))
    COSTS[criterion] += cost_usd(resp.usage, model)
    # No parsed verdict (refusal or truncation) counts as a FAIL, so it shows up in the report.
    return resp.output_parsed or Verdict(reason=f"no verdict (status={resp.status})", passed=False)

def judge(criterion: Criterion, **kwargs) -> Verdict:
    """Sync wrapper for scripts. Inside a running event loop (e.g. Jupyter), await ajudge() instead."""
    return asyncio.run(ajudge(criterion, **kwargs))

# script_id: day_030_answer_evaluation/answers
@dataclass
class CheckResult:        # Day 20's shape
    name: str
    passed: bool
    detail: str = ""

Check = Callable[[dict, RagRun], "CheckResult | None | Awaitable[CheckResult | None]"]

def abstained(run: RagRun) -> bool:
    return run.answer.text.strip() == IDK_TEXT and not run.answer.citations

def check_citations_valid(case: dict, run: RagRun) -> CheckResult | None:
    if abstained(run):
        return None                       # nothing cited, nothing to verify
    context_hits = run.hits[: len(run.context_ids)]   # context_ids come from hits[:k], in order
    bad = verify_citations(run.answer, context_hits)
    ok = bool(run.answer.citations) and not bad
    return CheckResult("citations_valid", ok, f"{len(bad)} bad: {[c.doc_id for c in bad]}"
                       if run.answer.citations else "answer has no citations")

def check_abstains(case: dict, run: RagRun) -> CheckResult:
    should = not case["answerable"]
    return CheckResult("abstains", abstained(run) == should, f"abstained={abstained(run)}, should={should}")

UNANSWERABLE = [{"id": f"unans-{i}", "question": q, "answer": "", "citations": [], "answerable": False}
                for i, q in enumerate([
                    "What open-source license is Orbit released under?",
                    "How do I pin Orbit workers to specific GPUs?",
                    "Is there an official TypeScript client for enqueueing Orbit jobs?",
                    "How do I run Orbit workers on AWS Lambda?",
                    "How do I migrate my Celery tasks to Orbit?"], start=1)]

# script_id: day_030_answer_evaluation/answers
def judge_check(criterion: Criterion) -> Check:
    async def check(case: dict, run: RagRun) -> CheckResult | None:
        if not case["answerable"]:
            return None                   # unanswerable: only check_abstains applies
        v = await ajudge(criterion, question=case["question"], answer=run.answer.text,
                         context=run.context, reference=case["answer"])
        return CheckResult(criterion, v.passed, v.reason)
    return check

def rag_answer_checks() -> list[Check]:
    return [check_citations_valid, check_abstains, *(judge_check(c) for c in JUDGE_PROMPTS)]

async def _apply(check: Check, case: dict, run: RagRun) -> CheckResult | None:
    result = check(case, run)
    return await result if inspect.isawaitable(result) else result

async def answer_eval(cases: list[dict], *, index, predict=arun_rag,
                      checks: list[Check] | None = None, concurrency: int = 5) -> dict:
    checks = checks or rag_answer_checks()
    sem = asyncio.Semaphore(concurrency)

    async def one(case: dict) -> dict:
        async with sem:
            run = await predict(case["question"], index=index)
            results = await asyncio.gather(*(_apply(c, case, run) for c in checks))
        return {"id": case["id"], "question": case["question"], "answer": run.answer.text,
                "bucket": diagnose(case, run), "cost_usd": run.cost_usd,
                "checks": {r.name: {"passed": r.passed, "detail": r.detail} for r in results if r}}

    rows = await asyncio.gather(*(one(c) for c in cases))
    summary: dict[str, dict[str, int]] = {}
    for row in rows:
        for name, r in row["checks"].items():
            s = summary.setdefault(name, {"passed": 0, "n": 0})
            s["passed"], s["n"] = s["passed"] + r["passed"], s["n"] + 1
    return {"model": MODEL, "judge_model": JUDGE_MODEL, "summary": summary, "rows": rows,
            "generation_usd": sum(r["cost_usd"] for r in rows), "judge_usd": dict(COSTS)}

def format_report(report: dict) -> str:
    lines = [f"{'check':<18}{'pass':>8}{'rate':>7}{'judge $':>10}"]
    for name, s in report["summary"].items():
        cost = f"{report['judge_usd'][name]:.4f}" if name in report["judge_usd"] else "-"
        lines.append(f"{name:<18}{s['passed']:>4}/{s['n']:<3}{s['passed'] / s['n']:>7.2f}{cost:>10}")
    total = report["generation_usd"] + sum(report["judge_usd"].values())
    lines.append(f"{len(report['rows'])} cases · generation ${report['generation_usd']:.4f} · total ${total:.4f}")
    return "\n".join(lines)

# script_id: day_030_answer_evaluation/run_answer_eval
# run_answer_eval.py: score Atlas on 30 qa_gold + 5 unanswerable questions.
# Needs rag.py and answers.py in the same folder; calls the API through them (AsyncOpenAI()).
# Env: OPENAI_API_KEY; EVAL_LIMIT=10 runs a cheap sample of qa_gold. Cost ≈ $0.07-0.10 (as of 2026-11, verify)
import asyncio
import json
import os
from pathlib import Path

from answers import UNANSWERABLE, answer_eval, format_report
from rag import KeywordIndex, load_chunks, load_qa_cases

async def main() -> None:
    limit = int(os.environ.get("EVAL_LIMIT", "0")) or None
    cases = load_qa_cases()[:limit] + UNANSWERABLE
    report = await answer_eval(cases, index=KeywordIndex(load_chunks()))
    print(format_report(report))
    out = Path("reports/answer_eval.json")
    out.parent.mkdir(exist_ok=True)
    out.write_text(json.dumps(report, indent=2), encoding="utf-8")
    print(f"saved {out}")

if __name__ == "__main__":
    asyncio.run(main())

check                 pass   rate   judge $
citations_valid     27/29    0.93         -
abstains            33/35    0.94         -
faithfulness        27/30    0.90    0.0094
answer_relevance    28/30    0.93    0.0061
context_relevance   25/30    0.83    0.0088
correctness         24/30    0.80    0.0079
35 cases · generation $0.0410 · total $0.0732
saved reports/answer_eval.json

# script_id: day_030_answer_evaluation/locate_failure
# Group failing cases by where to look next. Reads reports/answer_eval.json; no API key needed.
import json
from collections import defaultdict
from pathlib import Path

DEMO = [  # shape of a saved row, used only when there is no report yet
    {"id": "qa-001", "bucket": "ok", "checks": {"faithfulness": {"passed": True},
                                                "correctness": {"passed": False}}},
    {"id": "qa-011", "bucket": "ranking", "checks": {"context_relevance": {"passed": False}}},
    {"id": "unans-2", "bucket": "ok", "checks": {"abstains": {"passed": False}}},
]

def where_to_look(bucket: str, failed: set[str]) -> str:
    if "abstains" in failed:
        return "generation: abstain rule"
    if "context_relevance" in failed:
        return f"{bucket if bucket in ('retrieval', 'ranking') else 'retrieval'}: fix search"
    if failed & {"faithfulness", "citations_valid"}:
        return "generation: unfaithful"
    if "answer_relevance" in failed:
        return "generation: dodged"
    return "faithful but incorrect: check chunks and reference"

path = Path("reports/answer_eval.json")
rows = json.loads(path.read_text(encoding="utf-8"))["rows"] if path.exists() else DEMO
groups: dict[str, list[str]] = defaultdict(list)
for row in rows:
    failed = {name for name, r in row["checks"].items() if not r["passed"]}
    if failed:
        groups[where_to_look(row["bucket"], failed)].append(row["id"])
for place, ids in sorted(groups.items(), key=lambda kv: -len(kv[1])):
    print(f"{place:<52}{len(ids):>3}  {', '.join(ids)}")

faithful but incorrect: check chunks and reference    1  qa-001
ranking: fix search                                   1  qa-011
generation: abstain rule                              1  unans-2