"""RetrievalResultEvaluator - validates retrieved content via extract assertions. Filters retrieval hits by ``max_rank``, then judges each ``retrievalExtract_contains`` assertion against the candidate hits **one at a time, in ascending rank order, stopping at the first hit whose evidence supports the claim** (first-hit matching). Each judgment is delegated to an LLM judge (see :mod:`.extract_judge`) that decides whether the assertion is *supported by* that hit's evidence. The rank and URL of the supporting hit are recorded, so a match carries precise per-hit attribution. Returns a score dict with the pass/fail status, a diagnostic code, and the matched/missing items (see ``contracts/retrieval-result-evaluator.md``). ``retrievalExtract_contains`` is the sole hit-matching assertion. URL-based matching is deferred to V2 (see ``spec.md`` Out of Scope): the matched hit's ``webUrl`` is surfaced in ``matched_items[].matchedHitUrl`` for debugging only and is never compared. """ from __future__ import annotations from typing import Any, Dict, List, Optional from cli_logging.cli_logger import emit_structured_log from cli_logging.logging_utils import Operation from common import STATUS_FAIL, STATUS_PASS from .extract_judge import ExtractJudge from .utils import ( FAIL_SCORE, PASS_SCORE, THRESHOLD, extract_texts, flatten_hits_with_execution_context, get_retrieval_executions, has_retrieval_failure, ) _DEFAULT_MAX_RANK = 10 # Upper bound on the per-hit evidence blob handed to the judge, to keep token # usage bounded when a hit carries many or very long extracts. _MAX_EVIDENCE_CHARS = 24000 # Separator between a hit's individual extract texts in its evidence blob. _EVIDENCE_SEPARATOR = "\n---\n" # Surfaced when hits were retrieved but none carried any extract text to judge. _NO_EVIDENCE_REASON = "Retrieved hits contained no extract text to evaluate." # Markdown metacharacters backslash-escaped in user/model text so only the # per-item template (bold, bullets) formats when the reason is rendered. _MARKDOWN_SPECIAL = ("\\", "`", "*", "_", "[", "]", "(", ")") # Diagnostic_code values. _DIAG_PASS = "pass" _DIAG_NOT_RETRIEVED = "not_retrieved" _DIAG_COUNT_BELOW_MINIMUM = "count_below_minimum" _DIAG_ERROR = "error" class RetrievalResultEvaluator: """Validate that distinctive extract text appears in retrieved content. Configuration parameters (``capability``, ``expected_items``, ``min_expected_count``, ``max_rank``) are set via the constructor from evaluator config and are not accepted at call time. Constructor validation enforces: - ``capability`` is required (non-empty string) - At least one of ``expected_items`` or ``min_expected_count`` - Each expected_item has a non-empty ``retrievalExtract_contains`` - ``webUrl`` on an expected_item is rejected (V1 strict; see ADR) """ def __init__( self, capability: str, expected_items: Optional[List[Dict[str, Any]]] = None, min_expected_count: Optional[int] = None, max_rank: int = _DEFAULT_MAX_RANK, model_config: Any = None, extract_judge: Any = None, ) -> None: if not isinstance(capability, str) or not capability: raise ValueError( "RetrievalResultEvaluator: 'capability' is required and must be a non-empty string." ) if expected_items is not None and not isinstance(expected_items, list): raise ValueError( "RetrievalResultEvaluator: 'expected_items' must be a list when provided." ) if min_expected_count is not None: if not isinstance(min_expected_count, int) or isinstance(min_expected_count, bool): raise ValueError( "RetrievalResultEvaluator: 'min_expected_count' must be an integer." ) if min_expected_count < 1: raise ValueError( "RetrievalResultEvaluator: 'min_expected_count' must be >= 1." ) if not isinstance(max_rank, int) or isinstance(max_rank, bool) or max_rank < 1: raise ValueError( "RetrievalResultEvaluator: 'max_rank' must be a positive integer." ) normalized_items: List[Dict[str, Any]] = [] if expected_items: for idx, item in enumerate(expected_items): if not isinstance(item, dict): raise ValueError( f"RetrievalResultEvaluator: expected_items[{idx}] must be an object." ) if "webUrl" in item: raise ValueError( f"RetrievalResultEvaluator: expected_items[{idx}] uses 'webUrl', which is " "not supported in V1. Use 'retrievalExtract_contains' with a distinctive " "phrase from the expected resource's content instead. See spec.md " "Out of Scope for the V2 forward direction." ) text = item.get("retrievalExtract_contains") if not isinstance(text, str) or not text: raise ValueError( f"RetrievalResultEvaluator: expected_items[{idx}] must have a non-empty " "'retrievalExtract_contains' string." ) normalized_items.append({"retrievalExtract_contains": text}) if not normalized_items and min_expected_count is None: raise ValueError( "RetrievalResultEvaluator: at least one of 'expected_items' or " "'min_expected_count' must be configured." ) self.capability = capability self.expected_items = normalized_items self.min_expected_count = min_expected_count self.max_rank = max_rank self.model_config = model_config # A pre-built judge (test seam) takes precedence; otherwise one is built # lazily on first use so deterministic count-only runs never touch the # LLM dependency. self._judge = extract_judge def __call__( self, *, retrieval_telemetry: Optional[Dict[str, Any]] = None, response: Optional[str] = None, **kwargs: Any, ) -> Dict[str, Any]: """Evaluate retrieval telemetry against the configured assertions. Returns the score dict described in ``contracts/retrieval-result-evaluator.md`` Return Value Schema. """ expected_items = self.expected_items min_expected_count = self.min_expected_count executions = get_retrieval_executions( retrieval_telemetry, capability=self.capability ) if has_retrieval_failure(retrieval_telemetry, executions, capability=self.capability): return _error_result( "Retrieval telemetry unavailable or all matching executions failed" ) candidates = flatten_hits_with_execution_context( executions, max_rank=self.max_rank ) candidate_count = len(candidates) # Step 1: min_expected_count gate. A failed count makes the result fail # regardless of the extract assertions, so short-circuit here and skip # the (LLM-judged, costly) extract evaluation entirely. count_passed = min_expected_count is None or candidate_count >= min_expected_count if not count_passed: reason = ( f"Retrieved {candidate_count} results, expected at least {min_expected_count}" ) if expected_items: reason += "; extract assertions not evaluated because the count gate failed" return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=_DIAG_COUNT_BELOW_MINIMUM, reason=reason, results_evaluated=candidate_count, ) # Count-only path: count met, no items to check. if not expected_items: return _build_result( score=PASS_SCORE, status=STATUS_PASS, diagnostic_code=_DIAG_PASS, reason=f"{candidate_count} results retrieved (>= {min_expected_count})", results_evaluated=candidate_count, ) if not candidates: return _not_retrieved_result( "No retrieval results found in matching retrieval executions", expected_items=expected_items, ) try: judge = self._get_judge() except Exception as exc: return _error_result(f"Failed to initialize the extract judge: {exc}") if judge is None: return _error_result( "retrievalExtract_contains requires Azure OpenAI, but no model " "configuration is available. Configure Azure OpenAI or remove the " "extract assertions." ) # Step 2: judge each expected item against candidate hits in rank order, # stopping at the first hit whose evidence supports the claim (first-hit). # A match records the supporting hit's rank and URL for attribution. matched_items: List[Dict[str, Any]] = [] missing_items: List[Dict[str, Any]] = [] extract_failures: List[Dict[str, Any]] = [] # Build each hit's evidence blob once; every expected item is judged # against the same candidates, so there's no need to recompute per item. # Skip hits with no extract text: they can't support a claim, so # judging them only wastes an LLM call. The rest stay rank-sorted. hit_evidence = [ (hit, evidence) for hit in candidates for evidence in (_hit_evidence(hit),) if evidence ] # Memoize judgments for the duration of this call: duplicate assertion # text or hits sharing evidence produce identical (claim, evidence) pairs # that would otherwise be re-judged and re-billed. Scoped per call so it # never grows unbounded across the run. verdict_cache: Dict[tuple[str, str], Dict[str, Any]] = {} for expected in expected_items: extract_text = expected["retrievalExtract_contains"] if hit_evidence: matched_hit, verdict = _judge_first_supporting_hit( judge, extract_text, hit_evidence, verdict_cache ) reason = verdict.get("reason", "") else: # Hits were retrieved but none had evaluable text; nothing to judge. matched_hit, reason = None, _NO_EVIDENCE_REASON if matched_hit is not None: matched_items.append({ "extractMatch": extract_text, "matchedHitUrl": matched_hit.get("webUrl", ""), "matchedHitRank": matched_hit.get("rank"), "reason": reason, }) else: # The capability returned hits but none supported the assertion. missing = {"extractMatch": extract_text, "reason": reason} extract_failures.append(missing) missing_items.append(missing) items_score = len(matched_items) / len(expected_items) # Step 3: report items result. The count gate (if any) already passed - # a failed count short-circuited above before any judging. tally = ( f"{len(matched_items)}/{len(expected_items)} expected item(s) " f"supported by the retrieved evidence within rank {self.max_rank}." ) if items_score == 1.0: return _build_result( score=PASS_SCORE, status=STATUS_PASS, diagnostic_code=_DIAG_PASS, reason=tally + _format_item_reasons(matched_items), results_evaluated=candidate_count, matched_items=matched_items, ) # Partial/full miss: lead with the tally, then list what was supported # (when any) followed by what was not, mirroring the all-pass branch. sections = [tally] if matched_items: sections.append(f"Supported:{_format_item_reasons(matched_items)}") sections.append(f"Not supported:{_format_item_reasons(missing_items)}") return _build_result( score=items_score, status=STATUS_FAIL, diagnostic_code=_DIAG_NOT_RETRIEVED, reason="\n\n".join(sections), results_evaluated=candidate_count, matched_items=matched_items, missing_items=missing_items, extract_failures=extract_failures, ) def _get_judge(self) -> Any: """Return the extract judge, building one lazily from ``model_config``. Returns ``None`` when neither an injected judge nor a model configuration is available, signalling the caller to emit an error result rather than attempting to judge. """ if self._judge is None and self.model_config is not None: self._judge = ExtractJudge(self.model_config) return self._judge def _build_result( *, score: float, status: str, diagnostic_code: str, reason: str, results_evaluated: int = 0, matched_items: Optional[List[Dict[str, Any]]] = None, missing_items: Optional[List[Dict[str, Any]]] = None, extract_failures: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: return { "retrieval_result": score, "result": status, "threshold": THRESHOLD, "diagnostic_code": diagnostic_code, "reason": reason, "results_evaluated": results_evaluated, "matched_items": matched_items if matched_items is not None else [], "missing_items": missing_items if missing_items is not None else [], "extract_failures": extract_failures if extract_failures is not None else [], } def _escape_markdown(text: Any) -> str: """Backslash-escape Markdown metacharacters in user/model-provided text.""" s = str(text or "").replace("\r", " ").replace("\n", " ") for ch in _MARKDOWN_SPECIAL: s = s.replace(ch, f"\\{ch}") return s def _format_item_reasons(items: List[Dict[str, Any]]) -> str: """Render the judge's per-item rationale into one Markdown string. Keeps the top-level ``reason`` LLM-based (not a fixed template) by quoting each assertion alongside the judge's verdict reason. Falls back to the assertion text alone if the judge returned no reason. The assertion and reason are Markdown-escaped so only the bold/bullet template formats. """ lines: List[str] = [] for item in items: match = _escape_markdown(item.get("extractMatch", "")) reason = _escape_markdown(item.get("reason")) if item.get("reason") else "" lines.append(f"- **{match}**: {reason}" if reason else f"- **{match}**") return "\n\n" + "\n".join(lines) if lines else "" def _judge_first_supporting_hit( judge: Any, claim: str, hit_evidence: List[tuple[Dict[str, Any], str]], cache: Dict[tuple[str, str], Dict[str, Any]], ) -> tuple[Optional[Dict[str, Any]], Dict[str, Any]]: """Return the first candidate hit whose evidence supports ``claim``. Walks ``hit_evidence`` (already rank-sorted ``(hit, evidence)`` pairs, with empty-evidence hits filtered out by the caller) and judges the claim against each hit's evidence individually, short-circuiting on the first support. Returns ``(hit, verdict)`` for that hit, or ``(None, verdict)`` when no hit supports the claim - in which case the verdict is the best-ranked hit's, the most relevant rationale for why the claim went unsupported. ``cache`` memoizes verdicts by ``(claim, evidence)`` so an identical pair seen again (duplicate assertion text, or hits sharing evidence) skips the LLM call. The caller guarantees ``hit_evidence`` is non-empty. """ first_verdict: Dict[str, Any] = {"supported": False, "reason": ""} for index, (hit, evidence) in enumerate(hit_evidence): key = (claim, evidence) verdict = cache.get(key) if verdict is None: verdict = judge.judge(claim=claim, evidence=evidence) cache[key] = verdict if verdict.get("supported"): return hit, verdict if index == 0: first_verdict = verdict return None, first_verdict def _hit_evidence(hit: Dict[str, Any]) -> str: """Concatenate a single hit's deduped extract texts into one blob. Blank/whitespace-only and duplicate extract strings are dropped (so a hit with no real text yields ``""``, letting the caller skip it), and the result is truncated to ``_MAX_EVIDENCE_CHARS`` to bound the judge's token usage. """ seen = set() parts: List[str] = [] for text in extract_texts(hit): if not text.strip() or text in seen: continue seen.add(text) parts.append(text) blob = _EVIDENCE_SEPARATOR.join(parts) blob_len = len(blob) if blob_len > _MAX_EVIDENCE_CHARS: dropped_chars = blob_len - _MAX_EVIDENCE_CHARS blob = blob[:_MAX_EVIDENCE_CHARS] emit_structured_log( "warning", f"Truncated retrieved evidence to {_MAX_EVIDENCE_CHARS} chars " f"(dropped {dropped_chars}) for hit rank={hit.get('rank')} " f"url={hit.get('webUrl', '')!r}; " "text beyond the cap is not judged.", operation=Operation.EVALUATE, ) return blob def _error_result(reason: str) -> Dict[str, Any]: return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=_DIAG_ERROR, reason=reason, ) def _not_retrieved_result( reason: str, expected_items: List[Dict[str, Any]], ) -> Dict[str, Any]: """No candidate hits - every expected item is missing, no extract failures. Missing items carry an empty ``reason`` for shape-consistency with the judged paths: no hits were retrieved, so the judge never ran and there is no per-item rationale to report. """ return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=_DIAG_NOT_RETRIEVED, reason=reason, missing_items=[ {"extractMatch": item["retrievalExtract_contains"], "reason": ""} for item in expected_items ], )