Spaces:
Sleeping
Sleeping
File size: 5,045 Bytes
a671976 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """Human-readable dump of the pre-rerank retrieval shortlist, for auditing recall.
Enabled by ``settings.retrieval_debug_dump`` (env ``RETRIEVAL_DEBUG_DUMP``). For each
reference-mapping retrieval it appends one block to
``<data_dir>/retrieval_debug/retrieval_<date>.log`` capturing the hybrid candidates
(dense jina-embeddings-v3 + BM25, RRF-fused) exactly as they stand BEFORE
jina-reranker-v3 reorders them — so you can see whether the embedder is pulling the
right chunks independent of the reranker.
Best-effort only: any failure here is swallowed so it can never break generation.
"""
from __future__ import annotations
import logging
import threading
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING
from backend.config import settings
from backend.observability import token_audit
if TYPE_CHECKING:
from backend.rag.types import SearchHit
logger = logging.getLogger(__name__)
# Serialise writes: report generation fans many section retrievals across threads.
_write_lock = threading.Lock()
def enabled() -> bool:
return bool(getattr(settings, "retrieval_debug_dump", False))
def _dump_file() -> Path:
directory = settings.data_dir_path / "retrieval_debug"
directory.mkdir(parents=True, exist_ok=True)
return directory / f"retrieval_{datetime.now():%Y%m%d}.log"
def _clip(text: str, cap: int) -> str:
text = (text or "").strip()
if cap and len(text) > cap:
return f"{text[:cap]} …[+{len(text) - cap} more chars]"
return text
def dump_pre_rerank(
*,
tenant_id: str,
query: str,
section_id: str,
section_label: str,
interference_level: str,
retrieval_level: str,
hits: list[SearchHit],
rerank_top_n: int,
final_top_k: int,
) -> None:
"""Append the pre-rerank candidate shortlist to the audit file (no-op if off)."""
if not enabled():
return
try:
cap = int(getattr(settings, "retrieval_debug_max_text_chars", 1500) or 0)
embed_cap = int(getattr(settings, "local_embedding_max_seq_length", 0) or 0)
rerank_char_cap = int(
getattr(settings, "reference_cross_encoder_doc_chars", 0) or 0
)
query_summary = token_audit.summarize_text(query)
query_embed = token_audit.summarize_embedder_feed(
query, max_seq_length=embed_cap
)
out: list[str] = []
out.append("=" * 100)
out.append(
f"[{datetime.now():%Y-%m-%d %H:%M:%S}] tenant={tenant_id or '?'} "
f"section={section_id or '?'} ({section_label or '-'}) "
f"interference={interference_level} retrieval_level={retrieval_level}"
)
out.append(f"QUERY: {(query or '').strip()}")
out.append(
f"query chars={query_summary['chars']} tok={query_summary['tokens']} "
f"(embed_cap={embed_cap or 'model-default'}, "
f"embed_trunc={'yes' if query_embed['would_truncate'] else 'no'})"
)
out.append(
f"candidates={len(hits)} reranker_shortlist(top_n)={rerank_top_n} "
f"final_top_k={final_top_k} reranker_doc_chars={rerank_char_cap}"
)
out.append(
"token counts = tiktoken cl100k_base audit proxy "
"(jina models use XLM-R tokenizer; counts are relative, not exact)"
)
out.append("order below = EMBEDDER/HYBRID rank (dense+BM25 RRF), PRE-rerank")
out.append("-" * 100)
if not hits:
out.append("(no candidates retrieved)")
for i, h in enumerate(hits, start=1):
fed = "-> reranker" if i <= rerank_top_n else "tail (not reranked)"
embed = token_audit.summarize_embedder_feed(
h.text or "", max_seq_length=embed_cap
)
rerank = token_audit.summarize_reranker_feed(
h.text or "",
doc_chars_cap=rerank_char_cap if i <= rerank_top_n else 0,
)
rerank_tok_line = (
f"rerank_fed={rerank['fed_chars']}c/{rerank['fed_tokens']}tok"
if i <= rerank_top_n
else "rerank_fed=n/a"
)
out.append(
f"#{i:<2} [{fed}] score={h.score:.4f} fusion={h.fusion_score:.4f} "
f"sec={h.section_id or '-'} para={h.paragraph_index} "
f"src={h.source_filename or '-'} chunk={h.chunk_id or '-'} "
f"scrubbed={h.is_scrubbed} "
f"chars={embed['chars']} tok={embed['tokens']} "
f"embed_trunc={'yes' if embed['would_truncate'] else 'no'} "
f"{rerank_tok_line}"
)
out.append(f" {_clip(h.text, cap)}")
out.append("=" * 100)
out.append("")
blob = "\n".join(out)
with _write_lock, _dump_file().open("a", encoding="utf-8") as fh:
fh.write(blob)
except Exception: # noqa: BLE001 - auditing must never break retrieval
logger.debug("retrieval_debug dump failed", exc_info=True)
|