"""Export the candidate run by calling the application's own code. The gate only means something if it exercises the code that a pull request changes. So this script does not run a query of its own. It imports two functions from your application and calls them for every question in the golden set. A pull request that changes retrieval, the prompt or the model settings then changes the numbers that the gate compares. Contract (the application's own code) ------------------------------------- search(query: str, top_k: int) -> list[dict] Ranked results, best first. Each item has "id" (the document id used in the qrels), "score" (a number) and, for the generation gate, "content" (the chunk text). answer(query: str, chunks: list[dict]) -> str (optional) The application's answer, built with its real prompt and model settings. Usage ----- python eval/export_run.py \\ --search retrieval.search:search \\ --answer app.answer:answer \\ --golden baselines/golden_set.json \\ --out runs/candidate.json Outputs ------- runs/candidate.json run for ranx: {"": {"": }} runs/candidate.meta.json query count and search latency (p50 and p95, in ms) runs/candidate_cases.jsonl query, context and response, one per line (Azure AI Evaluation SDK) runs/candidate_promptfoo.jsonl the same rows in promptfoo test format (under "vars") The two cases files are written only when --answer is given. Golden set format ----------------- [{"query_id": "q1", "query": "...", "unanswerable": false}, ...] or {"q1": "...", ...} Set "unanswerable": true when the correct behaviour is to decline to answer. The application needs its usual settings (endpoint, index and so on). In CI, point them at the test index through pipeline variables. This script never reads a key. """ from __future__ import annotations import argparse import importlib import json import logging import math import sys import time from pathlib import Path from typing import Any, Callable, NoReturn logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") log = logging.getLogger("export_run") DEFAULT_TOP_K = 10 # team default: how many results the application passes to the model REFUSAL_RUBRIC = ( "The answer says it cannot answer from the provided context, or it declines to answer. " "It does not state a fact that is absent from the context." ) def fail(message: str) -> NoReturn: """Stop with exit code 2, so a setup error is not mistaken for a gate failure (exit code 1).""" log.error(message) sys.exit(2) def load_function(spec: str) -> Callable[..., Any]: """Load 'package.module:function' from the application, with the repository root on the path.""" if ":" not in spec: fail(f"'{spec}' must look like package.module:function") module_name, function_name = spec.split(":", 1) sys.path.insert(0, str(Path.cwd())) try: return getattr(importlib.import_module(module_name), function_name) except (ImportError, AttributeError) as exc: fail(f"Cannot load '{spec}': {exc}") def load_golden(path: Path) -> list[dict[str, Any]]: if not path.exists(): fail(f"Golden set not found at '{path}'. Complete the prerequisites in PB-RAG-02-04 first.") data = json.loads(path.read_text(encoding="utf-8-sig")) items: list[dict[str, Any]] = [] if isinstance(data, dict): for query_id, text in data.items(): items.append({"query_id": str(query_id), "query": str(text), "unanswerable": False}) elif isinstance(data, list): for index, item in enumerate(data, start=1): if not isinstance(item, dict) or not item.get("query"): fail(f"Golden set item {index} needs a 'query' field.") items.append( { "query_id": str(item.get("query_id") or f"q{index}"), "query": str(item["query"]), "unanswerable": bool(item.get("unanswerable", False)), } ) else: fail(f"'{path}' must hold a JSON object or array.") if not items: fail(f"The golden set '{path}' is empty.") log.info("Loaded %d questions from '%s'.", len(items), path) return items def normalise(query_id: str, results: Any) -> list[dict[str, Any]]: """Check the shape that search() returned and copy it into plain dictionaries.""" if not isinstance(results, list): fail(f"search() must return a list. For {query_id} it returned {type(results).__name__}.") chunks = [] for position, item in enumerate(results, start=1): if not isinstance(item, dict) or "id" not in item or "score" not in item: fail(f"search() result {position} for {query_id} needs 'id' and 'score'.") chunks.append( { "id": str(item["id"]), "score": float(item["score"]), "content": str(item.get("content", "")), } ) return chunks def percentile(values: list[float], fraction: float) -> float: ordered = sorted(values) return ordered[max(0, math.ceil(fraction * len(ordered)) - 1)] def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2), encoding="utf-8") def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Export the candidate run from the application.") parser.add_argument("--search", required=True, help="package.module:function for search()") parser.add_argument("--answer", help="package.module:function for answer() (generation gate)") parser.add_argument("--golden", type=Path, default=Path("baselines/golden_set.json")) parser.add_argument("--out", type=Path, default=Path("runs/candidate.json")) parser.add_argument("--meta-out", type=Path, default=Path("runs/candidate.meta.json")) parser.add_argument("--cases-out", type=Path, default=Path("runs/candidate_cases.jsonl")) parser.add_argument("--promptfoo-out", type=Path, default=Path("runs/candidate_promptfoo.jsonl")) parser.add_argument("--top-k", type=int, default=DEFAULT_TOP_K, help="team default: 10") parser.add_argument( "--skip-unanswerable", action="store_true", help="Omit questions marked unanswerable from the cases output.", ) return parser.parse_args() def main() -> None: args = parse_args() search = load_function(args.search) answer = load_function(args.answer) if args.answer else None golden = load_golden(args.golden) run: dict[str, dict[str, float]] = {} cases: list[dict[str, Any]] = [] latencies: list[float] = [] for item in golden: started = time.perf_counter() results = search(item["query"], args.top_k) latencies.append((time.perf_counter() - started) * 1000) chunks = normalise(item["query_id"], results) scores: dict[str, float] = {} for chunk in chunks: # several chunks of one document keep the best score scores[chunk["id"]] = max(chunk["score"], scores.get(chunk["id"], chunk["score"])) run[item["query_id"]] = scores if answer is not None: cases.append( { "query_id": item["query_id"], "query": item["query"], "context": "\n\n".join(c["content"] for c in chunks if c["content"]), "response": str(answer(item["query"], chunks)), "unanswerable": item["unanswerable"], } ) write_json(args.out, run) write_json( args.meta_out, { "queries": len(run), "search_latency_ms": { "p50": round(percentile(latencies, 0.5), 1), "p95": round(percentile(latencies, 0.95), 1), }, }, ) log.info("Run written to '%s' (%d questions).", args.out, len(run)) if answer is None: log.warning("No --answer function: the generation gate has nothing to judge.") return write_jsonl( args.cases_out, [ {"query": case["query"], "context": case["context"], "response": case["response"]} for case in cases if not (args.skip_unanswerable and case["unanswerable"]) ], ) promptfoo_rows = [] for case in cases: row: dict[str, Any] = { "vars": {k: case[k] for k in ("query_id", "query", "context", "response")} } if case["unanswerable"]: row["assert"] = [{"type": "llm-rubric", "value": REFUSAL_RUBRIC}] promptfoo_rows.append(row) write_jsonl(args.promptfoo_out, promptfoo_rows) log.info("Cases written to '%s' and '%s'.", args.cases_out, args.promptfoo_out) if __name__ == "__main__": main()