wallet/fts-sidecar: real FTS5 in the cloud sidecar; delete custom BM25
Replaces the hand-rolled BM25 sidecar (arborist/wallet/sidecar.py, ~690 LOC) with a slim SQLite file that just COPIES the source shard's FTS5 shadow tables verbatim + minimal doc/chunk metadata. Cloud retrieval then runs SQLite FTS5 bm25() on the same bytes the local shard uses — bit-for-bit parity by construction. 5/5 source + audit_mode agreement on the smoke fixture between local corpus-query and cloud-query against the new manifest-fts.json. Why --- Custom binary sidecar was per-document BM25; main encyclopedia articles got length-normalized so hard that on "why did the dinosaurs go extinct?" "Edwina, the Dinosaur Who Didn't Know She Was Extinct" beat "Dinosaur" (measured cloud-vs-local divergence). Local FTS5 indexes per-chunk so each chunk is a moderate-length doc and the main article wins multiply. Different granularity, not a tuning knob — fix is to use the same indexer cloud-side. What ships ---------- - arborist/wallet/fts_sidecar_build.py — builder. ATTACH source shard, copy documents (root/uri/title only), copy chunks (id/root/idx/leaf only, NO content), CREATE VIRTUAL TABLE chunks_fts/documents_fts with same DDL as source, bulk-copy the four shadow tables verbatim, VACUUM. 8.78 GB shard → 2.15 GB sidecar (24.5%) in ~45 s; full 4-shard wiki corpus 37.4 GB → 8.1 GB (21.7%) in ~3 min. - arborist/wallet/bucket.py: FtsSidecarShardClient — downloads slim sidecar once into ~/.arborist/sidecar-fts-cache/<hash>.idx.db, opens read-only sqlite3 (check_same_thread=False for parallel shard fan- out), runs FTS5 MATCH locally. Chunk content fetches via blobs/<hash> with HTTP-range big-shard fallback when blobs aren't published. MultiShardSidecarCorpus simplified to fts_sidecar_url ∨ bucket-direct (both are FTS5 backends; merge by raw bm25 MIN ascending). - arborist/qa/corpus.py: SidecarBucketCorpus.higher_is_better=False (FTS5 bm25 is negative, lower=better). chunks_for_doc dispatches on fetch_chunk_body attr for the slim-FTS5 client. apply_title_boost imports tokenizer helpers from new arborist/qa/_text_norm.py. - arborist/qa/_text_norm.py — fold_accents, numeral_expand, tokenize_text, STOPWORDS — extracted from the deleted sidecar.py so apply_title_boost keeps its lexical shape. - arborist/cli.py: `arborist sidecar build-fts` subcommand; old `sidecar build`/`sidecar search` removed. cloud_query recognizes fts_sidecar_url + sidecar_url alike. - Makefile: `sidecar-build-fts` + `sidecar-build-fts-all` targets; `sidecar-build` + `sidecar-search` removed. - scripts/upload_fts_sidecars.py — boto3 producer: uploads slim sidecars to clones/sidecars-fts/<n>.idx.db, publishes clones/manifest-fts.json (4 wikipedia shards inherit existing shard_url for content fallback; ACL public-read; idempotent on size match). Existing manifest-sidecar.json untouched. - tests/test_qa_corpus_functional.py + test_qa_corpus_integration.py converted from build_sidecar → build_fts_sidecar; 6 fixtures pass. - bench/slim_fts_parity_bench.py — local 3-way bench (legacy/corpus/slim_fts) over the smoke fixture. Validation ---------- - Per-shard FTS5 parity: slim sidecar returns IDENTICAL rowids + bm25 scores to the source shard for top-10 of "dinosaurs extinct". - 3-way bench (legacy local / corpus local / slim-FTS5 over real bucket fallback): 5/5 source agreement AND 5/5 audit_mode agreement between corpus and slim_fts. Q5 legacy disagreement (Edwina vs Dinosaur) is the pre-existing 2000-line query() retrieval quirk, unrelated. - End-to-end cloud query against published manifest-fts.json (cold- start, ~149 s sidecar download once): STRICT · Dinosaur, every quote verified (2/2). - Full pytest suite: 2737 passed, 28 skipped, 1 xfailed. One pre- existing failure (tests/test_doc_counts.py — claim_pack docs row- count drift) and one pre-existing cold_object failure, both reproduce on main HEAD. Bucket state ------------ - s3://arborist/clones/sidecars-fts/00[0-3].idx.db (8.1 GB) — new - s3://arborist/clones/manifest-fts.json — new - s3://arborist/clones/manifest-sidecar.json — kept live (deprecated but still readable; downstream callers should switch to manifest-fts.json)
This commit is contained in:
parent
2eea5b5655
commit
d9fb6a9b69
13 changed files with 1087 additions and 984 deletions
263
bench/slim_fts_parity_bench.py
Normal file
263
bench/slim_fts_parity_bench.py
Normal file
|
|
@ -0,0 +1,263 @@
|
|||
"""Slim FTS5 sidecar parity bench — does the new cloud path agree with local?
|
||||
|
||||
For each smoke fixture question, run:
|
||||
legacy: arborist query (current local path)
|
||||
corpus: arborist corpus-query (local shards via Corpus protocol)
|
||||
slim-fts: FtsSidecarShardClient + run_query (NEW cloud path, exercised
|
||||
locally by mocking the manifest with file:// sidecars + HTTP
|
||||
chunk-content fallback to the real bucket .db)
|
||||
|
||||
This proves the slim FTS5 cloud path produces the same primary-source
|
||||
pick as the local paths BEFORE we upload anything. If slim-fts matches
|
||||
corpus bit-for-bit (or better, matches legacy too), the bucket upload is
|
||||
safe to do.
|
||||
|
||||
Sidecars are expected at ~/.arborist/sidecar-fts/00*.idx.db. Build with:
|
||||
make sidecar-build-fts-all
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as _dt
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
ARBORIST = REPO / ".venv" / "bin" / "arborist"
|
||||
DEFAULT_SHARDS = Path.home() / ".arborist" / "shards"
|
||||
DEFAULT_SIDECAR_FTS = Path.home() / ".arborist" / "sidecar-fts"
|
||||
DEFAULT_FIXTURE = REPO / "bench" / "qa_questions_smoke.txt"
|
||||
DEFAULT_QWEN_ENDPOINT = "https://qwen.ai.unturf.com/v1"
|
||||
DEFAULT_QWEN_MODEL = "Qwen3.6-27B-UD-Q4_K_XL.gguf"
|
||||
DEFAULT_BUCKET_BLOB_BASE = "https://nyc3.digitaloceanspaces.com/arborist/blobs"
|
||||
DEFAULT_LIVE_MANIFEST = (
|
||||
"https://nyc3.digitaloceanspaces.com/arborist/clones/manifest-sidecar.json"
|
||||
)
|
||||
|
||||
|
||||
def _load_fixture(path: Path) -> list[str]:
|
||||
out = []
|
||||
for line in path.read_text().splitlines():
|
||||
s = line.strip()
|
||||
if s and not s.startswith("#"):
|
||||
out.append(s)
|
||||
return out
|
||||
|
||||
|
||||
def _run_cli(cmd: list[str], *, timeout_s: int) -> dict:
|
||||
t0 = time.time()
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=timeout_s,
|
||||
env={**os.environ, "ARBORIST_PROGRESS": "0"},
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"_error": f"timeout {timeout_s}s", "_elapsed_s": timeout_s}
|
||||
dt = time.time() - t0
|
||||
if proc.returncode != 0:
|
||||
return {"_error": f"exit {proc.returncode}",
|
||||
"_stderr": proc.stderr[:300], "_elapsed_s": round(dt, 2)}
|
||||
try:
|
||||
d = json.loads(proc.stdout)
|
||||
except json.JSONDecodeError as e:
|
||||
return {"_error": f"parse: {e}", "_elapsed_s": round(dt, 2)}
|
||||
d["_elapsed_s"] = round(dt, 2)
|
||||
return d
|
||||
|
||||
|
||||
def _legacy(q, *, shards_dir):
|
||||
return _run_cli([
|
||||
str(ARBORIST), "--shards-dir", str(shards_dir),
|
||||
"query", "--top-k", "8", "--json", "--burn",
|
||||
"--answer-mode", "claim_lattice", q,
|
||||
], timeout_s=120)
|
||||
|
||||
|
||||
def _corpus(q, *, shards_dir, endpoint, model):
|
||||
return _run_cli([
|
||||
str(ARBORIST), "--shards-dir", str(shards_dir),
|
||||
"corpus-query", q,
|
||||
"--top-k", "4", "--json",
|
||||
"--endpoint", endpoint, "--model", model,
|
||||
], timeout_s=120)
|
||||
|
||||
|
||||
def _slim_fts(
|
||||
q, *, sidecar_dir, live_manifest_url, endpoint, model,
|
||||
):
|
||||
"""Slim FTS5 path: load the LIVE manifest (so chunk-content fallback
|
||||
points at the real big-shard URLs) then splice in
|
||||
``fts_sidecar_url=file://...`` per shard so the slim FTS5 client
|
||||
opens locally without a download.
|
||||
|
||||
Pre-symlinks the file:// URL into the FTS-sidecar cache to avoid a
|
||||
duplicate 2 GB copy through urllib.
|
||||
"""
|
||||
from arborist.wallet.bucket import (
|
||||
load_bucket_manifest, MultiShardSidecarCorpus,
|
||||
_fts_sidecar_cache_path,
|
||||
)
|
||||
from arborist.qa.corpus import SidecarBucketCorpus
|
||||
from arborist.qa.client import OpenAICompatibleClient
|
||||
from arborist.qa.corpus_query import run_query
|
||||
|
||||
manifest = load_bucket_manifest(live_manifest_url)
|
||||
# Splice fts_sidecar_url per shard by matching basename (e.g. 000.db
|
||||
# → 000.idx.db); leave shard ``url`` + ``blob_base`` as the live
|
||||
# values so the missing-blob fallback hits the real big shard.
|
||||
for sh in manifest.shards:
|
||||
basename = sh["url"].rsplit("/", 1)[-1] # "000.db"
|
||||
n = basename.rsplit(".", 1)[0] # "000"
|
||||
sidecar_path = sidecar_dir / f"{n}.idx.db"
|
||||
if not sidecar_path.exists():
|
||||
continue
|
||||
fts_sidecar_url = f"file://{sidecar_path}"
|
||||
cache_path = _fts_sidecar_cache_path(fts_sidecar_url)
|
||||
if not os.path.exists(cache_path):
|
||||
os.symlink(sidecar_path, cache_path)
|
||||
sh["fts_sidecar_url"] = fts_sidecar_url
|
||||
# Drop shards without a slim sidecar (e.g. virtback) — parity bench
|
||||
# only exercises the four wikipedia shards we built.
|
||||
manifest.shards = [
|
||||
sh for sh in manifest.shards if sh.get("fts_sidecar_url")
|
||||
]
|
||||
|
||||
t0 = time.time()
|
||||
multi = MultiShardSidecarCorpus(manifest)
|
||||
corpus = SidecarBucketCorpus(multi)
|
||||
|
||||
client = OpenAICompatibleClient(base_url=endpoint)
|
||||
result = run_query(
|
||||
corpus, q, client,
|
||||
model_id=model, top_k=4, max_context_chars=24_000,
|
||||
)
|
||||
multi.close()
|
||||
elapsed = time.time() - t0
|
||||
|
||||
out = dict(result)
|
||||
out["_elapsed_s"] = round(elapsed, 2)
|
||||
return out
|
||||
|
||||
|
||||
def _primary(r: dict) -> dict:
|
||||
srcs = r.get("sources") or []
|
||||
if not srcs:
|
||||
return {"title": "", "uri": ""}
|
||||
pri = next(
|
||||
(s for s in srcs if s.get("source_role") == "primary_answer_source"),
|
||||
srcs[0],
|
||||
)
|
||||
return {"title": (pri.get("title") or "")[:55],
|
||||
"uri": pri.get("document_uri") or ""}
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE)
|
||||
p.add_argument("--shards-dir", type=Path, default=DEFAULT_SHARDS)
|
||||
p.add_argument("--sidecar-dir", type=Path, default=DEFAULT_SIDECAR_FTS)
|
||||
p.add_argument("--endpoint", default=DEFAULT_QWEN_ENDPOINT)
|
||||
p.add_argument("--model", default=DEFAULT_QWEN_MODEL)
|
||||
p.add_argument("--live-manifest", default=DEFAULT_LIVE_MANIFEST,
|
||||
help="real bucket manifest URL — provides the shard "
|
||||
"URLs used for the chunk-content fallback")
|
||||
p.add_argument(
|
||||
"--out-dir", type=Path,
|
||||
default=REPO / "bench" / "slim_fts_parity_results",
|
||||
)
|
||||
p.add_argument("--skip-legacy", action="store_true")
|
||||
p.add_argument("--skip-corpus", action="store_true")
|
||||
args = p.parse_args()
|
||||
|
||||
questions = _load_fixture(args.fixture)
|
||||
args.out_dir.mkdir(parents=True, exist_ok=True)
|
||||
stamp = _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ")
|
||||
out_path = args.out_dir / f"{stamp}.jsonl"
|
||||
|
||||
print(f"# {len(questions)} questions from {args.fixture}", file=sys.stderr)
|
||||
print(f"# shards: {args.shards_dir}", file=sys.stderr)
|
||||
print(f"# sidecars-fts: {args.sidecar_dir}", file=sys.stderr)
|
||||
print(f"# live manifest: {args.live_manifest}", file=sys.stderr)
|
||||
print(f"# llm: {args.endpoint} / {args.model}", file=sys.stderr)
|
||||
print(file=sys.stderr)
|
||||
|
||||
n_agree = 0
|
||||
n_disagree = 0
|
||||
|
||||
with open(out_path, "w") as f:
|
||||
for i, q in enumerate(questions, 1):
|
||||
print(f"[{i}/{len(questions)}] {q}", file=sys.stderr)
|
||||
legacy = {} if args.skip_legacy else _legacy(
|
||||
q, shards_dir=args.shards_dir)
|
||||
corpus = {} if args.skip_corpus else _corpus(
|
||||
q, shards_dir=args.shards_dir,
|
||||
endpoint=args.endpoint, model=args.model)
|
||||
slim = _slim_fts(
|
||||
q,
|
||||
sidecar_dir=args.sidecar_dir,
|
||||
live_manifest_url=args.live_manifest,
|
||||
endpoint=args.endpoint, model=args.model,
|
||||
)
|
||||
|
||||
row = {"question": q}
|
||||
for label, r in (
|
||||
("legacy", legacy), ("corpus", corpus), ("slim_fts", slim),
|
||||
):
|
||||
row[label] = {
|
||||
"audit_mode": r.get("audit_mode"),
|
||||
"n_verified": r.get("n_verified"),
|
||||
"n_quotes": r.get("n_quotes"),
|
||||
"primary": _primary(r),
|
||||
"elapsed_s": r.get("_elapsed_s") or (
|
||||
r.get("timings") or {}).get("total"),
|
||||
"error": r.get("_error") or r.get("error"),
|
||||
}
|
||||
uris = {
|
||||
lbl: row[lbl]["primary"]["uri"]
|
||||
for lbl in ("legacy", "corpus", "slim_fts")
|
||||
if row[lbl]["primary"]["uri"]
|
||||
}
|
||||
distinct = set(uris.values())
|
||||
row["agreement"] = {
|
||||
"n_paths_with_source": len(uris),
|
||||
"distinct_sources": len(distinct),
|
||||
"all_agree": len(distinct) <= 1,
|
||||
}
|
||||
if row["agreement"]["all_agree"]:
|
||||
n_agree += 1
|
||||
else:
|
||||
n_disagree += 1
|
||||
|
||||
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||||
f.flush()
|
||||
|
||||
for lbl in ("legacy", "corpus", "slim_fts"):
|
||||
r = row[lbl]
|
||||
badge = (
|
||||
" ←" if r["primary"]["uri"]
|
||||
and len(distinct) > 1
|
||||
and r["primary"]["uri"] != min(uris.values()) else ""
|
||||
)
|
||||
err = f" ERR={r['error']}" if r["error"] else ""
|
||||
print(f" {lbl:9s} {(r['audit_mode'] or '?'):<22} "
|
||||
f"{r['primary']['title']}{badge}{err}",
|
||||
file=sys.stderr)
|
||||
if not row["agreement"]["all_agree"]:
|
||||
print(f" SOURCE DISAGREEMENT: {len(distinct)} distinct primaries",
|
||||
file=sys.stderr)
|
||||
print(file=sys.stderr)
|
||||
|
||||
print(f"=== SUMMARY ===", file=sys.stderr)
|
||||
print(f" {len(questions)} questions", file=sys.stderr)
|
||||
print(f" {n_agree} all-paths agree on primary source", file=sys.stderr)
|
||||
print(f" {n_disagree} paths-disagree", file=sys.stderr)
|
||||
print(f" results: {out_path}", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue