diff --git a/arborist/cli.py b/arborist/cli.py index ff042cd..b875e6e 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -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