arborist/aborist/qa/query.py
russell@unturf.com 0192255f7d
three relevance signals: title + TF-IDF cores + body density
The "permacomputer" query fell into title-relevance's empty-fallback
because "permacomputer" appears nowhere in any title. Fallback was
top-1 by raw BM25, getting one tangential source. Same problem for
any user-coined term: title-LIKE alone can't find it.

_filter_by_title_relevance now grows three accept paths:

1. Title-token overlap (existing).
2. TF-IDF core keyword overlap. A doc's distinctive low-frequency
   terms get distilled into its tfidf-keywords-v1 core. A query for a
   rare term ("permacomputer", "unturf") hits docs whose core lists
   it as a keyword even when no title does.
3. Body density. Docs that mention a query token >= 3 times pass
   regardless of title or core. Cheap proxy for "actually about it."

Empty-all-three fallback widens from top-1 to top-N (default 5) — the
LLM gets enough context to honestly say "I don't know" rather than
fabricating from a single tangential source.

Implementation:
- _docs_with_core_keyword_match: SQL join chunks -> documents -> derivations
  finds source docs whose TF-IDF core content matches a token. Folded
  into the per-shard search loop alongside FTS5 + title-LIKE.
- _body_density_passes: counts case-insensitive occurrences in
  concatenated chunk content. Threshold 3.
- _search_corpus stashes core_match_roots + root_to_shard on the
  returned hits list as side-channel info for _rerank.

Makefile: distill-shards-tfidf-parallel for the per-shard TF-IDF run.

Demo on the live cluster (after TF-IDF distill of 95k cur surface docs):

  Q "tell me about permacomputer ?"
    sources: MOAD Cheat Sheet + Unturf AI Infrastructure Expansion
    answer: synthesized from both, bounds coverage honestly

  Q "what is unturf about ?"
    sources: Unturf Trademark + Unturf AI Infrastructure Expansion
    answer: "grassroots, community-driven initiative ... ethical AI
             hardware, software, and networks"

Both queries now find user-coined terms via TF-IDF cores even though
the terms never appear in any title.

79 tests passing.
2026-04-27 13:39:29 -04:00

683 lines
23 KiB
Python

