diff --git a/docs/RAG.md b/docs/RAG.md index 6cd02f3..6c7d3de 100644 --- a/docs/RAG.md +++ b/docs/RAG.md @@ -77,7 +77,44 @@ Chroma 同步会删除 replacement index 中不存在的 ID，避免删除、重 当前 K1 corpus 上的门控指标仍属于 `record_only`。即使小样本上 answerable supported rate 和 unanswerable block rate 达到 1.0，也不能直接升级为长期不可回退阈值，必须继续用更大真实语料和真实模型 replay 验证泛化。 -## 5. Stable capability boundary +## 5. Adaptive multi-source coverage + +复合学习问题不能简单地“每个来源只取一个 chunk”，也不能为了多样性随机塞入更多来源。K1d 使用**非回退的自适应来源覆盖**： + +1. 先执行普通 top-K 检索； +2. 保留普通 top-K 中已经出现的所有唯一来源； +3. 只删除同一来源占据的重复槽位； +4. 对明确的复合问题提取有限 facet； +5. 只有 facet 的 champion 来源尚未出现时，才使用重复槽位补充该来源。 + +因此 K1d 的硬约束是：**adaptive path 不得丢失 raw top-K 已召回的唯一来源。** 普通单来源问题不启用该策略，保持原排序行为。 + +当前确定性 K1 合同要求： + +- multi-source recall@K `>= 0.9`； +- multi-source precision@K `>= 0.7`； +- adaptive overall recall 不得低于 raw Hybrid recall； +- answerable supported rate 与 unanswerable block rate 不得回退； +- forbidden / stale source leakage 保持为 0。 + +这些阈值约束当前 fixture 上的回归，不代表更大真实语料已经证明同样性能。 + +## 6. Real-provider answer replay contract + +确定性 extractive lower bound 只能验证检索和评测合同，不能代表真实模型回答质量。K1e 因此单独定义 real-provider replay： + +- 复用 Study Agent 现有 Provider owner 和模型档位，不创建平行的 Provider 配置体系； +- 使用同一 K1 corpus、evidence manifest、local-knowledge 检索、K1b 资格过滤、K1c sufficiency 和 K1d adaptive coverage； +- 默认只回放 answer-quality gold cases，检索层继续由完整 K1 retrieval suite 负责； +- 报告记录 corpus fingerprint、prompt-template fingerprint、case prompt fingerprint、Provider profile、model name、endpoint fingerprint、latency 和 Provider 返回的 token usage； +- 原始 API key 和 raw endpoint 不得写入报告； +- raw replay artifact 默认写入 gitignored `output/`，不能无审查进入版本库。 + +provenance 不能由任意测试对象自行声明。只有生产 `OpenAICompatibleReplayProvider` adapter 才能被 runner 标记为 `real_provider`；其他实现即使伪造同名字符串，也只能记录为 `synthetic_test`。 + +真实 Provider 不可用时必须显式记录 `provider_unavailable` 或 `partial_failure`，不得生成或补齐伪造质量分数。当前 harness 是否已具备、某次真实 replay 是否已实际执行，属于进度信息，以 `PROJECT_STATUS.md` 为准。 + +## 7. Stable capability boundary 当前稳定能力包括： @@ -93,6 +130,8 @@ Chroma 同步会删除 replacement index 中不存在的 ID，避免删除、重 - 默认持久化 JSON index； - citation-first context formatting； - evidence sufficiency / refusal boundary； +- non-regressive adaptive multi-source coverage； +- real-provider replay harness 与 provenance schema； - 可选 Chroma adapter； - OpenAI-compatible embedding provider； - duplicate suppression、source diversity、metadata filters； @@ -116,7 +155,7 @@ Chroma 同步会删除 replacement index 中不存在的 ID，避免删除、重 这些项目的当前状态以 `PROJECT_STATUS.md` 为准。 -## 6. Quality evaluation contract +## 8. Quality evaluation contract RAG 质量评测复用同一套 `src/rag/eval.py` 检索评测域，不另建平行指标系统。K1 的确定性本地 corpus 同时包含干净查询、改写、多来源、相似主题干扰、旧版本和不可回答问题。 @@ -150,17 +189,27 @@ source-level nDCG 只在某个来源第一次出现时给予 relevance gain。 - required source coverage； - stale revision / forbidden-source leakage。 -确定性 CI baseline 使用检索片段拼成 extractive lower bound，只验证评测合同、检索证据和失败暴露链路；只有 sufficiency=`supported` 才构造该 lower-bound answer。**不得解释为生产模型真实回答质量**。真实模型或 Provider replay 必须单独标识运行来源、模型配置和成本。 +确定性 CI baseline 使用检索片段拼成 extractive lower bound，只验证评测合同、检索证据和失败暴露链路；只有 sufficiency=`supported` 才构造该 lower-bound answer。**不得解释为生产模型真实回答质量**。 -运行： +确定性基线运行： ```bash python tools/run_rag_quality_baseline.py --output rag-quality-baseline.json ``` -CI 始终上传 `rag-quality-baseline` artifact。当前 baseline 为 `record_only`：不以分数阈值阻断合并，但 corpus 使用 SHA-256 fingerprint，并由 checked-in baseline snapshot 固定当前语料、资格 manifest、门控结果和关键指标。算法、corpus 或 evidence manifest 变化导致指标变化时，必须显式更新 snapshot，避免不同证据集合被悄悄当成同一基线比较。 +真实 Provider replay 运行： + +```bash +python tools/run_rag_provider_replay.py \ + --provider-profile \ + --model-profile pro +``` + +真实 replay 的默认原始报告位于 `output/rag-provider-replay.json`。只有实际 Provider 调用成功且报告 `status=completed` 时，才能讨论该次模型质量；`synthetic_test`、`provider_unavailable` 和 `partial_failure` 都不能被描述成完成的真实模型 benchmark。 + +CI 始终上传 `rag-quality-baseline` artifact。确定性 baseline 继续使用 corpus SHA-256 fingerprint 和 checked-in snapshot 固定语料、资格 manifest、门控结果和关键指标；K1d 另有不可回退质量合同。算法、corpus 或 evidence manifest 变化导致指标变化时，必须显式更新 snapshot，避免不同证据集合被悄悄当成同一基线比较。 -## 7. Module map +## 9. Module map | Module | Responsibility | |---|---| @@ -171,14 +220,16 @@ CI 始终上传 `rag-quality-baseline` artifact。当前 baseline 为 `record_on | `src/rag/backends.py` | vector backend contract 与 adapter | | `src/rag/query_plan.py` | pedagogy-aware private query planning | | `src/rag/sufficiency.py` | retrieval 后的 evidence sufficiency / refusal 决策 | +| `src/rag/source_coverage.py` | 复合问题的非回退 adaptive multi-source coverage | | `src/rag/eval.py` | 检索质量、scenario 与 stale-source 评测 | | `src/rag/answer_eval.py` | 回答可信度与引用支持评测 | +| `src/rag/provider_replay.py` | real/synthetic provenance、Provider replay 与 usage/latency 报告 | | `src/application/rag_service.py` | RagRun application orchestration | | `src/repositories/rag_repository.py` | durable RagRun 和 index-state persistence | | `src/api/routes/rag_routes.py` | FastAPI adapter | | `frontend/src/features/rag/*` | React controller 与资料/来源展示层 | -## 8. Invariants +## 10. Invariants 1. 删除、重建或资格变更后的非 active chunk 不得从任何普通 active retrieval backend 返回。 2. 必需阶段失败不得激活新 index version。 @@ -187,8 +238,11 @@ CI 始终上传 `rag-quality-baseline` artifact。当前 baseline 为 `record_on 5. backend-vector 返回值必须再次受当前 active chunk identity 约束，不能信任外部 backend 的旧状态。 6. `uncertain / insufficient` retrieval candidate 只能保留在 debug，不能作为普通 answer evidence 或用户 citation。 7. retrieval constraint 只用于模型私有规划，不得被披露成来源。 -8. `local_hash` 不得被描述为生产语义 embedding。 -9. 检索证据默认私有，只有 disclosure policy 允许的完整 evidence unit 进入模型回答。 -10. RAG 候选、采用证据和 UI 引用计数必须保持一致。 -11. 运行状态、错误和部分成功必须进入 durable RagRun，不用普通日志替代产品状态。 -12. deterministic baseline 的 corpus fingerprint、sufficiency summary 或 snapshot 指标变化必须显式进入代码审查，不能静默覆盖。 +8. adaptive source coverage 不得丢失 raw top-K 已召回的唯一来源。 +9. `local_hash` 不得被描述为生产语义 embedding。 +10. 检索证据默认私有，只有 disclosure policy 允许的完整 evidence unit 进入模型回答。 +11. RAG 候选、采用证据和 UI 引用计数必须保持一致。 +12. 运行状态、错误和部分成功必须进入 durable RagRun，不用普通日志替代产品状态。 +13. deterministic baseline 的 corpus fingerprint、sufficiency summary 或 snapshot 指标变化必须显式进入代码审查，不能静默覆盖。 +14. synthetic replay 无论对象如何自报，都不得标记为 completed real-provider benchmark。 +15. real-provider report 不得持久化 API key 或 raw endpoint；未完成调用不得补造质量分数。 diff --git a/src/rag/provider_replay.py b/src/rag/provider_replay.py new file mode 100644 index 0000000..8655fca --- /dev/null +++ b/src/rag/provider_replay.py @@ -0,0 +1,452 @@ +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Literal, Protocol + +from src.llm_client import ModelProfile, get_client, get_model_name, get_provider_settings +from src.rag.answer_eval import ( + RagAnswerAssertion, + RagAnswerCandidate, + RagAnswerEvalCase, + evaluate_answer_suite, +) + +ReplayKind = Literal["real_provider", "synthetic_test"] +ReplayRunStatus = Literal[ + "completed", + "partial_failure", + "provider_unavailable", +] + +_REPLAY_SYSTEM_PROMPT = """You are evaluating source-grounded answers for Study Agent. +Use only the evidence supplied in the user message. Never fill missing facts from memory. +Return one JSON object and no markdown fences: +{ + "refused": true or false, + "answer": "concise answer or refusal", + "assertions": [ + {"text": "one factual assertion", "cited_sources": ["exact-source-name.md"]} + ] +} +Rules: +- If EVIDENCE_STATUS is not `found`, set refused=true and do not make the requested factual claim. +- If EVIDENCE_STATUS is `found`, answer only claims directly supported by EVIDENCE. +- cited_sources must use only exact SOURCE identifiers shown in EVIDENCE. +- Split materially different claims into separate assertions. +- Do not cite a source that does not support the assertion. +""" + + +@dataclass(frozen=True) +class ProviderUsage: + prompt_tokens: int | None = None + completion_tokens: int | None = None + total_tokens: int | None = None + + def to_dict(self) -> dict[str, int | None]: + return asdict(self) + + +@dataclass(frozen=True) +class ProviderCompletion: + content: str + provider_profile: str + model_name: str + endpoint_fingerprint: str + response_id: str + finish_reason: str + latency_seconds: float + usage: ProviderUsage + + def to_dict(self) -> dict[str, Any]: + return { + "provider_profile": self.provider_profile, + "model_name": self.model_name, + "endpoint_fingerprint": self.endpoint_fingerprint, + "response_id": self.response_id, + "finish_reason": self.finish_reason, + "latency_seconds": round(self.latency_seconds, 6), + "usage": self.usage.to_dict(), + } + + +class ReplayProvider(Protocol): + def complete(self, messages: list[dict[str, str]]) -> ProviderCompletion: + ... + + +class OpenAICompatibleReplayProvider: + """Use Study Agent's existing provider owner for a real answer replay.""" + + replay_kind: ReplayKind = "real_provider" + + def __init__( + self, + *, + provider_profile: str | None = None, + model_profile: ModelProfile = "pro", + temperature: float = 0.0, + max_tokens: int = 700, + timeout: float = 60.0, + ) -> None: + settings = get_provider_settings(provider_profile) + self.provider_profile = settings.profile_name + self.model_profile = model_profile + self.model_name = get_model_name( + model_profile, + provider_profile=self.provider_profile, + ) + self.temperature = temperature + self.max_tokens = max_tokens + self.timeout = timeout + self.endpoint_fingerprint = _fingerprint_endpoint(settings.base_url) + self._client = get_client(provider_profile=self.provider_profile) + + def complete(self, messages: list[dict[str, str]]) -> ProviderCompletion: + started = time.perf_counter() + response = self._client.chat.completions.create( + model=self.model_name, + messages=messages, + temperature=self.temperature, + max_tokens=self.max_tokens, + timeout=self.timeout, + stream=False, + response_format={"type": "json_object"}, + ) + latency = time.perf_counter() - started + choice = response.choices[0] + usage = getattr(response, "usage", None) + return ProviderCompletion( + content=choice.message.content or "", + provider_profile=self.provider_profile, + model_name=self.model_name, + endpoint_fingerprint=self.endpoint_fingerprint, + response_id=str(getattr(response, "id", "") or ""), + finish_reason=str(getattr(choice, "finish_reason", "") or ""), + latency_seconds=latency, + usage=ProviderUsage( + prompt_tokens=_optional_int(getattr(usage, "prompt_tokens", None)), + completion_tokens=_optional_int( + getattr(usage, "completion_tokens", None) + ), + total_tokens=_optional_int(getattr(usage, "total_tokens", None)), + ), + ) + + +@dataclass(frozen=True) +class ReplayRetrieval: + status: str + reason: str + context: str + results: tuple[Any, ...] + + +@dataclass(frozen=True) +class ParsedProviderAnswer: + candidate: RagAnswerCandidate + parse_error: str = "" + + +def build_replay_messages( + case: RagAnswerEvalCase, + retrieval: ReplayRetrieval, +) -> list[dict[str, str]]: + evidence_blocks: list[str] = [] + for result in retrieval.results: + chunk = result.chunk + source_name = Path(chunk.source_path).name + evidence_blocks.append( + f"SOURCE: {source_name}\n" + f"TITLE: {chunk.title}\n" + f"TEXT:\n{chunk.text.strip()}" + ) + evidence = "\n\n---\n\n".join(evidence_blocks) or "(no eligible evidence)" + limitation = retrieval.context.strip() if retrieval.status != "found" else "" + user_content = ( + f"QUESTION:\n{case.query}\n\n" + f"EVIDENCE_STATUS: {retrieval.status}\n" + f"EVIDENCE_REASON: {retrieval.reason}\n" + f"EVIDENCE:\n{evidence}" + ) + if limitation: + user_content += f"\n\nEVIDENCE_LIMITATION:\n{limitation}" + return [ + {"role": "system", "content": _REPLAY_SYSTEM_PROMPT}, + {"role": "user", "content": user_content}, + ] + + +def parse_provider_answer(case_id: str, raw_text: str) -> ParsedProviderAnswer: + cleaned = _strip_json_fence(raw_text) + try: + payload = json.loads(cleaned) + if not isinstance(payload, dict): + raise ValueError("provider answer must be a JSON object") + except (json.JSONDecodeError, ValueError) as exc: + stripped = raw_text.strip() + fallback = RagAnswerCandidate( + case_id=case_id, + answer=stripped, + cited_sources=(), + refused=False, + assertions=( + RagAnswerAssertion(text=stripped, cited_sources=()), + ) + if stripped + else (), + ) + return ParsedProviderAnswer( + candidate=fallback, + parse_error=f"{type(exc).__name__}: {exc}", + ) + + refused = bool(payload.get("refused", False)) + answer = str(payload.get("answer", "") or "").strip() + assertions: list[RagAnswerAssertion] = [] + raw_assertions = payload.get("assertions", []) + if isinstance(raw_assertions, list): + for item in raw_assertions: + if not isinstance(item, dict): + continue + text = str(item.get("text", "") or "").strip() + if not text: + continue + raw_sources = item.get("cited_sources", []) + sources = ( + tuple(str(source) for source in raw_sources if str(source).strip()) + if isinstance(raw_sources, list) + else () + ) + assertions.append( + RagAnswerAssertion( + text=text, + cited_sources=sources, + ) + ) + cited_sources = _unique_sources( + source + for assertion in assertions + for source in assertion.cited_sources + ) + return ParsedProviderAnswer( + candidate=RagAnswerCandidate( + case_id=case_id, + answer=answer, + cited_sources=cited_sources, + refused=refused, + assertions=tuple(assertions), + ) + ) + + +def run_provider_answer_replay( + *, + cases: tuple[RagAnswerEvalCase, ...], + retrieve_case: Callable[[RagAnswerEvalCase], ReplayRetrieval], + provider: ReplayProvider, + corpus_fingerprint: str, +) -> dict[str, Any]: + case_rows: list[dict[str, Any]] = [] + candidates: list[RagAnswerCandidate] = [] + failed_cases: list[str] = [] + total_prompt_tokens = 0 + total_completion_tokens = 0 + total_tokens = 0 + usage_complete = True + total_latency = 0.0 + + for case in cases: + retrieval = retrieve_case(case) + messages = build_replay_messages(case, retrieval) + prompt_fingerprint = _fingerprint_messages(messages) + try: + completion = provider.complete(messages) + except Exception as exc: + failed_cases.append(case.case_id) + case_rows.append( + { + "case_id": case.case_id, + "query": case.query, + "status": "provider_failed", + "retrieval_status": retrieval.status, + "retrieval_reason": retrieval.reason, + "prompt_fingerprint": prompt_fingerprint, + "error_type": type(exc).__name__, + } + ) + continue + + parsed = parse_provider_answer(case.case_id, completion.content) + candidates.append(parsed.candidate) + total_latency += completion.latency_seconds + usage = completion.usage + if ( + usage.prompt_tokens is None + or usage.completion_tokens is None + or usage.total_tokens is None + ): + usage_complete = False + else: + total_prompt_tokens += usage.prompt_tokens + total_completion_tokens += usage.completion_tokens + total_tokens += usage.total_tokens + case_rows.append( + { + "case_id": case.case_id, + "query": case.query, + "status": "completed", + "retrieval_status": retrieval.status, + "retrieval_reason": retrieval.reason, + "retrieved_sources": [ + Path(result.chunk.source_path).name + for result in retrieval.results + ], + "prompt_fingerprint": prompt_fingerprint, + "provider": completion.to_dict(), + "parse_error": parsed.parse_error, + "candidate": { + "answer": parsed.candidate.answer, + "refused": parsed.candidate.refused, + "cited_sources": list(parsed.candidate.cited_sources), + "assertions": [ + { + "text": assertion.text, + "cited_sources": list(assertion.cited_sources), + } + for assertion in parsed.candidate.assertions + ], + }, + } + ) + + status: ReplayRunStatus = ( + "completed" if not failed_cases else "partial_failure" + ) + answer_quality: dict[str, Any] | None = None + if not failed_cases and len(candidates) == len(cases): + answer_quality = evaluate_answer_suite(cases, tuple(candidates)).to_dict() + + provider_rows = [row["provider"] for row in case_rows if "provider" in row] + return { + "schema_version": 1, + "replay_kind": _replay_kind_for_provider(provider), + "status": status, + "generated_at": datetime.now(timezone.utc).isoformat(), + "corpus_fingerprint": corpus_fingerprint, + "prompt_template_fingerprint": hashlib.sha256( + _REPLAY_SYSTEM_PROMPT.encode("utf-8") + ).hexdigest(), + "provider": _provider_identity(provider_rows), + "cases": len(cases), + "completed_cases": len(candidates), + "failed_cases": failed_cases, + "latency": { + "total_seconds": round(total_latency, 6), + "mean_seconds": ( + round(total_latency / len(candidates), 6) + if candidates + else 0.0 + ), + }, + "usage": { + "complete": usage_complete and bool(candidates), + "prompt_tokens": total_prompt_tokens if usage_complete else None, + "completion_tokens": total_completion_tokens if usage_complete else None, + "total_tokens": total_tokens if usage_complete else None, + }, + "answer_quality": answer_quality, + "results": case_rows, + } + + +def provider_unavailable_report( + *, + corpus_fingerprint: str, + provider_profile: str, + reason: str, +) -> dict[str, Any]: + return { + "schema_version": 1, + "replay_kind": "real_provider", + "status": "provider_unavailable", + "generated_at": datetime.now(timezone.utc).isoformat(), + "corpus_fingerprint": corpus_fingerprint, + "prompt_template_fingerprint": hashlib.sha256( + _REPLAY_SYSTEM_PROMPT.encode("utf-8") + ).hexdigest(), + "provider": {"provider_profile": provider_profile}, + "reason": reason, + "cases": 0, + "completed_cases": 0, + "failed_cases": [], + "answer_quality": None, + "results": [], + } + + +def _replay_kind_for_provider(provider: ReplayProvider) -> ReplayKind: + if isinstance(provider, OpenAICompatibleReplayProvider): + return "real_provider" + return "synthetic_test" + + +def _provider_identity(rows: list[dict[str, Any]]) -> dict[str, Any]: + if not rows: + return {} + first = rows[0] + return { + "provider_profile": first.get("provider_profile", ""), + "model_name": first.get("model_name", ""), + "endpoint_fingerprint": first.get("endpoint_fingerprint", ""), + } + + +def _fingerprint_endpoint(base_url: str) -> str: + normalized = (base_url or "").strip().rstrip("/") + return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16] + + +def _fingerprint_messages(messages: list[dict[str, str]]) -> str: + payload = json.dumps( + messages, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _strip_json_fence(text: str) -> str: + cleaned = (text or "").strip() + if cleaned.startswith("```") and cleaned.endswith("```"): + lines = cleaned.splitlines() + if len(lines) >= 3: + return "\n".join(lines[1:-1]).strip() + return cleaned + + +def _optional_int(value: Any) -> int | None: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _unique_sources(values) -> tuple[str, ...]: + result: list[str] = [] + seen: set[str] = set() + for value in values: + source = str(value).strip() + if not source or source in seen: + continue + seen.add(source) + result.append(source) + return tuple(result) diff --git a/tests/test_rag_provider_replay.py b/tests/test_rag_provider_replay.py new file mode 100644 index 0000000..7f326c2 --- /dev/null +++ b/tests/test_rag_provider_replay.py @@ -0,0 +1,245 @@ +from __future__ import annotations + +import json +from types import SimpleNamespace + +from src.rag import build_rag_index +from src.rag.answer_eval import RagAnswerEvalCase, RagExpectedClaim +from src.rag.provider_replay import ( + OpenAICompatibleReplayProvider, + ProviderCompletion, + ProviderUsage, + ReplayRetrieval, + build_replay_messages, + parse_provider_answer, + run_provider_answer_replay, +) +from src.rag.service import search_documents + + +class _SyntheticProvider: + replay_kind = "synthetic_test" + + def complete(self, messages): + user = messages[-1]["content"] + if "EVIDENCE_STATUS: found" in user: + content = json.dumps( + { + "refused": False, + "answer": "A Session uses a connection pool and explicit timeout settings still matter.", + "assertions": [ + { + "text": "Session uses a connection pool.", + "cited_sources": ["requests.md"], + }, + { + "text": "Explicit timeout settings still matter.", + "cited_sources": ["requests.md"], + }, + ], + } + ) + else: + content = json.dumps( + { + "refused": True, + "answer": "The supplied evidence does not support that fact.", + "assertions": [], + } + ) + return ProviderCompletion( + content=content, + provider_profile="synthetic", + model_name="fixture-model", + endpoint_fingerprint="fixture-endpoint", + response_id="fixture-response", + finish_reason="stop", + latency_seconds=0.01, + usage=ProviderUsage( + prompt_tokens=10, + completion_tokens=5, + total_tokens=15, + ), + ) + + +class _SpoofedRealProvider(_SyntheticProvider): + replay_kind = "real_provider" + + +def _answerable_case() -> RagAnswerEvalCase: + return RagAnswerEvalCase( + case_id="answerable", + query="How does Session pooling work and what timeout setting matters?", + answerable=True, + expected_sources=("requests.md",), + expected_claims=( + RagExpectedClaim( + claim_id="pool", + match_terms=("Session", "connection pool"), + support_sources=("requests.md",), + ), + RagExpectedClaim( + claim_id="timeout", + match_terms=("timeout",), + support_sources=("requests.md",), + ), + ), + ) + + +def _unanswerable_case() -> RagAnswerEvalCase: + return RagAnswerEvalCase( + case_id="unanswerable", + query="Which exact GPU is required?", + answerable=False, + expected_sources=(), + expected_claims=(), + ) + + +def _fixture_retrieval(tmp_path): + document = tmp_path / "requests.md" + document.write_text( + "A Session uses a connection pool. Explicit timeout settings still matter.", + encoding="utf-8", + ) + index = build_rag_index([document], max_chars=300, overlap_chars=0) + found_results = tuple( + search_documents( + index, + _answerable_case().query, + top_k=2, + retrieval_mode="hybrid", + ) + ) + + def retrieve_case(case): + if case.answerable: + return ReplayRetrieval( + status="found", + reason="fixture", + context="", + results=found_results, + ) + return ReplayRetrieval( + status="insufficient", + reason="missing_explicit_anchor_concepts", + context="The active corpus does not support the requested fact.", + results=(), + ) + + return retrieve_case + + +def test_synthetic_replay_cannot_be_reported_as_real_provider(tmp_path): + report = run_provider_answer_replay( + cases=(_answerable_case(), _unanswerable_case()), + retrieve_case=_fixture_retrieval(tmp_path), + provider=_SyntheticProvider(), + corpus_fingerprint="fixture-corpus", + ) + + assert report["replay_kind"] == "synthetic_test" + assert report["status"] == "completed" + assert report["answer_quality"]["answerability_accuracy"] == 1.0 + assert report["answer_quality"]["mean_citation_recall"] == 1.0 + assert report["usage"] == { + "complete": True, + "prompt_tokens": 20, + "completion_tokens": 10, + "total_tokens": 30, + } + + +def test_provider_cannot_spoof_real_replay_kind_with_a_string_flag(tmp_path): + report = run_provider_answer_replay( + cases=(_answerable_case(),), + retrieve_case=_fixture_retrieval(tmp_path), + provider=_SpoofedRealProvider(), + corpus_fingerprint="fixture-corpus", + ) + + assert report["replay_kind"] == "synthetic_test" + + +def test_real_provider_adapter_records_usage_without_persisting_api_key(monkeypatch): + class _FakeCompletions: + def create(self, **_kwargs): + return SimpleNamespace( + id="response-1", + choices=[ + SimpleNamespace( + finish_reason="stop", + message=SimpleNamespace( + content='{"refused":true,"answer":"no evidence","assertions":[]}' + ), + ) + ], + usage=SimpleNamespace( + prompt_tokens=12, + completion_tokens=4, + total_tokens=16, + ), + ) + + fake_client = SimpleNamespace( + chat=SimpleNamespace(completions=_FakeCompletions()) + ) + fake_key = "test-placeholder-value" + fake_endpoint = "https://provider.example/v1" + monkeypatch.setattr( + "src.rag.provider_replay.get_provider_settings", + lambda _profile=None: SimpleNamespace( + profile_name="deepseek", + api_key=fake_key, + base_url=fake_endpoint, + ), + ) + monkeypatch.setattr( + "src.rag.provider_replay.get_model_name", + lambda _model_profile, provider_profile=None: "real-model", + ) + monkeypatch.setattr( + "src.rag.provider_replay.get_client", + lambda provider_profile=None: fake_client, + ) + + provider = OpenAICompatibleReplayProvider(provider_profile="deepseek") + completion = provider.complete( + [{"role": "user", "content": "Return JSON."}] + ) + serialized = json.dumps(completion.to_dict()) + + assert provider.replay_kind == "real_provider" + assert completion.provider_profile == "deepseek" + assert completion.model_name == "real-model" + assert completion.usage.total_tokens == 16 + assert fake_key not in serialized + assert "provider.example" not in serialized + + +def test_insufficient_replay_prompt_exposes_no_citable_evidence(): + case = _unanswerable_case() + messages = build_replay_messages( + case, + ReplayRetrieval( + status="insufficient", + reason="missing_explicit_anchor_concepts", + context="Do not present a direct answer as grounded in the user's materials.", + results=(), + ), + ) + + user_message = messages[-1]["content"] + assert "EVIDENCE_STATUS: insufficient" in user_message + assert "(no eligible evidence)" in user_message + assert "SOURCE:" not in user_message + + +def test_malformed_provider_json_is_not_silently_treated_as_a_refusal(): + parsed = parse_provider_answer("case-1", "not-json") + + assert parsed.parse_error + assert parsed.candidate.refused is False + assert parsed.candidate.answer == "not-json" diff --git a/tools/run_rag_provider_replay.py b/tools/run_rag_provider_replay.py new file mode 100644 index 0000000..9c20b2e --- /dev/null +++ b/tools/run_rag_provider_replay.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +import argparse +import json +import os +import tempfile +from pathlib import Path +from typing import Any + +from src.rag import build_rag_index, save_rag_index +from src.rag.answer_eval import RagAnswerEvalCase, load_answer_eval_fixture +from src.rag.provider_replay import ( + OpenAICompatibleReplayProvider, + ReplayRetrieval, + provider_unavailable_report, + run_provider_answer_replay, +) +from src.rag.eval import load_eval_cases +from src.tools.local_knowledge import retrieve_local_knowledge +from tools.run_rag_quality_baseline import ( + FIXTURE_DOCS, + _apply_evidence_manifest, + _corpus_fingerprint, +) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Run the K1e source-grounded answer benchmark through Study Agent's " + "configured real LLM provider." + ) + ) + parser.add_argument( + "--fixture-dir", + default="tests/fixtures/rag_eval", + help="Directory containing the fixed K1 RAG corpus and gold cases.", + ) + parser.add_argument( + "--output", + default="output/rag-provider-replay.json", + help="JSON report path. Defaults to the gitignored output directory.", + ) + parser.add_argument( + "--provider-profile", + default=None, + help="Study Agent provider profile. Defaults to LLM_PROVIDER_PROFILE/openai.", + ) + parser.add_argument( + "--model-profile", + choices=("flash", "pro"), + default="pro", + ) + parser.add_argument("--temperature", type=float, default=0.0) + parser.add_argument("--max-tokens", type=int, default=700) + parser.add_argument("--timeout", type=float, default=60.0) + parser.add_argument( + "--case-id", + action="append", + default=[], + help="Run only selected case IDs. Repeat for multiple cases.", + ) + return parser + + +def _fixture_paths(fixture_dir: Path) -> tuple[list[Path], list[Path]]: + documents = [fixture_dir / name for name in FIXTURE_DOCS] + metadata = [ + fixture_dir / "cases.json", + fixture_dir / "answer_cases.json", + fixture_dir / "evidence_manifest.json", + ] + return documents, metadata + + +def _write_report(path: Path, report: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def _resolved_provider_profile(value: str | None) -> str: + return (value or os.getenv("LLM_PROVIDER_PROFILE") or "openai").strip().lower() + + +def main() -> int: + args = _parser().parse_args() + fixture_dir = Path(args.fixture_dir) + document_paths, metadata_paths = _fixture_paths(fixture_dir) + corpus_paths = [*document_paths, *metadata_paths] + missing = [str(path) for path in corpus_paths if not path.is_file()] + if missing: + raise FileNotFoundError( + f"Missing K1e replay fixture files: {', '.join(missing)}" + ) + + corpus_fingerprint = _corpus_fingerprint(corpus_paths, fixture_dir) + answer_cases, _ = load_answer_eval_fixture(fixture_dir / "answer_cases.json") + if args.case_id: + selected = set(args.case_id) + answer_cases = tuple(case for case in answer_cases if case.case_id in selected) + missing_case_ids = sorted(selected - {case.case_id for case in answer_cases}) + if missing_case_ids: + raise ValueError( + "Unknown K1e case IDs: " + ", ".join(missing_case_ids) + ) + retrieval_cases = load_eval_cases(fixture_dir / "cases.json") + retrieval_by_id = {case.case_id: case for case in retrieval_cases} + + provider_profile = _resolved_provider_profile(args.provider_profile) + try: + provider = OpenAICompatibleReplayProvider( + provider_profile=provider_profile, + model_profile=args.model_profile, + temperature=args.temperature, + max_tokens=args.max_tokens, + timeout=args.timeout, + ) + except Exception as exc: + report = provider_unavailable_report( + corpus_fingerprint=corpus_fingerprint, + provider_profile=provider_profile, + reason=f"provider_initialization_failed:{type(exc).__name__}", + ) + _write_report(Path(args.output), report) + print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) + return 2 + + index = build_rag_index(document_paths, max_chars=700, overlap_chars=80) + index = _apply_evidence_manifest(index, fixture_dir / "evidence_manifest.json") + + with tempfile.TemporaryDirectory(prefix="study-agent-k1e-") as temp_dir: + index_path = Path(temp_dir) / "rag_index.json" + save_rag_index(index, index_path) + + def retrieve_case(case: RagAnswerEvalCase) -> ReplayRetrieval: + retrieval_case = retrieval_by_id.get(case.case_id) + top_k = retrieval_case.top_k if retrieval_case is not None else 4 + retrieval_mode = ( + retrieval_case.retrieval_mode + if retrieval_case is not None + else "hybrid" + ) + result = retrieve_local_knowledge( + case.query, + enabled=True, + force=True, + index_path=index_path, + top_k=top_k, + min_score=0.01, + retrieval_mode=retrieval_mode, + allow_rewrite=True, + ) + return ReplayRetrieval( + status=result.status, + reason=result.reason, + context=result.context, + results=tuple(result.results), + ) + + report = run_provider_answer_replay( + cases=answer_cases, + retrieve_case=retrieve_case, + provider=provider, + corpus_fingerprint=corpus_fingerprint, + ) + + report["scope"] = { + "case_ids": [case.case_id for case in answer_cases], + "full_gold_suite": not bool(args.case_id), + } + _write_report(Path(args.output), report) + compact = { + "replay_kind": report["replay_kind"], + "status": report["status"], + "corpus_fingerprint": report["corpus_fingerprint"], + "provider": report["provider"], + "cases": report["cases"], + "completed_cases": report["completed_cases"], + "latency": report["latency"], + "usage": report["usage"], + "answer_quality": report["answer_quality"], + } + print(json.dumps(compact, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if report["status"] == "completed" else 3 + + +if __name__ == "__main__": + raise SystemExit(main())