engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cloud ask: full claim_lattice pipeline (matches local query shape)

Browse source 
Was using a naive single-prompt path → HYBRID via entity verifier.
Now wires the same claim-lattice pipeline `arborist query` uses:

  1. FTS5 search bucket-direct → top-K document hits
  2. Pull first chunk per hit via SQL through the same VFS
  3. build_evidence_map → E1/E2/E3 pointer tags
  4. render_evidence_map → `=== E1 (title | source_role) === span`
  5. messages[]:
       system: CLAIM_LATTICE_SYSTEM_PROMPT (worked examples + rules)
       user:   EVIDENCE blocks + QUESTION + GROUNDING_REMINDER
  6. LLM emits pointer-line answer: `Claim text. [E1,E2]`
  7. verify_claim_lattice() parses + textual-coverage-checks each
     (claim, pointer) pair (warrant_check disabled — no derivations
     table on bucket-direct)
  8. Annotate sources with used / used_pointer_ids
  9. render_claim_lattice() interpolates runtime-owned literal spans
     beside each claim — model never types the quote string

Result shape:
    who developed virt-back?
      EVIDENCE-WARRANTED · via claim_lattice  1/1  11.93s  (bucket-direct)

    - Russell Ballestrini developed virt-back.
      [E1 | virt-back: ... | 99330e72: "...spotlight excerpt..."]

    sources (4):
      [1] ... — primary_answer_source — used (E1) — ...
      [2] ... — background_source — unused — ...
      [3] ... — background_source — unused — ...
      [4] ... — background_source — unused — ...

    bucket: 98 HTTP requests · 388.1 KB · hermes / Hermes-3-8B
This commit is contained in:
russell@unturf.com 2026-05-30 11:28:05 -04:00
parent cb8c1deff2
commit 87f7d920e7
No known key found for this signature in database
1 changed files with 149 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

207
arborist/cli.py
View file

