cloud query: stage-level progress + capacity/timings tail
Mirrors local 'arborist query' instrumentation so the cloud path is
no less observable. Uses the existing arborist.qa.progress.Progress
emitter (auto-on at TTY, override via ARBORIST_PROGRESS=0|1).
Stages emitted:
manifest.start/done URL + shard/sidecar counts
corpus.open.start/done sidecar download + dict parse cost
search.start/done per-shard FTS + RRF merge
context.start/done chunk-content pulls + assembled bytes
llm.start/done model + endpoint + ctx + answer chars
verify.start/done audit_mode + n_verified/n_quotes
Render tail adds (matches local query render):
capacity: prompt N chars (sys N + evidence N + question N) → answer N chars
timings: manifest Xs · corpus_open Xs · search Xs · context Xs ·
llm Xs · verify Xs · **total Xs**
bucket: N HTTP requests · KB · endpoint / model
JSON output (--json / JSON=1) gains 'timings' + 'capacity' fields
with the same shape, so bench harnesses can consume them directly.
Surfaces that sidecar.parse dominates a cold cloud-query (~22s/shard
on the genesis 6M-term sidecar). That's a real future-opt target
(ProcessPool instead of ThreadPool to escape the GIL) but the
mechanism is observable now, which is the prerequisite for tuning.
This commit is contained in:
parent
1b792147c2
commit
c501411ef5
1 changed files with 97 additions and 16 deletions
113
arborist/cli.py
113
arborist/cli.py
|
|
@ -7482,12 +7482,15 @@ def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int:
|
|||
|
||||
|
||||
def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
||||
"""Bucket-direct end-to-end ask: FTS → fetch chunks → LLM (claim-
|
||||
"""Bucket-direct end-to-end query: 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.
|
||||
|
||||
Emits stage-level progress to stderr (mirrors `arborist query`),
|
||||
auto-on at TTY, override via ARBORIST_PROGRESS=0/1.
|
||||
"""
|
||||
import os as _os
|
||||
import re as _re
|
||||
|
|
@ -7500,12 +7503,16 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
render_claim_lattice,
|
||||
render_evidence_map,
|
||||
)
|
||||
from arborist.qa.progress import from_env as _progress_from_env
|
||||
from arborist.qa.prompts import (
|
||||
CLAIM_LATTICE_GROUNDING_REMINDER,
|
||||
CLAIM_LATTICE_SYSTEM_PROMPT,
|
||||
)
|
||||
from arborist.qa.verify import verify_claim_lattice
|
||||
|
||||
progress = _progress_from_env()
|
||||
timings: dict[str, float] = {}
|
||||
|
||||
# Multi-shard via manifest, or single-shard if --shard-url given.
|
||||
if getattr(args, "bucket_url", None):
|
||||
from arborist.wallet.bucket import (
|
||||
|
|
@ -7513,16 +7520,22 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
MultiShardSidecarCorpus,
|
||||
load_bucket_manifest,
|
||||
)
|
||||
manifest_url = args.bucket_url
|
||||
# Convenience: allow passing the manifest path directly.
|
||||
if not manifest_url.rstrip("/").endswith(".json"):
|
||||
manifest_url = manifest_url # base URL → load_bucket_manifest appends path
|
||||
manifest = load_bucket_manifest(manifest_url)
|
||||
progress.emit("manifest.start", url=args.bucket_url)
|
||||
ts = _time.time()
|
||||
manifest = load_bucket_manifest(args.bucket_url)
|
||||
timings["manifest"] = _time.time() - ts
|
||||
any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards)
|
||||
progress.emit(
|
||||
"manifest.done", shards=len(manifest.shards),
|
||||
sidecars=sum(1 for sh in manifest.shards if sh.get("sidecar_url")),
|
||||
ms=int(timings["manifest"] * 1000),
|
||||
)
|
||||
# Pick the right corpus class based on whether the manifest
|
||||
# advertises per-shard sidecars. Mixed (some sidecar, some not)
|
||||
# also uses the sidecar-aware path — it falls back to bucket-
|
||||
# direct FTS5 for shards without one.
|
||||
any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards)
|
||||
progress.emit("corpus.open.start", mode="sidecar" if any_sidecar else "bucket-direct")
|
||||
ts = _time.time()
|
||||
if any_sidecar:
|
||||
client = MultiShardSidecarCorpus(
|
||||
manifest, cache_bytes_per_shard=args.cache_mb * 1024 * 1024,
|
||||
|
|
@ -7531,6 +7544,8 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
client = MultiShardBucketCorpus(
|
||||
manifest, cache_bytes_per_shard=args.cache_mb * 1024 * 1024,
|
||||
)
|
||||
timings["corpus_open"] = _time.time() - ts
|
||||
progress.emit("corpus.open.done", ms=int(timings["corpus_open"] * 1000))
|
||||
multi_shard = True
|
||||
else:
|
||||
if not getattr(args, "shard_url", None):
|
||||
|
|
@ -7544,7 +7559,11 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
multi_shard = False
|
||||
t0 = _time.time()
|
||||
try:
|
||||
progress.emit("search.start", top_k=args.top_k)
|
||||
ts = _time.time()
|
||||
hits = client.fts_search(args.question, limit=args.top_k)
|
||||
timings["search"] = _time.time() - ts
|
||||
progress.emit("search.done", hits=len(hits), ms=int(timings["search"] * 1000))
|
||||
if not hits:
|
||||
print(json.dumps({
|
||||
"answer_text": "",
|
||||
|
|
@ -7563,6 +7582,11 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
# 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.
|
||||
progress.emit(
|
||||
"context.start", chunks=len(hits),
|
||||
max_context_chars=args.max_context_chars,
|
||||
)
|
||||
ts = _time.time()
|
||||
chunks_for_evidence: list[dict] = []
|
||||
total_chars = 0
|
||||
per_doc_budget = max(
|
||||
|
|
@ -7606,6 +7630,12 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
if total_chars >= args.max_context_chars:
|
||||
break
|
||||
|
||||
timings["context"] = _time.time() - ts
|
||||
progress.emit(
|
||||
"context.done", sources=len(chunks_for_evidence),
|
||||
assembled_chars=total_chars, ms=int(timings["context"] * 1000),
|
||||
)
|
||||
|
||||
if not chunks_for_evidence:
|
||||
print(json.dumps({
|
||||
"answer_text": "",
|
||||
|
|
@ -7628,24 +7658,33 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
"adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
|
||||
)
|
||||
chat = OpenAICompatibleClient(base_url=endpoint)
|
||||
sys_prompt = CLAIM_LATTICE_SYSTEM_PROMPT
|
||||
user_prompt = (
|
||||
f"EVIDENCE:\n\n{evidence_text}\n\n"
|
||||
f"QUESTION: {args.question}\n\n"
|
||||
f"{CLAIM_LATTICE_GROUNDING_REMINDER}"
|
||||
)
|
||||
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}"
|
||||
),
|
||||
},
|
||||
{"role": "system", "content": sys_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
]
|
||||
prompt_chars = len(sys_prompt) + len(user_prompt)
|
||||
progress.emit(
|
||||
"llm.start", model=model, endpoint=endpoint, ctx_chars=prompt_chars,
|
||||
)
|
||||
try:
|
||||
ts = _time.time()
|
||||
answer = chat.chat_completion(
|
||||
messages=messages,
|
||||
model=model,
|
||||
max_tokens=512,
|
||||
temperature=0.1,
|
||||
)
|
||||
timings["llm"] = _time.time() - ts
|
||||
progress.emit(
|
||||
"llm.done", answer_chars=len(answer),
|
||||
ms=int(timings["llm"] * 1000),
|
||||
)
|
||||
except Exception as e:
|
||||
print(json.dumps({
|
||||
"answer_text": "",
|
||||
|
|
@ -7666,11 +7705,19 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
# 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).
|
||||
progress.emit("verify.start", mode="claim_lattice")
|
||||
ts = _time.time()
|
||||
verdict = verify_claim_lattice(
|
||||
answer, evidence_map,
|
||||
question=args.question,
|
||||
warrant_check_enabled=False,
|
||||
)
|
||||
timings["verify"] = _time.time() - ts
|
||||
progress.emit(
|
||||
"verify.done", audit_mode=verdict["audit_mode"],
|
||||
n_verified=verdict["n_verified"], n_quotes=verdict["n_quotes"],
|
||||
ms=int(timings["verify"] * 1000),
|
||||
)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
|
@ -7721,6 +7768,14 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
window=200,
|
||||
) or answer # fall back to raw output if parser dropped everything
|
||||
|
||||
timings["total"] = elapsed
|
||||
capacity = {
|
||||
"sys_prompt_chars": len(sys_prompt),
|
||||
"evidence_chars": len(evidence_text),
|
||||
"question_chars": len(args.question),
|
||||
"prompt_chars": prompt_chars,
|
||||
"answer_chars": len(answer),
|
||||
}
|
||||
result = {
|
||||
"answer_text": rendered_answer,
|
||||
"raw_answer": answer,
|
||||
|
|
@ -7732,6 +7787,8 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
"sources": sources_for_render,
|
||||
"stats": client.stats(),
|
||||
"timing_s": round(elapsed, 3),
|
||||
"timings": {k: round(v, 3) for k, v in timings.items()},
|
||||
"capacity": capacity,
|
||||
"endpoint": endpoint,
|
||||
"model": model,
|
||||
}
|
||||
|
|
@ -7804,6 +7861,30 @@ def _render_cloud_query_human(result: dict, question: str) -> str:
|
|||
)
|
||||
lines.append("")
|
||||
|
||||
# Capacity tail — same shape as local query, scoped to the bucket-
|
||||
# direct numbers we actually have (sys + evidence + question, no
|
||||
# reminder block because the claim-lattice grounding reminder is
|
||||
# folded into the user-prompt evidence section).
|
||||
cap = result.get("capacity") or {}
|
||||
if cap:
|
||||
lines.append(
|
||||
f"capacity: prompt {cap.get('prompt_chars', 0):,} chars "
|
||||
f"(sys {cap.get('sys_prompt_chars', 0):,} "
|
||||
f"+ evidence {cap.get('evidence_chars', 0):,} "
|
||||
f"+ question {cap.get('question_chars', 0):,}) → "
|
||||
f"answer {cap.get('answer_chars', 0):,} chars"
|
||||
)
|
||||
|
||||
# Per-stage timings — mirrors the local query 'timings: cache X · ...' line.
|
||||
t = result.get("timings") or {}
|
||||
if t:
|
||||
parts = []
|
||||
for k in ("manifest", "corpus_open", "search", "context", "llm", "verify"):
|
||||
if k in t:
|
||||
parts.append(f"{k} {t[k]:.2f}s")
|
||||
parts.append(f"**total {t.get('total', elapsed_s):.2f}s**")
|
||||
lines.append("timings: " + " · ".join(parts))
|
||||
|
||||
bytes_kb = bytes_fetched / 1024.0
|
||||
lines.append(
|
||||
f"bucket: {http_reqs} HTTP requests · {bytes_kb:.1f} KB · "
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue