diff --git a/Makefile b/Makefile index 7590dac..df6809b 100644 --- a/Makefile +++ b/Makefile @@ -188,11 +188,24 @@ ANSWER_MODE ?= claim_lattice # Override session-wide: LAYOUT_DEFAULT=bookend make query Q="..." LAYOUT_DEFAULT ?= tail LAYOUT ?= $(LAYOUT_DEFAULT) -query: bootstrap ## ask the corpus a question [JSON=1 BURN=1 REPAIR=1 REPROMPTS=N K="extra retrieval keywords" ANSWER_MODE=claim_lattice|claim_lattice_pointer|quote LAYOUT=tail|bookend|per_chunk BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 WITNESS=1 XLANG=1 XLANG_MT=1]; JSON by default + +# LLM endpoint toggle (shared with cloud-query — same ifeq lives down +# below; this one runs first because `query:` is defined ABOVE +# cloud-query and Make evaluates top-down). `make query LLM=qwen +# Q="..."` swaps to Qwen3.6-27B on uncloseai. LLM unset / LLM=hermes +# leaves --endpoint/--model unset so the CLI falls back to its +# built-in default (Hermes-3-8B on ai.unturf.com). +ifeq ($(LLM),qwen) +LLM_ENDPOINT ?= https://qwen.ai.unturf.com/v1 +LLM_MODEL ?= Qwen3.6-27B-UD-Q4_K_XL.gguf +endif + +query: bootstrap ## ask the corpus a question [JSON=1 BURN=1 LLM=qwen|hermes REPAIR=1 REPROMPTS=N K="extra retrieval keywords" ANSWER_MODE=claim_lattice|claim_lattice_pointer|quote LAYOUT=tail|bookend|per_chunk BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 WITNESS=1 XLANG=1 XLANG_MT=1]; JSON by default @if [ -z "$$Q" ] && [ -z "$(Q)" ]; then \ - echo "usage: make query Q=\"your question\" [JSON=1 BURN=1 REPAIR=1 REPROMPTS=N K=\"extra retrieval keywords\" ANSWER_MODE=claim_lattice|claim_lattice_pointer|quote LAYOUT=tail|bookend|per_chunk BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 WITNESS=1 XLANG=1 XLANG_MT=1]"; exit 2; \ + echo "usage: make query Q=\"your question\" [JSON=1 BURN=1 LLM=qwen|hermes REPAIR=1 REPROMPTS=N K=\"extra retrieval keywords\" ANSWER_MODE=claim_lattice|claim_lattice_pointer|quote LAYOUT=tail|bookend|per_chunk BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 WITNESS=1 XLANG=1 XLANG_MT=1]"; exit 2; \ fi - $(ARBORIST) --shards-dir $(SHARDS_DIR) query --top-k $(QUERY_TOP_K) $(if $(JSON),--json,) $(if $(BURN),--burn,) $(if $(REPAIR),--repair,) $(if $(REPROMPTS),--repair-reprompts $(REPROMPTS),) $(if $(ANSWER_MODE),--answer-mode $(ANSWER_MODE),) --user-payload-layout $(LAYOUT) $(if $(K),--retrieval-keywords "$(K)",) $(if $(BROAD),--apply-quantifier-caps,) $(if $(REJECT_BROAD),--reject-broad,) $(if $(ALLOW_BROAD),--allow-broad,) $(if $(WITNESS),--witness,) $(if $(XLANG),--crosslang-guard,) $(if $(XLANG_MT),--crosslang-translate,) "$(Q)" + $(if $(LLM_ENDPOINT),@echo "# llm: $(LLM_ENDPOINT) / $(LLM_MODEL)" >&2,) + $(ARBORIST) --shards-dir $(SHARDS_DIR) query --top-k $(QUERY_TOP_K) $(if $(LLM_ENDPOINT),--endpoint $(LLM_ENDPOINT) --model $(LLM_MODEL),) $(if $(JSON),--json,) $(if $(BURN),--burn,) $(if $(REPAIR),--repair,) $(if $(REPROMPTS),--repair-reprompts $(REPROMPTS),) $(if $(ANSWER_MODE),--answer-mode $(ANSWER_MODE),) --user-payload-layout $(LAYOUT) $(if $(K),--retrieval-keywords "$(K)",) $(if $(BROAD),--apply-quantifier-caps,) $(if $(REJECT_BROAD),--reject-broad,) $(if $(ALLOW_BROAD),--allow-broad,) $(if $(WITNESS),--witness,) $(if $(XLANG),--crosslang-guard,) $(if $(XLANG_MT),--crosslang-translate,) "$(Q)" query-dry: bootstrap ## like 'make query' but skip the LLM call (dry-run) [JSON=1 BURN=1 ANSWER_MODE=... BROAD=1 REJECT_BROAD=1 ALLOW_BROAD=1 XLANG=1 XLANG_MT=1] @if [ -z "$$Q" ] && [ -z "$(Q)" ]; then \ diff --git a/arborist/qa/corpus.py b/arborist/qa/corpus.py index a22a613..c527469 100644 --- a/arborist/qa/corpus.py +++ b/arborist/qa/corpus.py @@ -529,13 +529,22 @@ class MultiShardSqliteCorpus: """``shard_paths`` is an iterable of Path | str pointing at individual shard .db files. Each is connect()'d eagerly so the adapter is hot at construction (fts_body callers don't - pay open-cost on first query).""" + pay open-cost on first query). + + check_same_thread=False so MultiShardSqliteCorpus.fts_body + can fan out across shards in a ThreadPoolExecutor. Read-only + FTS5 queries serialize under SQLite's internal locks; the + per-connection thread-check is what blocks cross-thread use. + Without parallelism a 4-token natural-language query on the + 5-shard genesis corpus took 25 s sequentially (5 × ~6 s); + parallel fan-out drops that to ~6 s wall time. + """ import sqlite3 as _sqlite from arborist.store import connect as _connect self._shards: list[tuple[str, SqliteShardCorpus]] = [] for sp in shard_paths: - conn = _connect(sp) + conn = _connect(sp, check_same_thread=False) if conn.row_factory is None: conn.row_factory = _sqlite.Row self._shards.append((str(sp), SqliteShardCorpus(conn))) @@ -548,33 +557,44 @@ class MultiShardSqliteCorpus: pass def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]: - """Per-shard fts_body, sequential, then merge by raw score. + """Per-shard fts_body parallel fan-out, then merge by raw score. FTS5 BM25 is "lower is better" (negative range); sort ascending, take top-K. Each Hit gets shard_id annotated so chunks_for_doc can route the chunk read back to the owning shard's connection. - Sequential (not threaded): sqlite3.Connection enforces - single-thread access by default, so a ThreadPoolExecutor would - either silently fail (with except-swallow) or need - check_same_thread=False (race-prone). Per-shard FTS5 on local - SSDs is fast enough — measured ~10ms/shard on the genesis - corpus — that serial is fine. Threading-safe per-shard - connections is a future optimization if multi-shard query - latency becomes the bottleneck. + Parallelized via ThreadPoolExecutor — connections are opened + with check_same_thread=False in __init__ so each shard's + FTS5 query can run on its own thread. Multi-token natural- + language queries ("when did aliens film come out?") expand to + FTS5 OR-mode over many postings; a single shard's BM25 + evaluation took ~6s on the genesis 5-shard corpus, so serial + fan-out cost 25s+. Parallel fan-out drops that to roughly + max(per_shard) instead of sum. """ + from concurrent.futures import ThreadPoolExecutor + + def _one(item): + sh_path, sc = item + try: + return sh_path, sc.fts_body(query, limit=limit * 2) + except Exception: + return sh_path, [] + merged: list[Hit] = [] - for sh_path, sc in self._shards: - hits = sc.fts_body(query, limit=limit * 2) - for h in hits: - merged.append(Hit( - document_root=h.document_root, - document_uri=h.document_uri, - title=h.title, - score=h.score, - shard_id=sh_path, - extras=dict(h.extras), - )) + with ThreadPoolExecutor( + max_workers=max(1, len(self._shards)) + ) as ex: + for sh_path, hits in ex.map(_one, self._shards): + for h in hits: + merged.append(Hit( + document_root=h.document_root, + document_uri=h.document_uri, + title=h.title, + score=h.score, + shard_id=sh_path, + extras=dict(h.extras), + )) # Lower BM25 = stronger; sort ascending. merged.sort(key=lambda h: h.score) return merged[:limit] @@ -628,22 +648,36 @@ class MultiShardSqliteCorpus: return None def _fanout(self, method: str, limit: int, **kwargs) -> list[Hit]: - """Common per-shard fan-out + merge for title/phrase routes. + """Common per-shard parallel fan-out + merge for title/phrase + routes. Same shape as fts_body's merge: each shard returns + its top-K (in parallel), annotate shard_id, sort ASC by bm25. + """ + from concurrent.futures import ThreadPoolExecutor + + def _one(item): + sh_path, sc = item + fn = getattr(sc, method, None) + if fn is None: + return sh_path, [] + try: + return sh_path, fn(limit=limit * 2, **kwargs) + except Exception: + return sh_path, [] - Same shape as fts_body's merge: each shard returns its top-K, - we annotate shard_id, then sort ASC by bm25 (lower=better).""" merged: list[Hit] = [] - for sh_path, sc in self._shards: - hits = getattr(sc, method)(limit=limit * 2, **kwargs) - for h in hits: - merged.append(Hit( - document_root=h.document_root, - document_uri=h.document_uri, - title=h.title, - score=h.score, - shard_id=sh_path, - extras=dict(h.extras), - )) + with ThreadPoolExecutor( + max_workers=max(1, len(self._shards)) + ) as ex: + for sh_path, hits in ex.map(_one, self._shards): + for h in hits: + merged.append(Hit( + document_root=h.document_root, + document_uri=h.document_uri, + title=h.title, + score=h.score, + shard_id=sh_path, + extras=dict(h.extras), + )) merged.sort(key=lambda h: h.score) return merged[:limit] diff --git a/arborist/store.py b/arborist/store.py index 24e2957..ea832e2 100644 --- a/arborist/store.py +++ b/arborist/store.py @@ -565,7 +565,11 @@ def invalidate_migration_cache(db_path: Path | str) -> None: _MIGRATED_SHARDS.discard(str(Path(db_path).resolve())) -def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: +def connect( + db_path: Path | str = DEFAULT_DB_PATH, + *, + check_same_thread: bool = True, +) -> sqlite3.Connection: """Open a writable connection, creating the parent dir + schema if needed. Performance pragmas applied per-connection. Under WAL (set in the schema): @@ -590,7 +594,10 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: """ p = Path(db_path) p.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(p, isolation_level=None) # autocommit; we'll BEGIN manually + conn = sqlite3.connect( + p, isolation_level=None, + check_same_thread=check_same_thread, + ) # autocommit; we'll BEGIN manually conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout = 5000") # wait on a peer's write lock, don't fail-fast