"""Multi-source corpus Q&A.
Pose a question, the tree finds related cached docs, assembles them as context,
asks Hermes, caches the answer.
The flow:
1. FTS5 search across all shards (chunks_fts can't be UNION'd in views,
so we query each shard's index independently and merge by score).
2. Title-boost rerank: hits whose title contains query tokens get a
score bump. Title is a strong topical signal that BM25 alone misses
(BM25 favors short docs with rare body tokens — Tell_(poker) outranks
Back_to_the_Future without a title boost).
3. Pick top-K distinct documents within a character budget.
4. Compute `context_root` = Merkle root over the sorted source
document_roots — that's the "source" dimension of v9.8's 8-dim
cache_key for this multi-source answer.
5. Cache lookup; hit returns STRICT immediately.
6. Miss calls Hermes via the OpenAI-compatible client.
7. Persist record with merkle_proof = {context_root, sources: [...]}
so a verifier can recompute context_root and check each source
document_root against its shard.
Per-source proofs are not bundled here (the source roots themselves are
already content-addressed). A verifier asks the shards for any specific
chunk's proof on demand.
"""
from __future__ import annotations
import json
import re
import time
from dataclasses import dataclass
from pathlib import Path
from aborist import (
CANONICALIZATION_VERSION,
CHUNKING_VERSION,
SCHEMA_VERSION,
)
from aborist.merkle import MerkleTree
from aborist.qa.client import ChatClient
from aborist.qa.concepts import (
has_compare_phrasing,
rivalry_excluded,
synonym_expand,
)
from aborist.qa.keys import (
cache_key,
conversation_hash,
governance_policy_hash,
model_profile_hash,
question_hash,
)
from aborist.search import FTS5Backend
from aborist.store import (
append_audit,
connect,
discover_shards,
transaction,
)
_TITLE_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*")
# Kept in sync with FTS5 stopwords in aborist.search.fts5 — both filter
# the same set of question-shaping words. "tell" leaking into title-LIKE
# search caused "tell me about permacomputer" to pull Tell_(poker), the
# Tell-Tale_Heart movie, Tell_City Indiana, etc.
_TITLE_STOPWORDS = frozenset(
"""
the a an is are was were be been being of to in on at for with by from
as about into through during and or but not no nor so yet too very also just
what who where when why how which this that these those such i you he she
it we they me him her us them do does did have has had can could should
would will may might
tell show describe explain summarize say give list find make please
""".split()
)
def _title_query_tokens(s: str) -> set[str]:
return {
t.lower()
for t in _TITLE_TOKEN_RE.findall(s)
if t.lower() not in _TITLE_STOPWORDS and len(t) > 1
}
def _rerank_by_title(
hits: list,
question: str,
boost: float = 10.0,
) -> list:
"""Boost hits whose title overlaps query tokens. Pure ordering aid."""
qtokens = _title_query_tokens(question)
if not qtokens:
return hits
for h in hits:
if not h.title:
continue
ttokens = _title_query_tokens(h.title.replace("_", " "))
overlap = len(qtokens & ttokens)
if overlap:
h.score += overlap * boost
hits.sort(key=lambda h: -h.score)
return hits
def _filter_by_title_relevance(
hits: list,
question: str,
*,
core_match_roots: set[str] | None = None,
body_density_check: callable | None = None,
fallback_top_n: int = 5,
) -> list:
"""Concept-aware relevance filter with three accept paths:
1. Title-token overlap (after synonym expansion). Strongest signal.
2. TF-IDF core keyword overlap — `core_match_roots` is a precomputed
set of source document_roots whose TF-IDF cores contain query
tokens. Closes the gap for neologisms like "permacomputer" that
never appear in titles but are distinctive enough to be TF-IDF
keywords of conversation bodies.
3. Body density — docs mentioning the query token >= N times pass
even without title or core match. Cheap proxy for "actually about
the topic." `body_density_check(hit)` returns bool.
Rivalry exclusion (Intel-titled docs in AMD queries) still applies on
every accept path.
If all three accept paths together produce nothing, fall back to the
top `fallback_top_n` body-BM25 hits — the LLM gets enough context to
say "I don't know" rather than fabricating from a single tangential
source.
"""
qtokens = _title_query_tokens(question)
if not qtokens:
return hits
accept = synonym_expand(qtokens)
exclude = rivalry_excluded(qtokens, compare_phrasing=has_compare_phrasing(question))
core_roots = core_match_roots or set()
kept: list = []
for h in hits:
ttokens = _title_query_tokens(h.title.replace("_", " ")) if h.title else set()
if exclude & ttokens:
continue # rivalry: opposing-side title, drop it
if accept & ttokens:
kept.append(h)
continue
if h.document_root in core_roots:
kept.append(h)
continue
if body_density_check is not None and body_density_check(h):
kept.append(h)
continue
if not kept:
return hits[: max(1, fallback_top_n)] if hits else []
return kept
DEFAULT_QUERY_POLICY = {
"system_prompt": (
"You are answering a question using ONLY the sources provided below. "
"Each source is delimited by '=== Source: <URI> ===' headers.\n\n"
"ATTRIBUTION RULES:\n"
"1. A single source may discuss MULTIPLE products, companies, or "
"competitors. Read carefully and only attribute facts to the entity "
"the source explicitly names for that fact.\n"
"2. If the question asks about ONE specific entity (e.g., 'fastest "
"AMD CPU'), do NOT include facts about competitors (Intel, Pentium) "
"as if they were facts about the asked entity. They are different "
"products even when discussed in the same article.\n"
"3. When referencing model numbers like 'Athlon XP 3200+', remember "
"that AMD's PR-rating numbers (3200+, 2500+) are NOT the clock speed "
"in MHz. Quote the source's own wording for clock speeds rather than "
"interpreting model numbers.\n"
"4. Cite which source supports each claim by quoting it inline.\n"
"5. If the sources do not contain the answer, say 'I don't know "
"based on the provided sources.' Do not speculate beyond what the "
"sources say."
),
"temperature": 0.1,
"top_p": 1.0,
"max_tokens": 768,
}
@dataclass
class _Hit:
document_root: str
document_uri: str
title: str | None
score: float
shard_path: str
chunk_idx: int
def _search_titles(conn, qtokens: list[str], limit: int) -> list[tuple]:
"""SQL LIKE over documents.title — finds the HTTP article that FTS5
misses because list-pages with many URLs have higher 'http' term
frequency than the actual protocol article. Returns rows shaped
to match the FTS5 hit tuple."""
if not qtokens:
return []
clauses = " OR ".join(["LOWER(title) LIKE ?"] * len(qtokens))
params = [f"%{t.lower()}%" for t in qtokens]
params.append(limit)
rows = conn.execute(
f"SELECT document_root, document_uri, title FROM documents "
f"WHERE {clauses} LIMIT ?",
params,
).fetchall()
return rows
def _docs_with_core_keyword_match(
conn, qtokens: list[str], limit: int
) -> list[tuple]:
"""Find SOURCE docs whose TF-IDF core keywords contain a query token.
TF-IDF cores act as enriched titles: a doc's distinctive low-frequency
terms get distilled into the core's content as comma-separated keywords.
A query for "permacomputer" (a neologism that never makes it into a
title) can match the TF-IDF core of a Grok conversation that mentioned
it, because permacomputer is a rare term that ranks high under TF-IDF.
Returns rows of (document_root, document_uri, title) for the SOURCE docs
(not the cores) — the source is what gets fed to the LLM as context.
"""
if not qtokens:
return []
clauses = " OR ".join(["LOWER(c.content) LIKE ?"] * len(qtokens))
params = [f"%{t.lower()}%" for t in qtokens]
params.append(limit)
rows = conn.execute(
f"""
SELECT DISTINCT src.document_root, src.document_uri, src.title
FROM chunks c
JOIN documents core ON core.document_root = c.document_root
JOIN derivations der ON der.core_root = core.document_root
JOIN documents src ON src.document_root = der.src_root
WHERE core.source_type LIKE 'core:tfidf-%'
AND ({clauses})
LIMIT ?
""",
params,
).fetchall()
return rows
def _body_density_passes(
conn, document_root: str, qtokens: set[str], min_mentions: int = 3
) -> bool:
"""Body-token relevance: docs that mention a query token >= N times pass.
Cheap proxy for "this doc is actually about the topic" when title
doesn't help. Counts case-insensitive substring matches in concatenated
chunk content. min_mentions=3 skips docs that mention the term in
passing without being topical.
"""
if not qtokens:
return False
rows = conn.execute(
"SELECT content FROM chunks WHERE document_root = ? AND content IS NOT NULL",
(document_root,),
).fetchall()
if not rows:
return False
body = " ".join(r["content"] for r in rows).lower()
total_mentions = 0
for t in qtokens:
total_mentions += body.count(t.lower())
if total_mentions >= min_mentions:
return True
return False
def _search_corpus(
shards_dir: Path | None,
single_db: Path | None,
question: str,
over_fetch: int,
) -> list[_Hit]:
"""Two parallel searches across shards, merged:
- FTS5 body search (BM25 over chunk content).
- SQL title-LIKE search (catches articles whose body is short on the
query terms but whose title is the literal topic — e.g., the HTTP
protocol article doesn't out-frequency-score URL-heavy list pages
but it IS the topic).
Dedupe by document_root. Title hits get a baseline score that
out-ranks FTS5 body hits so the actual topic article rises to the top.
"""
qtokens = _title_query_tokens(question)
# Synonym expansion: a query for "athlon" also fetches AMD-titled docs.
accept_tokens = synonym_expand(qtokens)
paths: list[Path]
if shards_dir is not None:
paths = discover_shards(shards_dir)
elif single_db is not None:
paths = [Path(single_db)]
else:
paths = []
raw: list[tuple] = []
# Roots whose TF-IDF cores contain a query token — collected across
# shards. Used downstream by _filter_by_title_relevance as accept-path 2.
core_match_roots: set[str] = set()
# Per-shard mapping of doc root -> shard path, for body-density lookups
# in accept-path 3. Lets us reach back to the source shard cheaply.
root_to_shard: dict[str, str] = {}
for p in paths:
conn = connect(p)
try:
backend = FTS5Backend(conn)
for h in backend.search(question, limit=over_fetch):
raw.append(
(
h.score,
h.document_root,
h.document_uri,
h.title,
h.chunk_idx,
str(p.resolve()),
)
)
root_to_shard[h.document_root] = str(p.resolve())
# Parallel title search using synonym-expanded tokens.
for r in _search_titles(conn, list(accept_tokens), over_fetch):
title_lower = (r["title"] or "").lower().replace("_", " ")
overlap = sum(1 for t in accept_tokens if t in title_lower)
if overlap == 0:
continue
title_score = 50.0 + overlap * 10.0
raw.append(
(
title_score,
r["document_root"],
r["document_uri"],
r["title"],
0,
str(p.resolve()),
)
)
root_to_shard[r["document_root"]] = str(p.resolve())
# Core-keyword search: docs whose TF-IDF core keywords match.
# The query token doesn't need to be in title or even in body —
# being a TF-IDF keyword of the doc's core is enough signal.
for r in _docs_with_core_keyword_match(
conn, list(accept_tokens), over_fetch
):
core_match_roots.add(r["document_root"])
# Add as a hit too — give it a moderate score below title
# exact-match but above raw body BM25.
raw.append(
(
40.0,
r["document_root"],
r["document_uri"],
r["title"],
0,
str(p.resolve()),
)
)
root_to_shard[r["document_root"]] = str(p.resolve())
finally:
conn.close()
raw.sort(key=lambda r: -r[0])
seen: set[str] = set()
out: list[_Hit] = []
for score, root, uri, title, idx, sp in raw:
if root in seen:
continue
seen.add(root)
out.append(
_Hit(
document_root=root,
document_uri=uri,
title=title,
score=score,
shard_path=sp,
chunk_idx=idx,
)
)
return out
def _rerank(
hits: list[_Hit],
question: str,
*,
core_match_roots: set[str] | None = None,
body_density_check: callable | None = None,
) -> list[_Hit]:
"""Filter out off-topic, then boost title-token matches by overlap count."""
hits = _filter_by_title_relevance(
hits,
question,
core_match_roots=core_match_roots,
body_density_check=body_density_check,
)
return _rerank_by_title(hits, question)
def _load_doc_text(shard_path: str, document_root: str) -> str | None:
"""Concatenate all hot chunks of a document. Returns None if cold or missing."""
conn = connect(shard_path)
try:
rows = conn.execute(
"SELECT content FROM chunks "
"WHERE document_root = ? AND content IS NOT NULL "
"ORDER BY idx ASC",
(document_root,),
).fetchall()
finally:
conn.close()
if not rows:
return None
return "\n\n".join(r["content"] for r in rows)
def _context_root(source_roots: list[str]) -> str:
"""Merkle root over sorted source document_roots — the v9.8 'source' dim
for multi-source answers. Sorting makes the root deterministic regardless
of search ranking order."""
if not source_roots:
return "00" * 32
sorted_roots = sorted(source_roots)
if len(sorted_roots) == 1:
return sorted_roots[0]
leaves = [bytes.fromhex(r) for r in sorted_roots]
return MerkleTree.build(leaves).root.hex()
def query(
*,
question: str,
qa_db: Path,
chat_client: ChatClient,
model_id: str,
revision: str = "",
quantization: str = "",
shards_dir: Path | None = None,
single_db: Path | None = None,
top_k: int = 8,
over_fetch: int = 32,
max_context_chars: int = 60000,
policy: dict | None = None,
chain: str = "private",
) -> dict:
"""Answer `question` using the corpus. Cache to qa_db. Returns a result dict."""
policy = policy or DEFAULT_QUERY_POLICY
t_start = time.monotonic()
# 1. Search.
t_phase = time.monotonic()
hits = _search_corpus(shards_dir, single_db, question, over_fetch)
if not hits:
return {
"status": "no_sources",
"msg": "FTS5 search returned no hits",
"timings": {
"search_ms": _ms_since(t_phase),
"total_ms": _ms_since(t_start),
},
}
core_match_roots = getattr(hits, "_core_match_roots", set())
root_to_shard = getattr(hits, "_root_to_shard", {})
qtokens_lower = {t.lower() for t in _title_query_tokens(question)}
def _body_density_check(h) -> bool:
# Lazy per-hit check: open the shard, count token mentions in this doc.
sp = root_to_shard.get(h.document_root) or h.shard_path
if not sp:
return False
c = connect(sp)
try:
return _body_density_passes(c, h.document_root, qtokens_lower)
finally:
c.close()
hits = _rerank(
hits,
question,
core_match_roots=core_match_roots,
body_density_check=_body_density_check,
)
search_ms = _ms_since(t_phase)
# 2. Pull doc texts within budget.
t_phase = time.monotonic()
chosen: list[_Hit] = []
context_parts: list[str] = []
char_budget = max_context_chars
for h in hits[:top_k]:
text = _load_doc_text(h.shard_path, h.document_root)
if not text:
continue
if len(text) > char_budget:
text = text[:char_budget]
context_parts.append(
f"=== Source: {h.document_uri} ===\n{text}"
)
chosen.append(h)
char_budget -= len(text)
if char_budget <= 0:
break
if not chosen:
return {
"status": "no_sources",
"msg": "top-k hits had cold or empty content",
"timings": {
"search_ms": search_ms,
"context_ms": _ms_since(t_phase),
"total_ms": _ms_since(t_start),
},
}
context = "\n\n".join(context_parts)
context_ms = _ms_since(t_phase)
# 3. Build messages + hashes.
messages = [
{"role": "system", "content": policy["system_prompt"]},
{
"role": "user",
"content": f"Sources:\n\n{context}\n\n---\n\nQuestion: {question}",
},
]
context_root = _context_root([h.document_root for h in chosen])
qhash = question_hash(question)
mhash = model_profile_hash(model_id, revision, quantization)
chash = conversation_hash(messages)
ghash = governance_policy_hash(policy)
ckey = cache_key(
context_root,
qhash,
mhash,
chash,
ghash,
SCHEMA_VERSION,
CANONICALIZATION_VERSION,
CHUNKING_VERSION,
)
qa_conn = connect(qa_db)
try:
# 4. Cache lookup.
t_phase = time.monotonic()
cached = qa_conn.execute(
"SELECT * FROM providence_cache "
"WHERE cache_key = ? AND falsification_state = 'live'",
(ckey,),
).fetchone()
cache_lookup_ms = _ms_since(t_phase)
if cached is not None:
now = int(time.time())
with transaction(qa_conn):
qa_conn.execute(
"UPDATE providence_cache "
"SET hit_count = hit_count + 1, last_hit_at = ? "
"WHERE cache_key = ?",
(now, ckey),
)
return {
"status": "cache_hit",
"audit_mode": "STRICT",
"cache_key": ckey,
"context_root": context_root,
"answer_text": cached["answer_text"],
"sources": json.loads(cached["merkle_proof"])["sources"],
"timings": {
"search_ms": search_ms,
"context_ms": context_ms,
"cache_lookup_ms": cache_lookup_ms,
"llm_ms": None,
"persist_ms": None,
"total_ms": _ms_since(t_start),
},
}
# 5. Cache miss — call LLM.
t_phase = time.monotonic()
answer_text = chat_client.chat_completion(
messages,
model=model_id,
temperature=policy["temperature"],
max_tokens=policy["max_tokens"],
top_p=policy.get("top_p", 1.0),
)
llm_ms = _ms_since(t_phase)
# 6. Persist record + audit event.
t_phase = time.monotonic()
proof_obj = {
"context_root": context_root,
"sources": [
{
"document_root": h.document_root,
"document_uri": h.document_uri,
"title": h.title,
"score": h.score,
"chunk_idx": h.chunk_idx,
"shard": Path(h.shard_path).name,
}
for h in chosen
],
}
proof_blob = json.dumps(proof_obj, separators=(",", ":"))
now = int(time.time())
with transaction(qa_conn):
event_hash = append_audit(
qa_conn,
event_type="providence_query",
subject_root=ckey,
body={
"context_root": context_root,
"n_sources": len(chosen),
"model_id": model_id,
"revision": revision,
"quantization": quantization,
"answer_chars": len(answer_text),
"context_chars": len(context),
},
ts=now,
)
qa_conn.execute(
"INSERT INTO providence_cache "
"(cache_key, source_root, document_uri, question_hash, question_text, "
" answer_text, merkle_proof, model_profile_hash, conversation_hash, "
" governance_policy_hash, schema_version, canonicalization_version, "
" chunking_version, falsification_state, chain, audit_event_hash, "
" created_at, hit_count) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0)",
(
ckey,
context_root,
"corpus://multi-source",
qhash,
question,
answer_text,
proof_blob,
mhash,
chash,
ghash,
SCHEMA_VERSION,
CANONICALIZATION_VERSION,
CHUNKING_VERSION,
chain,
event_hash,
now,
),
)
finally:
qa_conn.close()
persist_ms = _ms_since(t_phase)
return {
"status": "cache_miss_then_written",
"audit_mode": "STRICT",
"cache_key": ckey,
"context_root": context_root,
"answer_text": answer_text,
"sources": proof_obj["sources"],
"timings": {
"search_ms": search_ms,
"context_ms": context_ms,
"cache_lookup_ms": cache_lookup_ms,
"llm_ms": llm_ms,
"persist_ms": persist_ms,
"total_ms": _ms_since(t_start),
},
}
def _ms_since(t: float) -> float:
"""Wall-time elapsed in milliseconds, rounded to 1 decimal."""
return round((time.monotonic() - t) * 1000, 1)