cli: arborist corpus-query — local shards through the unified Corpus pipeline
Parallel to `arborist cloud query`: same corpus_query.run_query
orchestrator, only the Corpus adapter differs (SqliteShardCorpus
instead of SidecarBucketCorpus). Proves the protocol works against
both backends with identical pipeline code:
arborist query "Q" legacy 2000-line query() (untouched)
arborist cloud query "Q" corpus_query.run_query + SidecarBucketCorpus
arborist corpus-query "Q" corpus_query.run_query + SqliteShardCorpus ← NEW
Same render layer (_render_cloud_query_human), same audit_mode +
sources + capacity + timings shape across both new paths.
Caveat: single-DB mode only for now. SqliteShardCorpus.fts_body uses
`chunks_fts JOIN chunks ON rowid`, which doesn't bridge per-shard
rowid namespaces under connect_query() ATTACH-and-UNION. Multi-shard
search needs a per-shard query + merge — the next protocol method
(fts_body_per_shard) to add. Until then, --db <path> or auto-picks
the first non-system .db under --shards-dir.
Live demo (homer + virt-back queries):
make corpus-query Q="who developed virt-back?" LLM=qwen
→ EVIDENCE-WARRANTED · via claim_lattice 1/1 2.29s
(search 0.01s · llm 2.26s · verify 0.01s · total 2.29s)
Same E1 citation + render as `make cloud-query` on the same
question; only "0 HTTP requests" footer betrays local vs cloud.
Makefile: `make corpus-query Q="..." [LLM=qwen]`.
This commit is contained in:
parent
c3529ad571
commit
96417cd958
2 changed files with 141 additions and 0 deletions
10
Makefile
10
Makefile
|
|
@ -1629,6 +1629,16 @@ cloud-query: bootstrap ## bucket-direct end-to-end via manifest [Q="..." JSON=1
|
|||
$(if $(LLM_MODEL),--model $(LLM_MODEL),) \
|
||||
$(if $(JSON),--json,)
|
||||
|
||||
corpus-query: bootstrap ## LOCAL shards via the same Corpus pipeline cloud-query uses [Q="..." JSON=1 LLM=qwen|hermes TOP_K=N MAX_CONTEXT=N]
|
||||
@test -n "$(Q)" || { echo 'usage: make corpus-query Q="your question" [JSON=1] [LLM=qwen|hermes] [TOP_K=4] [MAX_CONTEXT=24000]'; exit 2; }
|
||||
$(if $(LLM_ENDPOINT),@echo "# llm: $(LLM_ENDPOINT) / $(LLM_MODEL)" >&2,)
|
||||
@$(ARBORIST) --shards-dir $(SHARDS_DIR) corpus-query '$(Q)' \
|
||||
--top-k $(or $(TOP_K),4) \
|
||||
--max-context-chars $(or $(MAX_CONTEXT),24000) \
|
||||
$(if $(LLM_ENDPOINT),--endpoint $(LLM_ENDPOINT),) \
|
||||
$(if $(LLM_MODEL),--model $(LLM_MODEL),) \
|
||||
$(if $(JSON),--json,)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Head-to-head: local query (BURN=1) vs cloud-query, same question, same data.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
131
arborist/cli.py
131
arborist/cli.py
|
|
@ -7212,6 +7212,33 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
wallet_anchor.add_argument("--server-url", required=True)
|
||||
wallet_anchor.set_defaults(func=_cmd_wallet_anchor)
|
||||
|
||||
corpus_query_cmd = sub.add_parser(
|
||||
"corpus-query",
|
||||
help=(
|
||||
"run a question against LOCAL shards through the same "
|
||||
"Corpus protocol pipeline as `cloud query` — proves that "
|
||||
"the unified retrieval/verify orchestrator (corpus_query."
|
||||
"run_query) handles both backends with the only difference "
|
||||
"being the Corpus adapter."
|
||||
),
|
||||
)
|
||||
corpus_query_cmd.add_argument("question", type=str)
|
||||
corpus_query_cmd.add_argument("--top-k", type=int, default=4)
|
||||
corpus_query_cmd.add_argument("--max-context-chars", type=int, default=24_000)
|
||||
corpus_query_cmd.add_argument(
|
||||
"--endpoint", default=None,
|
||||
help="LLM endpoint (default: $ARBORIST_LLM_ENDPOINT or Hermes).",
|
||||
)
|
||||
corpus_query_cmd.add_argument(
|
||||
"--model", default=None,
|
||||
help="LLM model id (default: $ARBORIST_LLM_MODEL or Hermes-3-8B).",
|
||||
)
|
||||
corpus_query_cmd.add_argument(
|
||||
"--json", action="store_true",
|
||||
help="emit full result as JSON. Default: human-rendered.",
|
||||
)
|
||||
corpus_query_cmd.set_defaults(func=_cmd_corpus_query)
|
||||
|
||||
cloud_cmd = sub.add_parser(
|
||||
"cloud",
|
||||
help=(
|
||||
|
|
@ -7481,6 +7508,110 @@ def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int:
|
|||
return 0
|
||||
|
||||
|
||||
def _cmd_corpus_query(args: argparse.Namespace) -> int:
|
||||
"""Local-shards query through the Corpus protocol + run_query.
|
||||
|
||||
Same orchestrator as `cloud query`, only the adapter differs
|
||||
(SqliteShardCorpus instead of SidecarBucketCorpus). When this and
|
||||
`cloud query` produce different results, the diff is pure
|
||||
retrieval-quality (sidecar BM25 + boost vs FTS5 + sanitizer), not
|
||||
pipeline drift — every prompt-building, LLM-handling, verifier,
|
||||
and render-layer change auto-applies to both.
|
||||
|
||||
Note: legacy `arborist query` (the 2000-line query() function) is
|
||||
untouched; this is the parallel path that proves the DRY landing.
|
||||
"""
|
||||
import os as _os
|
||||
from arborist.qa.client import OpenAICompatibleClient
|
||||
from arborist.qa.corpus import SqliteShardCorpus
|
||||
from arborist.qa.corpus_query import run_query
|
||||
from arborist.qa.progress import from_env as _progress_from_env
|
||||
from arborist.store import connect_query
|
||||
|
||||
progress = _progress_from_env()
|
||||
shards_dir = getattr(args, "shards_dir", None) or getattr(args, "global_shards_dir", None)
|
||||
db = getattr(args, "db", None)
|
||||
if not shards_dir and not db:
|
||||
print(
|
||||
"corpus-query needs --shards-dir or --db (global args before the "
|
||||
"subcommand): `arborist --shards-dir ~/.arborist/shards corpus-query ...`",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
# corpus-query today only supports single-DB mode because
|
||||
# SqliteShardCorpus.fts_body does `chunks_fts JOIN chunks ON rowid`,
|
||||
# which doesn't bridge per-shard rowid namespaces under connect_query
|
||||
# ATTACH-and-UNION. Multi-shard support waits on a fts_body
|
||||
# variant that runs the JOIN per attached shard and merges — that's
|
||||
# the next protocol method to add. For now: pass a single shard.
|
||||
progress.emit("corpus.open.start", mode="sqlite-shard")
|
||||
if db:
|
||||
from arborist.store import connect as _connect
|
||||
conn = _connect(db)
|
||||
else:
|
||||
# Pick the first .db in shards_dir as a pragmatic single-shard
|
||||
# default. Multi-shard query goes through legacy `arborist query`
|
||||
# for now.
|
||||
from pathlib import Path
|
||||
candidates = sorted(Path(shards_dir).glob("*.db"))
|
||||
# Prefer non-system shards (skip qa.db, snapshots.db, etc.).
|
||||
skip = {"qa.db", "snapshots.db", "selfmodel-chain.db"}
|
||||
candidates = [p for p in candidates if p.name not in skip] or candidates
|
||||
if not candidates:
|
||||
print(f"no .db files under {shards_dir}", file=sys.stderr)
|
||||
return 2
|
||||
first = candidates[0]
|
||||
print(
|
||||
f"# corpus-query: single-DB mode (first shard found: {first.name})",
|
||||
file=sys.stderr,
|
||||
)
|
||||
from arborist.store import connect as _connect
|
||||
conn = _connect(first)
|
||||
corpus = SqliteShardCorpus(conn)
|
||||
progress.emit("corpus.open.done")
|
||||
|
||||
endpoint = args.endpoint or _os.environ.get(
|
||||
"ARBORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1",
|
||||
)
|
||||
model = args.model or _os.environ.get(
|
||||
"ARBORIST_LLM_MODEL",
|
||||
"adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
|
||||
)
|
||||
chat = OpenAICompatibleClient(base_url=endpoint)
|
||||
|
||||
try:
|
||||
progress.emit("search.start", top_k=args.top_k)
|
||||
result = run_query(
|
||||
corpus, args.question, chat,
|
||||
model_id=model,
|
||||
top_k=args.top_k,
|
||||
max_context_chars=args.max_context_chars,
|
||||
)
|
||||
t = result.get("timings") or {}
|
||||
progress.emit("search.done", hits=len(result.get("sources") or []),
|
||||
ms=int((t.get("search") or 0) * 1000))
|
||||
progress.emit("llm.done",
|
||||
answer_chars=len(result.get("raw_answer") or ""),
|
||||
ms=int((t.get("llm") or 0) * 1000))
|
||||
progress.emit("verify.done",
|
||||
audit_mode=result.get("audit_mode"),
|
||||
n_verified=result.get("n_verified"),
|
||||
n_quotes=result.get("n_quotes"),
|
||||
ms=int((t.get("verify") or 0) * 1000))
|
||||
result["endpoint"] = endpoint
|
||||
result["stats"] = {} # corpus-query has no bucket stats
|
||||
result["timing_s"] = round(t.get("total") or 0, 3)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
if getattr(args, "json", False):
|
||||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||||
else:
|
||||
print(_render_cloud_query_human(result, args.question))
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
||||
"""Bucket-direct end-to-end query routed through the Corpus protocol.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue