Spaces:
Sleeping
Sleeping
File size: 5,933 Bytes
f0307a2 3f1f29b f0307a2 3f1f29b f0307a2 1121d82 | 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 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | from pathlib import Path
from ragforge.context_budget import adaptive_context_budget, adaptive_retrieval_top_k
from ragforge.evidence_compression import compress_text_for_query, focused_evidence_compression
from ragforge.schemas import Chunk, PipelineConfig, QueryPlan, SearchHit
ROOT = Path(__file__).resolve().parents[1]
def _plan(task="fact_lookup", strategy="semantic") -> QueryPlan:
return QueryPlan(
route="documents",
knowledge_scope="corpus",
task_type=task,
retrieval_strategy=strategy,
web_relevance="irrelevant",
rewritten_query="Sev-1 acknowledgement target",
document_queries=["Sev-1 acknowledgement target"],
)
def _hit(idx: int, source: str, dense: float, sparse: float, text: str | None = None) -> SearchHit:
return SearchHit(
chunk=Chunk(
id=f"c{idx}",
source=source,
text=text or ("generic evidence about service operations. " * 35),
),
score=max(dense, sparse),
dense_score=dense,
sparse_score=sparse,
)
def test_v19_adaptive_budget_can_use_two_chunks_for_clear_small_corpus_match():
hits = [
_hit(0, "acme.md", 0.95, 0.95),
_hit(1, "acme.md", 0.55, 0.30),
_hit(2, "nist.pdf", 0.40, 0.10),
_hit(3, "other.md", 0.30, 0.05),
]
decision = adaptive_context_budget(
hits,
_plan(),
PipelineConfig(top_k=6, use_context_pruning=True),
corpus_chunks=90,
corpus_sources=5,
)
assert decision.used is True
assert decision.target_chunks == 2
assert decision.hits == hits[:2]
assert decision.corpus_scale == "small"
assert decision.retrieval_confidence > 0.75
def test_v19_adaptive_budget_expands_safety_floor_for_large_ambiguous_corpus():
hits = [
_hit(0, "a.md", 0.62, 0.45),
_hit(1, "b.md", 0.61, 0.43),
_hit(2, "c.md", 0.59, 0.40),
_hit(3, "d.md", 0.55, 0.35),
_hit(4, "e.md", 0.50, 0.30),
_hit(5, "f.md", 0.45, 0.25),
]
decision = adaptive_context_budget(
hits,
_plan(),
PipelineConfig(top_k=6, use_context_pruning=True),
corpus_chunks=2200,
corpus_sources=45,
)
assert decision.target_chunks == 5
assert decision.chunks_after == 5
assert decision.corpus_scale == "large"
def test_v19_adaptive_retrieval_depth_scales_beyond_small_corpus_baseline():
cfg = PipelineConfig(top_k=6, use_adaptive_top_k=True)
assert adaptive_retrieval_top_k(cfg, _plan(), corpus_chunks=90, corpus_sources=5) == 6
assert adaptive_retrieval_top_k(cfg, _plan(), corpus_chunks=500, corpus_sources=15) == 8
assert adaptive_retrieval_top_k(cfg, _plan(), corpus_chunks=1600, corpus_sources=40) == 10
assert adaptive_retrieval_top_k(cfg, _plan(), corpus_chunks=3500, corpus_sources=90) == 12
def test_v19_sentence_compression_keeps_query_relevant_fact():
text = (
"The Checkout API has a monthly availability SLO of 99.95%. "
"A Sev-1 incident is a complete outage or payment failure. "
"The on-call engineer must acknowledge a Sev-1 page within 5 minutes. "
+ "Unrelated deployment documentation is included here. " * 40
)
compressed = compress_text_for_query("What is the Sev-1 acknowledgement target?", text, max_chars=500)
assert "5 minutes" in compressed
assert len(compressed) < len(text)
def test_v19_focused_compression_does_not_touch_broad_synthesis():
hits = [_hit(0, "a.md", 0.9, 0.8)]
result = focused_evidence_compression(
hits,
_plan(task="insight_synthesis", strategy="analytical"),
query="What trends stand out?",
enabled=True,
)
assert result.used is False
assert result.reason == "broad_or_multi_source_task"
def test_v19_evaluation_contains_scale_stress_compression_and_readiness():
text = (ROOT / "src/ragforge/evaluation.py").read_text(encoding="utf-8")
assert "scale_stress_retrieval_eval" in text
assert '"evidence_compression_ablation": compression_rows' in text
assert '"release_readiness": readiness_rows' in text
assert '"scale_stress": scale_stress_rows' in text
assert "Adaptive budget" in text
def test_v19_api_and_ui_surface_operational_diagnostics_and_new_eval_tabs():
api = (ROOT / "src/ragforge/api.py").read_text(encoding="utf-8")
ui = (ROOT / "src/ragforge/ui.py").read_text(encoding="utf-8")
assert "/api/v1/session/{session_id}/diagnostics" in api
assert 'with gr.Tab("Evidence compression")' in ui
assert 'with gr.Tab("Scale stress")' in ui
assert 'with gr.Tab("Acceptance checks")' in ui
assert "Adaptive retrieval depth" in ui
assert "Focused evidence sentence compression" in ui
def test_v19_planner_keeps_overview_and_insight_strategies_structurally_consistent():
text = (ROOT / "src/ragforge/llm.py").read_text(encoding="utf-8")
assert 'if plan.task_type == "overview"' in text
assert 'plan.retrieval_strategy = "global"' in text
assert 'elif plan.task_type == "insight_synthesis"' in text
assert 'plan.retrieval_strategy = "analytical"' in text
def test_v203_sentence_compression_keeps_sev1_value_and_excludes_inherited_sev2_target():
text = (
"A Sev-1 incident is a complete outage, confirmed data-loss event, or payment-processing failure. "
"The on-call engineer must acknowledge a Sev-1 page within 5 minutes and establish an incident channel within 10 minutes. "
"A Sev-2 incident is a major degradation affecting at least 5% of requests. "
"The acknowledgement target is 15 minutes. "
+ "Unrelated recovery documentation. " * 40
)
compressed = compress_text_for_query("What is the Sev-1 acknowledgement target?", text, max_chars=500)
assert "within 5 minutes" in compressed
assert "acknowledgement target is 15 minutes" not in compressed.lower()
|