qmd_py.bench¶

def load_fixture(path: str | Path) -> BenchmarkFixture:
    """Load and validate a benchmark fixture from JSON.

    Raises:
        ValueError: The file has no `queries` array.
        KeyError: A query is missing a required field.
        json.JSONDecodeError: The file is not valid JSON.
    """
    data = json.loads(Path(path).read_text())
    if not isinstance(data.get("queries"), list):
        raise ValueError("Invalid fixture: missing 'queries' array")
    queries = [
        BenchmarkQuery(
            id=q["id"],
            query=q["query"],
            type=q.get("type", ""),
            description=q.get("description", ""),
            expected_files=q["expected_files"],
            expected_in_top_k=q["expected_in_top_k"],
        )
        for q in data["queries"]
    ]
    return BenchmarkFixture(
        description=data.get("description", ""),
        version=data.get("version", 1),
        queries=queries,
        collection=data.get("collection"),
    )

def normalize_path(path: str) -> str:
    """Reduce a path to a comparable form for scoring.

    `marq://collection/docs/readme.md` becomes `docs/readme.md`:
    lowercased, stripped of the scheme, collection prefix and surrounding
    slashes, so a fixture can name files without knowing which
    collection they were indexed into.
    """
    if path.startswith("marq://"):
        without_scheme = path[len("marq://") :]
        slash_idx = without_scheme.find("/")
        path = without_scheme[slash_idx + 1 :] if slash_idx >= 0 else without_scheme
    return path.lower().strip("/")

def paths_match(result: str, expected: str) -> bool:
    """Whether a retrieved path satisfies an expected one.

    Deliberately loose: equal after normalization, or either being a
    suffix of the other. That lets a fixture write `readme.md` for a
    document actually stored at `docs/readme.md` - convenient for
    hand-written fixtures, but it can credit a near-miss, so treat scores
    as comparative rather than absolute.
    """
    nr, ne = normalize_path(result), normalize_path(expected)
    return nr == ne or nr.endswith(ne) or ne.endswith(nr)

def score_results(result_files: list[str], expected_files: list[str], top_k: int) -> ScoreMetrics:
    """Score one ranked result list against its expected files.

    Args:
        result_files: Retrieved paths, best first.
        expected_files: Paths that should have been retrieved.
        top_k: Cutoff for the @k metrics.

    Returns:
        All metrics for this query. An empty `expected_files` yields
        zeros rather than dividing by zero.
    """
    hits_at_k = _hits_within(result_files, expected_files, top_k)
    matched = [e for e in expected_files if any(paths_match(r, e) for r in result_files)]
    matched_set = set(matched)
    unmatched = [e for e in expected_files if e not in matched_set]

    mrr = 0.0
    for i, r in enumerate(result_files):
        if any(paths_match(r, e) for e in expected_files):
            mrr = 1 / (i + 1)
            break

    n_expected = len(expected_files)

    def recall_at(k: int) -> float:
        return _hits_within(result_files, expected_files, k) / n_expected if n_expected else 0.0

    denominator = min(top_k, len(expected_files))
    precision_at_k = hits_at_k / denominator if denominator > 0 else 0.0
    recall = len(matched) / n_expected if n_expected else 0.0
    recall_at_1 = recall_at(1)
    recall_at_3 = recall_at(3)
    recall_at_5 = recall_at(5)
    f1 = (
        2 * (precision_at_k * recall) / (precision_at_k + recall)
        if (precision_at_k + recall) > 0
        else 0.0
    )
    return ScoreMetrics(
        precision_at_k=precision_at_k,
        recall=recall,
        recall_at_1=recall_at_1,
        recall_at_3=recall_at_3,
        recall_at_5=recall_at_5,
        mrr=mrr,
        f1=f1,
        hits_at_k=hits_at_k,
        matched_files=matched,
        unmatched_expected_files=unmatched,
    )

async def run_benchmark(
    session: AsyncSession,
    user: CurrentUser,
    llm_client: LlmClient,
    settings: Settings,
    fixture_path: str | Path,
    *,
    collection: str | None = None,
    backend_names: list[str] | None = None,
    on_progress: Callable[[str, str, float], None] | None = None,
) -> BenchmarkResult:
    """Run every fixture query against every selected backend.

    Framework-agnostic on purpose - no click dependency - so the CLI
    layer supplies progress reporting rather than this owning output.

    Args:
        fixture_path: JSON fixture to load.
        collection: Restrict all queries to one collection, overriding
            the fixture's own setting.
        backend_names: Which backends to run (`bm25`, `vector`,
            `hybrid`, `full`). None runs all of them.
        on_progress: Called as `(query_id, backend_name, latency_ms)`
            after each backend finishes, for incremental output.

    Returns:
        Per-query results plus the averaged summary.

    Note:
        Runs real searches against real infrastructure, LLM calls
        included - a full fixture against the hybrid backends takes
        minutes, not seconds.
    """
    fixture = load_fixture(fixture_path)
    active = backend_names or list(BACKENDS.keys())
    effective_collection = collection or fixture.collection

    results: list[QueryResult] = []
    for query in fixture.queries:
        backend_results: dict[str, BackendResult] = {}
        for name in active:
            backend_result = await _run_query(
                session, user, llm_client, settings, BACKENDS[name], query, effective_collection
            )
            backend_results[name] = backend_result
            if on_progress:
                on_progress(query.id, name, backend_result.latency_ms)
        results.append(
            QueryResult(id=query.id, query=query.query, type=query.type, backends=backend_results)
        )

    summary = _compute_summary(results)
    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S")
    return BenchmarkResult(
        timestamp=timestamp, fixture=str(fixture_path), results=results, summary=summary
    )

def bench_result_to_json(result: BenchmarkResult) -> str:
    """Serialize a whole run as indented JSON, for diffing runs."""
    return json.dumps(asdict(result), indent=2)

def format_bench_table(results: list[QueryResult]) -> str:
    """Render per-query, per-backend metrics as a fixed-width table."""
    def pad(s: str, n: int) -> str:
        return s[:n].ljust(n)

    def num(n: float) -> str:
        return f"{n:.2f}".rjust(5)

    lines = [
        f"{pad('Query', 25)} {pad('Backend', 8)} {pad('P@k', 6)} {pad('R@1', 6)} "
        f"{pad('R@3', 6)} {pad('R@5', 6)} {pad('MRR', 6)} {pad('F1', 6)} {pad('ms', 8)}",
        "-" * 88,
    ]
    for r in results:
        for backend, br in r.backends.items():
            lines.append(
                f"{pad(r.id, 25)} {pad(backend, 8)} {num(br.precision_at_k)} {num(br.recall_at_1)} "
                f"{num(br.recall_at_3)} {num(br.recall_at_5)} {num(br.mrr)} {num(br.f1)} "
                f"{str(round(br.latency_ms)).rjust(7)}ms"
            )
        lines.append("")
    return "\n".join(lines)

def format_bench_summary(summary: dict[str, dict[str, float]]) -> str:
    """Render the averaged per-backend summary as a fixed-width table."""
    def pad(s: str, n: int) -> str:
        return s[:n].ljust(n)

    def num(n: float) -> str:
        return f"{n:.3f}".rjust(6)

    lines = []
    for name, s in summary.items():
        lines.append(
            f"  {pad(name, 8)} P@k={num(s['avg_precision'])} R@1={num(s['avg_recall_at_1'])} "
            f"R@3={num(s['avg_recall_at_3'])} R@5={num(s['avg_recall_at_5'])} "
            f"MRR={num(s['avg_mrr'])} F1={num(s['avg_f1'])} Avg={round(s['avg_latency_ms'])}ms"
        )
    return "\n".join(lines)