| import json |
| import unicodedata |
| from pathlib import Path |
| from typing import Sequence |
|
|
| from .model import ReframrModel |
|
|
|
|
| META_VOICE_PHRASES = ( |
| "the answer should", |
| "the response should", |
| "a strong answer", |
| "a safe answer", |
| "the safe answer", |
| "the safe move", |
| "the passage", |
| ) |
|
|
| PROTOCOL_STARTS = ( |
| "<tool_call>", |
| "<tool_result>", |
| "<source>", |
| "<final>", |
| "<reason>", |
| "<answer>", |
| ) |
|
|
|
|
| def load_manifest(path: str | Path) -> dict[str, object]: |
| return json.loads(Path(path).read_text(encoding="utf-8")) |
|
|
|
|
| def _expected_next_token(model: ReframrModel, expected_text: str) -> str: |
| assert model.tokenizer is not None |
| encoded = model.tokenizer.encode(f" {expected_text}") |
| return encoded[0] if encoded else "" |
|
|
|
|
| def _normalize_text(text: str) -> str: |
| return " ".join(text.casefold().split()) |
|
|
|
|
| def _word_ngrams(words: list[str], size: int) -> list[tuple[str, ...]]: |
| if size <= 0 or len(words) < size: |
| return [] |
| return [tuple(words[index : index + size]) for index in range(len(words) - size + 1)] |
|
|
|
|
| def _distinct_ratio(words: list[str], size: int) -> float: |
| grams = _word_ngrams(words, size) |
| if not grams: |
| return 0.0 |
| return len(set(grams)) / len(grams) |
|
|
|
|
| def _repetition_ratio(words: list[str], size: int) -> float: |
| grams = _word_ngrams(words, size) |
| if not grams: |
| return 0.0 |
| repeated = len(grams) - len(set(grams)) |
| return repeated / len(grams) |
|
|
|
|
| def _source_replay_index( |
| sources: Sequence[str] | None, |
| *, |
| ngram_size: int, |
| ) -> list[tuple[str, set[tuple[str, ...]]]]: |
| if not sources: |
| return [] |
| index: list[tuple[str, set[tuple[str, ...]]]] = [] |
| for source in sources: |
| normalized = _normalize_text(str(source)) |
| grams = set(_word_ngrams(normalized.split(), ngram_size)) |
| if grams: |
| index.append((normalized, grams)) |
| return index |
|
|
|
|
| def _source_replay_overlap( |
| generated: str, |
| replay_index: list[tuple[str, set[tuple[str, ...]]]], |
| *, |
| ngram_size: int, |
| ) -> tuple[float, str]: |
| generated_grams = set(_word_ngrams(_normalize_text(generated).split(), ngram_size)) |
| if not generated_grams or not replay_index: |
| return 0.0, "" |
| best_overlap = 0.0 |
| best_source = "" |
| for normalized_source, source_grams in replay_index: |
| overlap = len(generated_grams & source_grams) / len(generated_grams) |
| if overlap > best_overlap: |
| best_overlap = overlap |
| best_source = normalized_source |
| return best_overlap, best_source |
|
|
|
|
| def _text_from_replay_row(row: object) -> str: |
| if isinstance(row, str): |
| return row.strip() |
| if not isinstance(row, dict): |
| return "" |
| for field in ("answer", "response", "chosen", "text", "content", "completion"): |
| value = row.get(field) |
| if isinstance(value, str) and value.strip(): |
| return value.strip() |
| if "messages" in row: |
| return _content_to_text(row["messages"]) |
| return "" |
|
|
|
|
| def load_replay_sources( |
| paths: Sequence[str | Path], |
| *, |
| limit: int = 10_000, |
| ) -> list[str]: |
| sources: list[str] = [] |
| for source_path in paths: |
| path = Path(source_path) |
| if not path.exists(): |
| continue |
| suffix = path.suffix.lower() |
| if suffix == ".jsonl": |
| for line in path.read_text(encoding="utf-8").splitlines(): |
| if limit > 0 and len(sources) >= limit: |
| return sources |
| if not line.strip(): |
| continue |
| text = _text_from_replay_row(json.loads(line)) |
| if text: |
| sources.append(text) |
| continue |
| if suffix == ".json": |
| payload = json.loads(path.read_text(encoding="utf-8")) |
| rows = payload.get("records", payload.get("texts", payload)) if isinstance(payload, dict) else payload |
| if isinstance(rows, list): |
| for row in rows: |
| if limit > 0 and len(sources) >= limit: |
| return sources |
| text = _text_from_replay_row(row) |
| if text: |
| sources.append(text) |
| else: |
| text = _text_from_replay_row(rows) |
| if text: |
| sources.append(text) |
| continue |
| text = path.read_text(encoding="utf-8").strip() |
| if text: |
| sources.append(text) |
| if limit > 0 and len(sources) >= limit: |
| return sources[:limit] |
| return sources[:limit] if limit > 0 else sources |
|
|
|
|
| def _normalize_phrase_list(value: object) -> list[str]: |
| if not isinstance(value, list): |
| return [] |
| phrases: list[str] = [] |
| for item in value: |
| if isinstance(item, str): |
| phrase = item.strip() |
| if phrase: |
| phrases.append(phrase) |
| return phrases |
|
|
|
|
| def _normalize_required_groups(value: object) -> list[list[str]]: |
| if not isinstance(value, list): |
| return [] |
| groups: list[list[str]] = [] |
| for raw_group in value: |
| if isinstance(raw_group, list): |
| group = [ |
| str(term).casefold().strip() |
| for term in raw_group |
| if str(term).strip() |
| ] |
| else: |
| term = str(raw_group).casefold().strip() |
| group = [term] if term else [] |
| if group: |
| groups.append(group) |
| return groups |
|
|
|
|
| def _required_group_summary( |
| normalized_text: str, |
| required_groups: object, |
| ) -> tuple[int, int, float]: |
| groups = _normalize_required_groups(required_groups) |
| hit_count = sum( |
| 1 |
| for group in groups |
| if any(term in normalized_text for term in group) |
| ) |
| group_count = len(groups) |
| coverage = hit_count / group_count if group_count else 0.0 |
| return hit_count, group_count, coverage |
|
|
|
|
| def _banned_phrase_hit(normalized_text: str, banned_phrases: object) -> bool: |
| return any( |
| _normalize_text(phrase) in normalized_text |
| for phrase in _normalize_phrase_list(banned_phrases) |
| if _normalize_text(phrase) |
| ) |
|
|
|
|
| def _meta_voice_hit(normalized_text: str) -> bool: |
| return any(phrase in normalized_text for phrase in META_VOICE_PHRASES) |
|
|
|
|
| def _has_malformed_sentence_start(text: str) -> bool: |
| stripped = text.strip() |
| if not stripped: |
| return True |
| if any(stripped.startswith(protocol) for protocol in PROTOCOL_STARTS): |
| return False |
| leading_quote = False |
| for character in stripped: |
| if character.isspace(): |
| continue |
| category = unicodedata.category(character) |
| if category.startswith(("P", "S")): |
| if character in {"'", '"', "‘", "’", "“", "”"}: |
| leading_quote = True |
| continue |
| if character.isalpha(): |
| if leading_quote: |
| return False |
| return character.islower() |
| return False |
| return False |
|
|
|
|
| def _quality_gate_passed( |
| *, |
| word_count: int, |
| punctuation_hit: bool, |
| required_group_coverage: float, |
| exact_copy: bool, |
| banned_phrase_hit: bool, |
| meta_voice_hit: bool, |
| malformed_start: bool, |
| repetition_3: float, |
| tool_call_hit: bool, |
| fabricated_tool_result_hit: bool, |
| fabricated_source_hit: bool, |
| source_replay_hit: bool, |
| item: dict[str, object], |
| ) -> bool: |
| blocking_failure = any( |
| ( |
| exact_copy, |
| banned_phrase_hit, |
| meta_voice_hit, |
| malformed_start, |
| fabricated_tool_result_hit, |
| fabricated_source_hit, |
| source_replay_hit, |
| ) |
| ) |
| if bool(item.get("allow_tool_call", False)) and tool_call_hit: |
| return not blocking_failure |
|
|
| min_words = int(item.get("min_words", 1)) |
| required_min_coverage = float( |
| item.get( |
| "min_required_group_coverage", |
| 1.0 if item.get("required_groups") else 0.0, |
| ) |
| ) |
| require_punctuation = bool(item.get("require_punctuation", False)) |
| max_repetition_3 = float(item.get("max_repetition_3", 0.35)) |
| if ( |
| _item_contains_source_evidence(item) |
| and required_group_coverage >= required_min_coverage |
| and (punctuation_hit or not require_punctuation) |
| and repetition_3 <= max_repetition_3 |
| ): |
| return not blocking_failure |
| if word_count < min_words: |
| return False |
| if required_group_coverage < required_min_coverage: |
| return False |
| if require_punctuation and not punctuation_hit: |
| return False |
| if repetition_3 > max_repetition_3: |
| return False |
| return not blocking_failure |
|
|
|
|
| def _item_contains_source_evidence(value: object) -> bool: |
| if isinstance(value, dict): |
| sources = value.get("sources") |
| if isinstance(sources, list) and any(isinstance(source, dict) for source in sources): |
| return True |
| if {"title", "url", "snippet"}.intersection(value.keys()) and ( |
| value.get("title") or value.get("snippet") |
| ): |
| return True |
| return any(_item_contains_source_evidence(child) for child in value.values()) |
| if isinstance(value, list): |
| return any(_item_contains_source_evidence(child) for child in value) |
| return False |
|
|
|
|
| def _variation_group_summary(samples: list[dict[str, object]]) -> dict[str, dict[str, object]]: |
| grouped: dict[str, list[str]] = {} |
| for sample in samples: |
| key = str(sample.get("variation_key", "")).strip() |
| if not key: |
| continue |
| grouped.setdefault(key, []).append( |
| _normalize_text(str(sample.get("generated_text", ""))) |
| ) |
| summaries: dict[str, dict[str, object]] = {} |
| for key, responses in grouped.items(): |
| sample_count = len(responses) |
| unique_count = len(set(responses)) |
| summaries[key] = { |
| "sample_count": sample_count, |
| "unique_response_count": unique_count, |
| "unique_response_rate": unique_count / sample_count if sample_count else 0.0, |
| "duplicate_response_rate": ( |
| (sample_count - unique_count) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| } |
| return summaries |
|
|
|
|
| def _content_to_text(content: object) -> str: |
| if isinstance(content, str): |
| return content.strip() |
| if isinstance(content, list): |
| parts: list[str] = [] |
| for item in content: |
| if isinstance(item, dict): |
| if "text" in item: |
| parts.append(str(item["text"])) |
| elif item.get("type") == "text" and "content" in item: |
| parts.append(str(item["content"])) |
| elif item is not None: |
| parts.append(str(item)) |
| return " ".join(part.strip() for part in parts if part and part.strip()).strip() |
| if content is None: |
| return "" |
| return str(content).strip() |
|
|
|
|
| def _render_tool_call(call: object) -> str: |
| if not isinstance(call, dict): |
| return f"<tool_call> {str(call).strip()}" |
| function_payload = call.get("function", {}) |
| function = function_payload if isinstance(function_payload, dict) else {} |
| name = str(call.get("name", function.get("name", "tool"))).strip() or "tool" |
| arguments = call.get("arguments", function.get("arguments", {})) |
| if not isinstance(arguments, str): |
| arguments = json.dumps(arguments, ensure_ascii=False, separators=(",", ":")) |
| return f"<tool_call> {name} {arguments}".strip() |
|
|
|
|
| def _render_tool_result(tool_name: str, result: object) -> list[str]: |
| if isinstance(result, dict): |
| status = str(result.get("status", "ok")).strip() or "ok" |
| if status != "ok": |
| error = str(result.get("error", status)).strip() or status |
| return [f"<tool_result> {tool_name} failed: {error}"] |
| lines = [f"<tool_result> {tool_name} ok"] |
| sources = result.get("sources", []) |
| if isinstance(sources, list): |
| for source in sources: |
| if not isinstance(source, dict): |
| continue |
| title = str(source.get("title", "Source")).strip() or "Source" |
| url = str(source.get("url", "")).strip() |
| snippet = str(source.get("snippet", source.get("text", ""))).strip() |
| lines.append(f"<source> {title} | {url} | {snippet}".strip()) |
| return lines |
| content = _content_to_text(result) |
| return [f"<tool_result> {tool_name} {content or 'empty'}"] |
|
|
|
|
| def _compose_prompt_context(item: dict[str, object]) -> str: |
| prompt = str(item.get("prompt", "")).strip() |
| system = str(item.get("system", "")).strip() |
| lines: list[str] = [] |
| tool_protocol_seen = False |
| if system: |
| lines.append(system) |
|
|
| messages = item.get("messages") |
| if isinstance(messages, list): |
| for message in messages: |
| if not isinstance(message, dict): |
| continue |
| role = str(message.get("role", "")).casefold() |
| content = _content_to_text(message.get("content", "")) |
| if role == "system": |
| if content: |
| lines.append(f"System instruction: {content}") |
| elif role == "user": |
| if content: |
| lines.append(f"User: {content}") |
| elif role == "assistant": |
| if content: |
| lines.append(f"Assistant: {content}") |
| if "<tool_call>" in content: |
| tool_protocol_seen = True |
| tool_calls = message.get("tool_calls", []) |
| if isinstance(tool_calls, list): |
| for call in tool_calls: |
| lines.append(_render_tool_call(call)) |
| tool_protocol_seen = True |
| elif role == "tool": |
| tool_name = str(message.get("name", message.get("tool_call_id", "tool"))) |
| lines.extend(_render_tool_result(tool_name, message.get("content", ""))) |
| tool_protocol_seen = True |
| elif content: |
| lines.append(f"{role.capitalize()}: {content}") |
|
|
| if prompt: |
| lines.append(f"User: {prompt}" if isinstance(messages, list) else prompt) |
|
|
| tool_results = item.get("tool_results") |
| if isinstance(tool_results, list): |
| for result in tool_results: |
| tool_name = "tool" |
| if isinstance(result, dict): |
| tool_name = str(result.get("name", result.get("tool", "tool"))) |
| lines.extend(_render_tool_result(tool_name, result)) |
| tool_protocol_seen = True |
| elif tool_results: |
| lines.extend(_render_tool_result("tool", tool_results)) |
| tool_protocol_seen = True |
|
|
| if tool_protocol_seen: |
| lines.append("<final>") |
| return "\n".join(line for line in lines if line).strip() |
|
|
|
|
| def _open_ended_score( |
| model: ReframrModel, |
| sample: dict[str, object], |
| *, |
| reasoning_mode: str | None, |
| ) -> dict[str, object]: |
| generated = model.generate_text( |
| str(sample["context"]), |
| max_tokens=int(sample.get("max_tokens", 56)), |
| reasoning_mode=reasoning_mode, |
| ) |
| normalized = _normalize_text(generated) |
| required_groups = [ |
| [str(term).casefold() for term in group] |
| for group in sample.get("required_groups", []) |
| ] |
| satisfied_groups = sum( |
| 1 |
| for group in required_groups |
| if any(term in normalized for term in group) |
| ) |
| group_coverage = ( |
| satisfied_groups / len(required_groups) if required_groups else 0.0 |
| ) |
| punctuation_hit = any(mark in generated for mark in ".,;:?!") |
| min_words = int(sample.get("min_words", 12)) |
| min_word_hit = len(generated.split()) >= min_words |
| banned_phrases = [str(phrase) for phrase in sample.get("banned_phrases", [])] |
| exact_copy = any(normalized == _normalize_text(phrase) for phrase in banned_phrases) |
| novelty_hit = not exact_copy |
| require_punctuation = bool(sample.get("require_punctuation", True)) |
|
|
| score_components = [ |
| group_coverage, |
| 1.0 if min_word_hit else 0.0, |
| 1.0 if novelty_hit else 0.0, |
| ] |
| if require_punctuation: |
| score_components.append(1.0 if punctuation_hit else 0.0) |
|
|
| return { |
| "section": str(sample["section"]), |
| "context": str(sample["context"]), |
| "generated_text": generated, |
| "group_coverage": group_coverage, |
| "punctuation_hit": punctuation_hit, |
| "min_word_hit": min_word_hit, |
| "exact_copy": exact_copy, |
| "score": sum(score_components) / len(score_components) if score_components else 0.0, |
| } |
|
|
|
|
| def evaluate_manifest( |
| model: ReframrModel, |
| manifest: dict[str, object], |
| *, |
| reasoning_mode: str | None = None, |
| top_k: int = 5, |
| ) -> dict[str, object]: |
| results: dict[str, object] = { |
| "corpus_name": manifest["name"], |
| "reasoning_mode": reasoning_mode or model.config.default_reasoning_profile, |
| "splits": {}, |
| } |
|
|
| splits = manifest["splits"] |
| for split_name in ("memorization", "generalization"): |
| samples = splits[split_name] |
| top1_hits = 0 |
| topk_hits = 0 |
| expected_probabilities = [] |
|
|
| for sample in samples: |
| distribution = model.predict_next_token_distribution( |
| sample["context"], |
| reasoning_mode=reasoning_mode, |
| ) |
| ranked = sorted(distribution.items(), key=lambda item: item[1], reverse=True) |
| predicted = ranked[0][0] if ranked else "" |
| top_tokens = [token for token, _ in ranked[:top_k]] |
| expected = _expected_next_token(model, sample["expected"]) |
| expected_probability = distribution.get(expected, 0.0) |
|
|
| if predicted == expected: |
| top1_hits += 1 |
| if expected in top_tokens: |
| topk_hits += 1 |
| expected_probabilities.append(expected_probability) |
|
|
| sample_count = len(samples) |
| mean_expected_probability = ( |
| sum(expected_probabilities) / sample_count if sample_count else 0.0 |
| ) |
| results["splits"][split_name] = { |
| "sample_count": sample_count, |
| "top1_accuracy": top1_hits / sample_count if sample_count else 0.0, |
| "topk_accuracy": topk_hits / sample_count if sample_count else 0.0, |
| "mean_expected_probability": mean_expected_probability, |
| } |
|
|
| open_ended_samples = splits.get("open_ended", []) |
| if open_ended_samples: |
| sample_results = [ |
| _open_ended_score( |
| model, |
| sample, |
| reasoning_mode=reasoning_mode, |
| ) |
| for sample in open_ended_samples |
| ] |
| sample_count = len(sample_results) |
| results["open_ended"] = { |
| "sample_count": sample_count, |
| "mean_score": ( |
| sum(float(sample["score"]) for sample in sample_results) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "mean_group_coverage": ( |
| sum(float(sample["group_coverage"]) for sample in sample_results) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "punctuation_rate": ( |
| sum(1 for sample in sample_results if bool(sample["punctuation_hit"])) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "min_word_rate": ( |
| sum(1 for sample in sample_results if bool(sample["min_word_hit"])) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "exact_copy_rate": ( |
| sum(1 for sample in sample_results if bool(sample["exact_copy"])) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "samples": sample_results, |
| } |
|
|
| return results |
|
|
|
|
| def benchmark_open_prompts( |
| model: ReframrModel, |
| prompts: list[dict[str, object]], |
| *, |
| reasoning_mode: str | None = None, |
| max_tokens: int = 64, |
| temperature: float = 0.82, |
| top_k: int = 24, |
| top_p: float = 0.92, |
| repetition_penalty: float = 1.18, |
| replay_sources: Sequence[str] | None = None, |
| replay_ngram_size: int = 8, |
| replay_overlap_threshold: float = 0.70, |
| ) -> dict[str, object]: |
| samples: list[dict[str, object]] = [] |
| normalized_replay_ngram_size = max(3, int(replay_ngram_size)) |
| replay_index = _source_replay_index( |
| replay_sources, |
| ngram_size=normalized_replay_ngram_size, |
| ) |
| avoid_texts = list(replay_sources or []) |
| for item in prompts: |
| prompt = str(item["prompt"]) |
| context = _compose_prompt_context(item) |
| generated = model.generate_text( |
| context, |
| max_tokens=max_tokens, |
| reasoning_mode=reasoning_mode, |
| temperature=temperature, |
| top_k=top_k, |
| top_p=top_p, |
| repetition_penalty=repetition_penalty, |
| avoid_texts=avoid_texts, |
| ) |
| normalized = _normalize_text(generated) |
| banned_phrases = [str(phrase) for phrase in item.get("banned_phrases", [])] |
| exact_copy = any( |
| normalized == _normalize_text(phrase) |
| for phrase in banned_phrases |
| ) |
| words = generated.split() |
| punctuation_hit = any(mark in generated for mark in ".,;:?!") |
| tool_call_hit = "<tool_call>" in generated |
| generated_tool_result_hit = "<tool_result>" in generated |
| generated_source_hit = "<source>" in generated |
| fabricated_tool_result_hit = generated_tool_result_hit and "<tool_result>" not in context |
| fabricated_source_hit = generated_source_hit and "<source>" not in context |
| required_group_hits, required_group_count, required_group_coverage = ( |
| _required_group_summary(normalized, item.get("required_groups", [])) |
| ) |
| source_replay_overlap, source_replay_source = _source_replay_overlap( |
| generated, |
| replay_index, |
| ngram_size=normalized_replay_ngram_size, |
| ) |
| source_replay_hit = ( |
| bool(replay_index) |
| and source_replay_overlap >= float(replay_overlap_threshold) |
| ) |
| banned_hit = _banned_phrase_hit(normalized, item.get("banned_phrases", [])) |
| meta_hit = _meta_voice_hit(normalized) |
| malformed_start = _has_malformed_sentence_start(generated) |
| distinct_2 = _distinct_ratio(words, 2) |
| distinct_3 = _distinct_ratio(words, 3) |
| repetition_3 = _repetition_ratio(words, 3) |
| passed_quality_gate = _quality_gate_passed( |
| word_count=len(words), |
| punctuation_hit=punctuation_hit, |
| required_group_coverage=required_group_coverage, |
| exact_copy=exact_copy, |
| banned_phrase_hit=banned_hit, |
| meta_voice_hit=meta_hit, |
| malformed_start=malformed_start, |
| repetition_3=repetition_3, |
| tool_call_hit=tool_call_hit, |
| fabricated_tool_result_hit=fabricated_tool_result_hit, |
| fabricated_source_hit=fabricated_source_hit, |
| source_replay_hit=source_replay_hit, |
| item=item, |
| ) |
| samples.append( |
| { |
| "prompt": prompt, |
| "context": context, |
| "tags": [str(tag) for tag in item.get("tags", [])], |
| "variation_key": str(item.get("variation_key", "")).strip(), |
| "generated_text": generated, |
| "word_count": len(words), |
| "char_count": len(generated), |
| "punctuation_hit": punctuation_hit, |
| "distinct_2": distinct_2, |
| "distinct_3": distinct_3, |
| "repetition_3": repetition_3, |
| "exact_copy": exact_copy, |
| "banned_phrase_hit": banned_hit, |
| "tool_call_hit": tool_call_hit, |
| "generated_tool_result_hit": generated_tool_result_hit, |
| "generated_source_hit": generated_source_hit, |
| "fabricated_tool_result_hit": fabricated_tool_result_hit, |
| "fabricated_source_hit": fabricated_source_hit, |
| "source_replay_overlap": source_replay_overlap, |
| "source_replay_hit": source_replay_hit, |
| "source_replay_source": source_replay_source, |
| "required_group_hits": required_group_hits, |
| "required_group_count": required_group_count, |
| "required_group_coverage": required_group_coverage, |
| "malformed_start": malformed_start, |
| "meta_voice_hit": meta_hit, |
| "passed_quality_gate": passed_quality_gate, |
| } |
| ) |
|
|
| sample_count = len(samples) |
| normalized_responses = [ |
| _normalize_text(str(sample["generated_text"])) |
| for sample in samples |
| ] |
| unique_response_count = len(set(normalized_responses)) |
| exact_copy_count = sum(1 for sample in samples if bool(sample["exact_copy"])) |
| banned_phrase_count = sum( |
| 1 for sample in samples if bool(sample["banned_phrase_hit"]) |
| ) |
| malformed_start_count = sum( |
| 1 for sample in samples if bool(sample["malformed_start"]) |
| ) |
| meta_voice_count = sum(1 for sample in samples if bool(sample["meta_voice_hit"])) |
| tool_call_count = sum(1 for sample in samples if bool(sample["tool_call_hit"])) |
| fabricated_tool_result_count = sum( |
| 1 for sample in samples if bool(sample["fabricated_tool_result_hit"]) |
| ) |
| fabricated_source_count = sum( |
| 1 for sample in samples if bool(sample["fabricated_source_hit"]) |
| ) |
| source_replay_count = sum( |
| 1 for sample in samples if bool(sample["source_replay_hit"]) |
| ) |
| quality_pass_count = sum( |
| 1 for sample in samples if bool(sample["passed_quality_gate"]) |
| ) |
| variation_groups = _variation_group_summary(samples) |
| worst_variation_group_unique_rate = ( |
| min( |
| float(summary["unique_response_rate"]) |
| for summary in variation_groups.values() |
| ) |
| if variation_groups |
| else 1.0 |
| ) |
| required_group_samples = [ |
| sample |
| for sample in samples |
| if int(sample.get("required_group_count", 0)) > 0 |
| ] |
| required_group_sample_count = len(required_group_samples) |
| mean_required_group_coverage = ( |
| sum(float(sample["required_group_coverage"]) for sample in required_group_samples) |
| / required_group_sample_count |
| if required_group_sample_count |
| else 0.0 |
| ) |
| quality_scores = [ |
| quality_pass_count / sample_count if sample_count else 0.0, |
| unique_response_count / sample_count if sample_count else 0.0, |
| mean_required_group_coverage, |
| 1.0 - (exact_copy_count / sample_count if sample_count else 0.0), |
| 1.0 - (banned_phrase_count / sample_count if sample_count else 0.0), |
| 1.0 - (fabricated_tool_result_count / sample_count if sample_count else 0.0), |
| 1.0 - (fabricated_source_count / sample_count if sample_count else 0.0), |
| 1.0 - (source_replay_count / sample_count if sample_count else 0.0), |
| 1.0 - (malformed_start_count / sample_count if sample_count else 0.0), |
| 1.0 - (meta_voice_count / sample_count if sample_count else 0.0), |
| worst_variation_group_unique_rate, |
| ] |
| return { |
| "schema_version": "reframr.open_benchmark.v2", |
| "sample_count": sample_count, |
| "reasoning_mode": reasoning_mode or model.config.default_reasoning_profile, |
| "generation_policy": { |
| "temperature": temperature, |
| "top_k": top_k, |
| "top_p": top_p, |
| "repetition_penalty": repetition_penalty, |
| }, |
| "mean_word_count": ( |
| sum(int(sample["word_count"]) for sample in samples) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "mean_char_count": ( |
| sum(int(sample["char_count"]) for sample in samples) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "punctuation_rate": ( |
| sum(1 for sample in samples if bool(sample["punctuation_hit"])) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "required_group_sample_count": required_group_sample_count, |
| "mean_required_group_coverage": mean_required_group_coverage, |
| "mean_distinct_2": ( |
| sum(float(sample["distinct_2"]) for sample in samples) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "mean_distinct_3": ( |
| sum(float(sample["distinct_3"]) for sample in samples) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "mean_repetition_3": ( |
| sum(float(sample["repetition_3"]) for sample in samples) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "exact_copy_count": exact_copy_count, |
| "exact_copy_rate": exact_copy_count / sample_count if sample_count else 0.0, |
| "banned_phrase_count": banned_phrase_count, |
| "banned_phrase_rate": ( |
| banned_phrase_count / sample_count if sample_count else 0.0 |
| ), |
| "malformed_start_count": malformed_start_count, |
| "malformed_start_rate": ( |
| malformed_start_count / sample_count if sample_count else 0.0 |
| ), |
| "meta_voice_count": meta_voice_count, |
| "meta_voice_rate": meta_voice_count / sample_count if sample_count else 0.0, |
| "tool_call_count": tool_call_count, |
| "tool_call_rate": tool_call_count / sample_count if sample_count else 0.0, |
| "fabricated_tool_result_count": fabricated_tool_result_count, |
| "fabricated_tool_result_rate": ( |
| fabricated_tool_result_count / sample_count if sample_count else 0.0 |
| ), |
| "fabricated_source_count": fabricated_source_count, |
| "fabricated_source_rate": ( |
| fabricated_source_count / sample_count if sample_count else 0.0 |
| ), |
| "source_replay_count": source_replay_count, |
| "source_replay_rate": ( |
| source_replay_count / sample_count if sample_count else 0.0 |
| ), |
| "replay_ngram_size": normalized_replay_ngram_size, |
| "replay_overlap_threshold": float(replay_overlap_threshold), |
| "quality_pass_count": quality_pass_count, |
| "quality_pass_rate": quality_pass_count / sample_count if sample_count else 0.0, |
| "unique_response_count": unique_response_count, |
| "unique_response_rate": unique_response_count / sample_count if sample_count else 0.0, |
| "duplicate_response_rate": ( |
| (sample_count - unique_response_count) / sample_count |
| if sample_count |
| else 0.0 |
| ), |
| "variation_groups": variation_groups, |
| "worst_variation_group_unique_rate": worst_variation_group_unique_rate, |
| "v2_readiness_score": sum(quality_scores) / len(quality_scores), |
| "samples": samples, |
| } |
|
|
