Capture retrieved passages

Use this optional reference when a retrieval comparison needs the exact chunks, source IDs and tenant ownership evidence. Start with the verification checklist to set the target and evidence directory; this calls live services and writes Langfuse evaluation runs. Keep raw results gitignored.

The standard evaluator scores relevance but does not export all passages. When comparing missing or duplicated retrieval evidence, save this temporary wrapper as $KB_VERIFY_EVIDENCE/capture_retrieval.py:

import dataclasses
import json
import os
from pathlib import Path

from cli.cli_eval import main
from ixevaluation.rewriter.evaluator import LangfuseDataset, QueryRewriterEvaluator
from ixrag.lightrag.langgraph_retriever import LightRAGRetriever

folder = Path(os.environ["KB_VERIFY_EVIDENCE"])
output = folder / os.environ["KB_VERIFY_RESULT"]
domain = os.environ["KB_VERIFY_DOMAIN"]
original_fetch = LangfuseDataset.fetch
original_run = QueryRewriterEvaluator.run


def fetch_tenant(self):
    items = [item for item in original_fetch(self)
             if (item.metadata or {}).get("site_name") == domain]
    assert items, f"No dataset items for {domain}"
    snapshot = [{"id": item.id, "input": item.input,
                 "expected_output": item.expected_output, "metadata": item.metadata}
                for item in items]
    snapshot.sort(key=lambda item: item["id"] or "")
    path = folder / "retrieval-dataset.json"
    if path.exists():
        assert json.loads(path.read_text()) == snapshot, "Comparison dataset changed"
    else:
        path.write_text(json.dumps(snapshot, indent=2, ensure_ascii=False) + "\n")
    return items


async def record_run(self):
    original_invoke = LightRAGRetriever.ainvoke
    passages = []

    async def capture(retriever, query, *args, **kwargs):
        documents = await original_invoke(retriever, query, *args, **kwargs)
        passages.append({"retrieval_input": query, "retrieval_kwargs": kwargs,
                         "documents": [{"page_content": doc.page_content,
                                        "metadata": doc.metadata} for doc in documents]})
        return documents

    LightRAGRetriever.ainvoke = capture
    try:
        result = await original_run(self)
        output.write_text(json.dumps(dataclasses.asdict(result), indent=2,
                                     ensure_ascii=False) + "\n")
        return result
    finally:
        LightRAGRetriever.ainvoke = original_invoke
        output.with_name(output.stem + "-passages.json").write_text(
            json.dumps(passages, indent=2, ensure_ascii=False, default=str) + "\n")


LangfuseDataset.fetch = fetch_tenant
QueryRewriterEvaluator.run = record_run
raise SystemExit(main())

From backend/, with the exported variables from the verification checklist, source the CLI secrets helper, then run:

source scripts/load-dev-secrets.sh
export KB_VERIFY_PHASE='test-before'
export KB_VERIFY_ENV='test'
git rev-parse HEAD > "$KB_VERIFY_EVIDENCE/$KB_VERIFY_PHASE-commit.txt"
for kb_trial in 1 2; do
  KB_VERIFY_RESULT="$KB_VERIFY_PHASE-$kb_trial.json" \
    .venv/bin/python "$KB_VERIFY_EVIDENCE/capture_retrieval.py" \
    query-rewriter query-rewriter-v1 --arm production --rag-mode mix \
    --env "$KB_VERIFY_ENV" --concurrency 1 --json \
    > "$KB_VERIFY_EVIDENCE/$KB_VERIFY_PHASE-$kb_trial.log" 2>&1
  kb_result=$?
  printf '%s\n' "$kb_result" > "$KB_VERIFY_EVIDENCE/$KB_VERIFY_PHASE-$kb_trial.exit"
done

For post-ingestion or local-fix trials, change the phase to a distinct name such as test-after or test-identity-fixed. For an approved production comparison, set both phase and environment to production. Never overwrite baseline files; the wrapper asserts dataset equality across phases. Use naive instead of mix when testing a lite tenant, consistently in all comparison arms. Inspect every trial even if the evaluator exits nonzero: separate a relevance-floor failure from a runtime failure, and report both rather than silently treating either as success. If interrupted before evaluation completes, partial passages do not constitute a completed trial.

For each question, compare scores, latency and the actual ordered passage text, logical chunk IDs, source paths and parent documents. Count unique content hashes as well as unique IDs: chunk-X and tenant:chunk-X can be two identities for the same evidence. Save representative passages and judge reasons in the private report. When identities disagree, trace both vector and graph paths at the merge boundary; verify the adapter contract with the focused unit regression above.

Summarize exact-content uniqueness from the saved passages (retain source IDs and texts for manual comparison; repeated text across different queries is normal):

.venv/bin/python - <<'PY'
import hashlib
import json
import os
from pathlib import Path

for path in sorted(Path(os.environ["KB_VERIFY_EVIDENCE"]).glob("*-passages.json")):
    for call in json.loads(path.read_text()):
        docs = call["documents"]
        hashes = {hashlib.sha256(doc["page_content"].encode()).hexdigest()
                  for doc in docs}
        print(path.name, repr(call["retrieval_input"]),
              f"{len(hashes)} unique bodies / {len(docs)} returned")
PY

Check suspected tenant leakage against stored rows using the existing diagnostic collector and scoped read queries: Mongo composite _id, explicit tenantId, content hash, source path and owner document must agree, and graph nodes/edges must have the expected tenant filter. A logical result ID without a tenant prefix is intentional; it does not alone prove a missing filter or cross-tenant read. Scope the conclusion to the records checked. Do not strip storage prefixes or change database rows to make a comparison look clean.