From 71eb264448aa81c1f2c93c9045cc529bd4829b84 Mon Sep 17 00:00:00 2001 From: oskar Date: Thu, 30 Jul 2026 16:36:39 +0200 Subject: [PATCH] =?UTF-8?q?feat(eval):=20retrieval=5Feval=20--transport=20?= =?UTF-8?q?http=20=E2=80=94=20bramka=20=C2=A79=20przez=20=C5=BCywe=20/sear?= =?UTF-8?q?ch=20kb-query=20(salvage=20S1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cherry-pick 1:1 z porzucanego brancha task/kb-f4-fallback (3d4ee38) wg decyzji z raportu dedup (docs/kb/modules/05-fallback-dedup-raport.md): plan §2 D6/§9 wymaga trybu HTTP-equivalence, a master go nie miał — e7625cd nie tknął tego pliku, patch aplikuje się czysto i woła wyłącznie GET /search (pola envelope_id/dist/source zgodne z odpowiedzią mastera). Live-PASS 2026-07-27 na kodzie brancha; smoke na masterze: CLI + pełny przebieg http przeciwko stubowi /search (raport i werdykt bramki generują się poprawnie). Co-Authored-By: Claude Fable 5 --- jobs/documents-ingest/README.md | 6 + jobs/documents-ingest/eval/retrieval_eval.py | 109 ++++++++++++++++++- 2 files changed, 111 insertions(+), 4 deletions(-) diff --git a/jobs/documents-ingest/README.md b/jobs/documents-ingest/README.md index bf3fce1..624a05f 100644 --- a/jobs/documents-ingest/README.md +++ b/jobs/documents-ingest/README.md @@ -517,6 +517,12 @@ python eval/retrieval_eval.py --dsn postgresql://kb:@piha:5433/kb \ --ollama-url http://solaria:11434 --n-sweep 5,10,20 ``` +`--transport http --base-url http://:8230` (module 5 phase 4 plan §2 decision 6 / +§9) calls a live `kb-query`'s `/search` instead of embedding+querying locally — no `--dsn` +needed, `--n-sweep` is ignored (kb-query serves one server-side default N per request). Gate +criterion: `dist` must be **identical** to the same run with `--transport direct` against the +same live SOLARIA (same DB, same retrieval code — HTTP is only a wrapper). + **Result (2026-07-17, live run)**: PASS at N=10, k=5 — see plan §6.3 for the full table, the N-sweep calibration (N=5 is the measured safety floor; the plan's N=10 default carries a 2× margin), and the cost/improvement analysis. `cascade_query` (N=10, k=5, diff --git a/jobs/documents-ingest/eval/retrieval_eval.py b/jobs/documents-ingest/eval/retrieval_eval.py index 416eab3..f0aa063 100644 --- a/jobs/documents-ingest/eval/retrieval_eval.py +++ b/jobs/documents-ingest/eval/retrieval_eval.py @@ -23,9 +23,23 @@ mocked test). Runs every query in `queries.yaml`'s `queries:` list through flat writes nothing. Query embeddings go through Ollama on localhost/SOLARIA (bge-m3), same as `chunk_embed.py`/`summarize.py`. +`--transport {direct,http}` (module 5 phase 4 plan §2 decision 6 / §9, added alongside the +fallback task): `direct` (default) is everything above, unchanged. `http` instead calls +`GET {base_url}/search?q=...&mode=flat|cascade|hybrid` on a live `kb-query` and reshapes its +JSON `results` into the same `{"chunks": [...]}` shape the direct-mode functions return, so +`summarize_query_result`/`evaluate_gate` run identically either way. The gate criterion (plan +§9): `dist` for `http` must be **identical** to `direct` against the same live SOLARIA -- same +DB, same retrieval code, HTTP is only a wrapper, so any difference is a serialization/handler +bug, never expected numerical drift. `http` mode cannot sweep `N` (kb-query serves one +server-side default per request, plan §4) -- it reports only at `--gate-n`, and needs no `--dsn` +(kb-query already owns the DB connection; `envelope.source` for criterion 4 comes straight from +each result's `source` field instead of a separate DB lookup). + Usage: python retrieval_eval.py --dsn postgresql://kb:@piha:5433/kb \\ --ollama-url http://solaria:11434 --n-sweep 5,10,20 + python retrieval_eval.py --transport http --base-url http://192.168.31.5:8230 \\ + --gate-n 10 """ from __future__ import annotations @@ -156,6 +170,44 @@ async def run_query_all_tracks( return {"query": query, "flat": flat, "cascades": cascades, "hybrid": hybrid} +async def call_search_http( + session: aiohttp.ClientSession, base_url: str, query_text: str, mode: str +) -> list[dict]: + """One `GET {base_url}/search?q=...&mode=...` call -> its `results` list. Each result + already carries `envelope_id`/`dist`/`source` -- exactly the fields `top1_dist`/`hit_at_3`/ + `mail_hit_at_3` need, no DB lookup required on this side.""" + async with session.get( + f"{base_url}/search", params={"q": query_text, "mode": mode} + ) as resp: + resp.raise_for_status() + data = await resp.json() + return data["results"] + + +async def run_query_all_tracks_http( + session: aiohttp.ClientSession, base_url: str, query: dict, gate_n: int +) -> dict: + """HTTP-transport equivalent of `run_query_all_tracks` -- three `/search` calls (one per + mode) instead of embedding+querying locally. `stage1_summaries` isn't part of the HTTP + response shape (plan §4) so it's reported empty; nothing in `evaluate_gate` reads it.""" + flat_chunks = await call_search_http(session, base_url, query["text"], "flat") + cascade_chunks = await call_search_http(session, base_url, query["text"], "cascade") + hybrid_chunks = await call_search_http(session, base_url, query["text"], "hybrid") + return { + "query": query, + "flat": {"chunks": flat_chunks}, + "cascades": {gate_n: {"chunks": cascade_chunks, "stage1_summaries": []}}, + "hybrid": {"chunks": hybrid_chunks}, + } + + +def envelope_sources_from_results(*chunk_lists: list[dict]) -> dict[str, str]: + """http transport has no DB to `fetch_envelope_sources` from -- each `/search` result + already carries its envelope's `source`, so build the same envelope_id -> source mapping + straight from the response bodies already fetched for this query.""" + return {c["envelope_id"]: c["source"] for chunks in chunk_lists for c in chunks} + + def summarize_query_result(result: dict, envelope_sources: Optional[dict[str, str]] = None) -> dict: query = result["query"] expected = query["expected_envelope"] @@ -342,7 +394,34 @@ def print_report( "hybrid = 1 embed (shared) + cascade's queries + 1 extra SQL query (mail branch).") -async def main_async(args: argparse.Namespace) -> dict: +async def main_async_http(args: argparse.Namespace) -> tuple[list[dict], list[dict], list[int]]: + """`--transport http` path -- no DB connection, three `/search` calls per query. Returns + only at `--gate-n` (see module docstring: kb-query serves one server-side N per request).""" + queries = load_queries(Path(args.queries)) + mail_queries = load_mail_queries(Path(args.queries)) + n_values = [args.gate_n] + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session: + results = [] + envelope_sources: dict[str, str] = {} + for query in queries: + result = await run_query_all_tracks_http(session, args.base_url, query, args.gate_n) + envelope_sources.update(envelope_sources_from_results( + result["flat"]["chunks"], result["cascades"][args.gate_n]["chunks"], result["hybrid"]["chunks"], + )) + results.append(summarize_query_result(result)) + + mail_results_raw = [] + for query in mail_queries: + result = await run_query_all_tracks_http(session, args.base_url, query, args.gate_n) + envelope_sources.update(envelope_sources_from_results(result["hybrid"]["chunks"])) + mail_results_raw.append(result) + mail_results = [summarize_query_result(r, envelope_sources) for r in mail_results_raw] + + return results, mail_results, n_values + + +async def main_async_direct(args: argparse.Namespace) -> tuple[list[dict], list[dict], list[int]]: queries = load_queries(Path(args.queries)) mail_queries = load_mail_queries(Path(args.queries)) n_values = [int(n) for n in args.n_sweep.split(",")] @@ -380,6 +459,21 @@ async def main_async(args: argparse.Namespace) -> dict: finally: await conn.close() + return results, mail_results, n_values + + +async def main_async(args: argparse.Namespace) -> dict: + if args.transport == "http": + if args.n_sweep != "5,10,20": # the argparse default -- operator didn't ask for a sweep + print( + "note: --transport http ignores --n-sweep (kb-query serves a single " + f"server-side default N per request); reporting only --gate-n={args.gate_n}", + file=sys.stderr, + ) + results, mail_results, n_values = await main_async_http(args) + else: + results, mail_results, n_values = await main_async_direct(args) + gate_result = evaluate_gate(results, gate_n=args.gate_n, mail_rows=mail_results) print_report(results, n_values, gate_result, mail_rows=mail_results) @@ -395,19 +489,26 @@ async def main_async(args: argparse.Namespace) -> dict: def main() -> None: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), help="asyncpg DSN for kb-postgres (or KB_DSN env var)") + parser.add_argument("--transport", choices=["direct", "http"], default="direct", + help="direct = query DB+Ollama locally (default); http = call a live kb-query's /search") + parser.add_argument("--base-url", default=None, help="kb-query base URL, required for --transport http (e.g. http://192.168.31.5:8230)") + parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), help="asyncpg DSN for kb-postgres (or KB_DSN env var); required for --transport direct") parser.add_argument("--ollama-url", default=os.environ.get("OLLAMA_URL", "http://localhost:11434")) parser.add_argument("--embed-model", default=DEFAULT_EMBED_MODEL) parser.add_argument("--summary-model", default=DEFAULT_SUMMARY_MODEL, help="document_summary.model to pre-filter on (plan §2 D3 resolution)") parser.add_argument("--k", type=int, default=DEFAULT_K) parser.add_argument("--gate-n", type=int, default=DEFAULT_N, help="N used for the PASS/FAIL verdict") - parser.add_argument("--n-sweep", default="5,10,20", help="comma-separated N values to report (diagnostic)") + parser.add_argument("--n-sweep", default="5,10,20", help="comma-separated N values to report (diagnostic; ignored by --transport http)") parser.add_argument("--queries", default=str(DEFAULT_QUERIES_PATH)) parser.add_argument("--json-out", default=None, help="optional path to dump full results as JSON") args = parser.parse_args() - if not args.dsn: + if args.transport == "http": + if not args.base_url: + print("error: --transport http requires --base-url", file=sys.stderr) + sys.exit(1) + elif not args.dsn: print("error: pass --dsn or set KB_DSN", file=sys.stderr) sys.exit(1)