@ -7440,17 +7440,29 @@ def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int:
def _cmd_cloud_ask(args: argparse.Namespace) -> int:
"""Bucket-direct end-to-end ask: FTS → fetch chunks → LLM → verify.
Mirrors the local `query()` result shape (answer_text, audit_mode,
sources w/ source_role, n_quotes, n_verified, verifier_method) but
every byte read comes from the bucket via HttpRangeVFS. No local DB.
"""Bucket-direct end-to-end ask: FTS → fetch chunks → LLM (claim-
lattice mode) → verify. Same audited-answer shape as local
`query()` — EVIDENCE blocks with E1/E2 pointer tags, pointer-line
answer format, four-rung audit ladder (POINTER-LINKED →
ANCHOR-WARRANTED → EVIDENCE-WARRANTED) — but every byte goes
through HttpRangeVFS from the bucket.
"""
import os as _os
import re as _re
import time as _time
from arborist.compress import unpack_chunk
from arborist.qa.client import OpenAICompatibleClient
from arborist.qa.verify import verify_quotes
from arborist.qa.evidence import (
build_evidence_map,
evidence_map_by_pointer_id,
render_claim_lattice,
render_evidence_map,
)
from arborist.qa.prompts import (
CLAIM_LATTICE_GROUNDING_REMINDER,
CLAIM_LATTICE_SYSTEM_PROMPT,
)
from arborist.qa.verify import verify_claim_lattice
client = _make_bucket_client(args)
t0 = _time.time()
@ -7466,58 +7478,64 @@ def _cmd_cloud_ask(args: argparse.Namespace) -> int:
"sources": [],
"stats": client.stats(),
"timing_s": round(_time.time() - t0, 3),
"note": "FTS5 returned no hits — sanitized query may have "
"lost too many tokens; try `cloud search` first.",
"note": "FTS5 returned no hits — sanitized query may "
"have lost too many tokens; try `cloud search` first.",
}, indent=2, ensure_ascii=False))
return 0
# Pull body content for each hit via SQL through the same VFS;
# chunks.content lives in the .db file, so this reuses the
# already-warm page cache. Cap total context bytes.
sources: list[dict] = []
context_parts: list[str] = []
# Pull the first body chunk of each hit via SQL through the
# same VFS; reuses the already-warm page cache. Per-doc budget
# keeps one huge article from monopolizing the context.
chunks_for_evidence: list[dict] = []
total_chars = 0
per_doc_budget = max(
1000, args.max_context_chars // max(1, len(hits))
)
for rank, h in enumerate(hits, 1):
rows = list(client.conn.execute(
"SELECT idx, content FROM chunks "
"SELECT idx, leaf_hash, content FROM chunks "
"WHERE document_root = ? AND content IS NOT NULL "
"ORDER BY idx ASC",
"ORDER BY idx ASC LIMIT 1",
(h["document_root"],),
))
body = "\n\n".join(unpack_chunk(r[1]) or "" for r in rows).strip()
if not body:
if not rows:
continue
# Truncate per-doc so one huge article doesn't monopolize the budget.
per_doc_budget = max(
1000, args.max_context_chars // max(1, len(hits))
)
if len(body) > per_doc_budget:
body = body[:per_doc_budget] + "…"
sources.append({
"document_root": h["document_root"],
idx, leaf_hash, content_blob = rows[0]
span = unpack_chunk(content_blob) or ""
if not span:
continue
if len(span) > per_doc_budget:
span = span[:per_doc_budget]
chunks_for_evidence.append({
"source_root": h["document_root"],
"document_uri": h["document_uri"],
"title": h["title"],
"score": h["score"],
"chunk_idx": idx,
"chunk_root": leaf_hash,
"span": span,
"source_role": (
"primary_answer_source" if rank == 1 else "background_source"
"primary_answer_source" if rank == 1
else "background_source"
),
"n_chunks": len(rows),
})
context_parts.append(f"## [{rank}] {h['title']}\n{h['document_uri']}\n\n{body}")
total_chars += len(body)
total_chars += len(span)
if total_chars >= args.max_context_chars:
break
if not sources:
if not chunks_for_evidence:
print(json.dumps({
"answer_text": "",
"audit_mode": "UNGROUNDED",
"sources": hits,
"note": "FTS hit found, but document had no readable chunk content",
"note": "FTS hit found, but no readable chunk content",
}, indent=2, ensure_ascii=False))
return 0
context = "\n\n".join(context_parts)
# Build the evidence map + format the LLM prompt with E1/E2/…
# pointer tags — same shape the local query.py pipeline uses.
evidence_map = build_evidence_map(chunks_for_evidence)
evidence_text = render_evidence_map(evidence_map)
endpoint = args.endpoint or _os.environ.get(
"ARBORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1",
)
@ -7526,16 +7544,20 @@ def _cmd_cloud_ask(args: argparse.Namespace) -> int:
"adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
)
chat = OpenAICompatibleClient(base_url=endpoint)
prompt = (
"Answer the user's question using only the provided context. "
"When you state a fact, quote the supporting span verbatim in "
'double quotes ("…") so the verifier can confirm grounding. '
"If the context does not contain the answer, say so plainly."
f"\n\nContext:\n{context}\n\nQuestion: {args.question}"
)
messages = [
{"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT},
{
"role": "user",
"content": (
f"EVIDENCE:\n\n{evidence_text}\n\n"
f"QUESTION: {args.question}\n\n"
f"{CLAIM_LATTICE_GROUNDING_REMINDER}"
),
},
]
try:
answer = chat.chat_completion(
messages=[{"role": "user", "content": prompt}],
messages=messages,
model=model,
max_tokens=512,
temperature=0.1,
@ -7544,26 +7566,86 @@ def _cmd_cloud_ask(args: argparse.Namespace) -> int:
print(json.dumps({
"answer_text": "",
"audit_mode": "UNGROUNDED",
"sources": sources,
"sources": [
{"document_root": h["document_root"],
"document_uri": h["document_uri"],
"title": h["title"], "score": h["score"]}
for h in hits
],
"error": f"LLM call failed: {type(e).__name__}: {e}",
"endpoint": endpoint,
"model": model,
}, indent=2, ensure_ascii=False))
return 4
verdict = verify_quotes(answer, context)
# Claim-lattice verifier: parses the model's pointer-line
# output, resolves each [E#] back through the evidence map,
# runs the textual-coverage check per claim. Warrant chain
# disabled in cloud mode (no derivations table available).
verdict = verify_claim_lattice(
answer, evidence_map,
question=args.question,
warrant_check_enabled=False,
)
finally:
client.close()
elapsed = _time.time() - t0
# Annotate sources with used / used_pointer_ids per the model's
# cited pointers (parsed from the answer text). Same render-time
# signal `arborist query` surfaces for lattice modes.
by_pid = evidence_map_by_pointer_id(evidence_map)
pointer_re = _re.compile(r"\[E\d+(?:,\s*E\d+)*\]")
used_doc_roots: set[str] = set()
doc_root_to_pointers: dict[str, list[str]] = {}
for m in pointer_re.finditer(answer):
for pid in (p.strip() for p in m.group(0).strip("[]").split(",")):
ev = by_pid.get(pid)
if ev:
used_doc_roots.add(ev.source_root)
doc_root_to_pointers.setdefault(ev.source_root, []).append(pid)
sources_for_render: list[dict] = []
seen_roots: set[str] = set()
for ev in evidence_map:
if ev.source_root in seen_roots:
continue
seen_roots.add(ev.source_root)
sources_for_render.append({
"document_root": ev.source_root,
"document_uri": ev.document_uri,
"title": ev.title,
"source_role": ev.source_role,
"used": ev.source_root in used_doc_roots,
"used_pointer_ids": sorted(set(
doc_root_to_pointers.get(ev.source_root, [])
)),
})
# Render the answer in lattice form (claim line + spotlight excerpts).
claim_statuses = verdict.get("claim_statuses") or []
rendered_answer = render_claim_lattice(
[
{
"text": cs.get("text", ""),
"pointer_ids": cs.get("pointer_ids") or [],
}
for cs in claim_statuses
],
by_pid,
window=200,
) or answer # fall back to raw output if parser dropped everything
result = {
"answer_text": answer,
"answer_text": rendered_answer,
"raw_answer": answer,
"audit_mode": verdict["audit_mode"],
"verifier_method": verdict["verifier_method"],
"n_quotes": verdict["n_quotes"],
"n_verified": verdict["n_verified"],
"unverified_quotes": verdict.get("unverified_quotes", []),
"sources": sources,
"violations": verdict.get("violations") or [],
"sources": sources_for_render,
"stats": client.stats(),
"timing_s": round(elapsed, 3),
"endpoint": endpoint,
@ -7578,14 +7660,16 @@ def _cmd_cloud_ask(args: argparse.Namespace) -> int:
def _render_cloud_ask_human(result: dict, question: str) -> str:
"""Pretty-print a cloud-ask result. Mirrors `_render_query_human`'s
layout but trims sections that the bucket-direct path doesn't
layout — same audit-label ladder, same source roles + used / pointer
annotations — but trims sections the bucket-direct path doesn't
produce (warrant tails, run-DAG, retrieval-purity)."""
audit = result.get("audit_mode") or "UNGROUNDED"
method = result.get("verifier_method") or "?"
n_quotes = result.get("n_quotes") or 0
n_verified = result.get("n_verified") or 0
elapsed_s = result.get("timing_s") or 0.0
label = _render_audit_label(audit, method, [])
violations = result.get("violations") or []
label = _render_audit_label(audit, method, violations)
stats = result.get("stats") or {}
cache_stats = stats.get("cache") or {}
http_reqs = stats.get("http_requests")
@ -7607,17 +7691,24 @@ def _render_cloud_ask_human(result: dict, question: str) -> str:
uri = s.get("document_uri", "")
title = (s.get("title") or "").strip() or _short_path(uri)
role = s.get("source_role")
role_part = f" — {role}" if role else ""
lines.append(
f" [{i}] {title}{role_part} — {_strip_scheme(uri)}"
used = s.get("used")
pointer_ids = s.get("used_pointer_ids") or []
annotations: list[str] = []
if role:
annotations.append(role)
if used is True:
if pointer_ids:
annotations.append(f"used ({','.join(pointer_ids)})")
else:
annotations.append("used")
elif used is False:
annotations.append("unused")
annotation_part = (
" — " + " — ".join(annotations) if annotations else ""
)
lines.append(
f" [{i}] {title}{annotation_part} — {_strip_scheme(uri)}"
)
lines.append("")
unverified = result.get("unverified_quotes") or []
if unverified:
lines.append(f"unverified ({len(unverified)}):")
for q in unverified[:5]:
lines.append(f' - "{q[:140]}{"…" if len(q) > 140 else ""}"')
lines.append("")
bytes_kb = bytes_fetched / 1024.0