From 263120925628cd9036182b9b91d37f1a6dc1aceb Mon Sep 17 00:00:00 2001 From: 3un01a <3un01a.labs@gmail.com> Date: Tue, 24 Feb 2026 11:30:30 +0900 Subject: [PATCH] feat: Update Honcho system benchmarks (#393) * feat: Update Honcho system benchmarks * fix: Align with runner common functions and fixed basedpyright issues * fix: Address coderabbit issues * chore: Address Ruff Errors --------- Co-authored-by: 3un01a <3un01a@plasticlabs.ai> Co-authored-by: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> --- tests/bench/coverage.py | 1300 ++++++++++++++++++++++++++++++ tests/bench/explicit.py | 1443 ---------------------------------- tests/bench/molecular.py | 1222 ++++++++++++++++++++++++++++ tests/bench/obex.py | 643 --------------- tests/bench/runner_common.py | 150 +++- 5 files changed, 2662 insertions(+), 2096 deletions(-) create mode 100644 tests/bench/coverage.py delete mode 100644 tests/bench/explicit.py create mode 100644 tests/bench/molecular.py delete mode 100644 tests/bench/obex.py diff --git a/tests/bench/coverage.py b/tests/bench/coverage.py new file mode 100644 index 00000000..26734ccd --- /dev/null +++ b/tests/bench/coverage.py @@ -0,0 +1,1300 @@ +""" +CoverageBench - Information Recall Benchmark for Fact Extraction +Author: Based on FActScore, SAFE, and QuestEval methodologies + +## Core Principle + +Measures how much extractable information from the source was actually captured. +Complements MolecularBench (quality) with recall metrics (quantity/coverage). + +## Evaluation Pipeline + +### Stage 1: Gold Fact Extraction +Extract ALL facts that could reasonably be derived from source messages. +Categories: +- Explicit facts (directly stated) +- Implicit facts (clearly implied by context) +- Relational facts (relationships between entities) + +### Stage 2: Coverage Matching +For each gold fact, determine if it's covered by the extraction: +- COVERED: Semantically equivalent fact exists in extraction +- PARTIAL: Core information present but incomplete +- MISSING: Not present in extraction + +### Stage 3: QA Verification (Optional) +Generate questions from source; verify answerability from extraction. + +## Metrics + +- Recall = covered_facts / gold_facts +- Partial Recall = (covered + 0.5 * partial) / gold_facts +- QA Coverage = answerable_questions / total_questions +- Density = extracted_facts / source_tokens +- F1 = harmonic_mean(molecular_quality, coverage_recall) + +## Usage + +python -m tests.bench.coverage --traces path/to/traces.jsonl +""" + +import argparse +import asyncio +import json +import logging +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Any, cast + +import anthropic +import openai +from anthropic import AsyncAnthropic +from dotenv import load_dotenv +from openai import AsyncOpenAI +from tqdm.asyncio import tqdm + +from .runner_common import ( + configure_logging, + create_anthropic_client, + create_openai_client, + extract_messages, + extract_propositions, + format_duration, + load_traces, +) + +# Load .env from bench directory +bench_dir = Path(__file__).parent +load_dotenv(bench_dir / ".env") + +logger = configure_logging(level=logging.INFO, name=__name__) + + +# ============================================================================= +# ENUMS AND TYPES +# ============================================================================= + + +class FactCategory(Enum): + """Categories of extractable facts.""" + EXPLICIT = "explicit" # Directly stated: "I work at Google" + IMPLICIT = "implicit" # Clearly implied: "my commute to Mountain View" → lives near MV + RELATIONAL = "relational" # Relationships: "my sister's husband" → has sister, sister is married + TEMPORAL = "temporal" # Time-bound facts: "started last year" + PREFERENCE = "preference" # Likes/dislikes: "I love hiking" + BIOGRAPHICAL = "biographical" # Personal details: name, age, location + BEHAVIORAL = "behavioral" # Habits/patterns: "I usually wake up early" + + +class CoverageStatus(Enum): + """How well a gold fact is covered by extraction.""" + COVERED = "covered" # Fully present (possibly rephrased) + PARTIAL = "partial" # Core info present but incomplete + MISSING = "missing" # Not present at all + OVERCLAIMED = "overclaimed" # Extraction claims more than source supports + + +class ImportanceLevel(Enum): + """Importance weighting for facts (Pyramid-inspired).""" + CRITICAL = "critical" # Core identifying information + IMPORTANT = "important" # Significant details + MINOR = "minor" # Nice-to-have details + TRIVIAL = "trivial" # Marginal information + + +# ============================================================================= +# DATA STRUCTURES +# ============================================================================= + + +@dataclass +class GoldFact: + """A fact that should be extractable from the source.""" + content: str + category: FactCategory + importance: ImportanceLevel + source_span: str = "" # The text that supports this fact + requires_inference: bool = False # Whether extraction requires reasoning + + +@dataclass +class CoverageMatch: + """Result of matching a gold fact against extraction.""" + gold_fact: GoldFact + status: CoverageStatus + matched_extraction: str = "" # Which extracted fact covers this (if any) + match_quality: float = 0.0 # 0-1, how well it matches + explanation: str = "" + + +@dataclass +class ExtractionAnalysis: + """Analysis of an extracted fact.""" + content: str + is_grounded: bool = True # Supported by source + is_hallucinated: bool = False # Claims something not in source + matched_gold: list[str] = field(default_factory=list) # Which gold facts it covers + + +@dataclass +class CoverageReport: + """Complete coverage analysis for a trace.""" + conversation_id: str + source_message_count: int + source_token_count: int + + # Gold facts + gold_facts: list[GoldFact] = field(default_factory=list) + gold_fact_count: int = 0 + + # Extracted facts + extracted_facts: list[str] = field(default_factory=list) + extracted_count: int = 0 + + # Coverage matching + matches: list[CoverageMatch] = field(default_factory=list) + + # Core metrics + recall: float = 0.0 # covered / gold + partial_recall: float = 0.0 # (covered + 0.5*partial) / gold + weighted_recall: float = 0.0 # importance-weighted recall + precision: float = 0.0 # grounded / extracted + f1: float = 0.0 # harmonic mean + + # Detailed metrics + coverage_by_category: dict[str, float] = field(default_factory=dict) + coverage_by_importance: dict[str, float] = field(default_factory=dict) + + # Density metrics + extraction_density: float = 0.0 # extracted / source_tokens + gold_density: float = 0.0 # gold / source_tokens + density_ratio: float = 0.0 # extraction_density / gold_density + + # QA verification (optional) + qa_questions: list[str] = field(default_factory=list) + qa_answerable: int = 0 + qa_coverage: float = 0.0 + + # Issues + missing_critical: list[str] = field(default_factory=list) + hallucinations: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "conversation_id": self.conversation_id, + "source_messages": self.source_message_count, + "source_tokens": self.source_token_count, + "counts": { + "gold_facts": self.gold_fact_count, + "extracted_facts": self.extracted_count, + "covered": sum(1 for m in self.matches if m.status == CoverageStatus.COVERED), + "partial": sum(1 for m in self.matches if m.status == CoverageStatus.PARTIAL), + "missing": sum(1 for m in self.matches if m.status == CoverageStatus.MISSING), + }, + "scores": { + "recall": round(self.recall, 4), + "partial_recall": round(self.partial_recall, 4), + "weighted_recall": round(self.weighted_recall, 4), + "precision": round(self.precision, 4), + "f1": round(self.f1, 4), + "qa_coverage": round(self.qa_coverage, 4) if self.qa_questions else None, + }, + "density": { + "extraction_density": round(self.extraction_density, 4), + "gold_density": round(self.gold_density, 4), + "density_ratio": round(self.density_ratio, 4), + }, + "coverage_by_category": {k: round(v, 3) for k, v in self.coverage_by_category.items()}, + "coverage_by_importance": {k: round(v, 3) for k, v in self.coverage_by_importance.items()}, + "issues": { + "missing_critical": self.missing_critical[:5], # Top 5 + "hallucinations": self.hallucinations[:5], + }, + "gold_facts": [ + { + "content": gf.content, + "category": gf.category.value, + "importance": gf.importance.value, + } + for gf in self.gold_facts + ], + "matches": [ + { + "gold": m.gold_fact.content, + "status": m.status.value, + "matched_to": m.matched_extraction, + "quality": round(m.match_quality, 2), + } + for m in self.matches + ], + } + + +# ============================================================================= +# PROMPTS +# ============================================================================= + + +GOLD_EXTRACTION_PROMPT = """You are extracting ALL facts that could reasonably be derived from conversation messages. + +## Your Task + +Given messages from a conversation, extract every piece of factual information about the speaker. +Be EXHAUSTIVE - capture everything that a careful reader could learn about the speaker. + +## Fact Categories + +**EXPLICIT**: Directly stated facts +- "I work at Google" → "Speaker works at Google" + +**IMPLICIT**: Clearly implied by context (not speculation) +- "My commute to the Googleplex is 30 minutes" → "Speaker lives within 30 minutes of Googleplex" + +**RELATIONAL**: Facts about relationships +- "My sister's wedding was beautiful" → "Speaker has a sister", "Speaker's sister is married" + +**TEMPORAL**: Time-bound information +- "I started this job last year" → "Speaker started current job within the past year" + +**PREFERENCE**: Likes, dislikes, opinions +- "I love hiking on weekends" → "Speaker enjoys hiking" + +**BIOGRAPHICAL**: Personal details +- "I'm 32 years old" → "Speaker is 32 years old" + +**BEHAVIORAL**: Habits and patterns +- "I usually wake up at 6am" → "Speaker typically wakes up at 6am" + +## Importance Levels + +**CRITICAL**: Core identifying info (name, occupation, location, key relationships) +**IMPORTANT**: Significant details that build a clear picture +**MINOR**: Nice-to-have details +**TRIVIAL**: Marginal information that adds little value + +## Rules + +1. Extract ATOMIC facts (one claim per fact) +2. Resolve all pronouns and references to be standalone +3. Include the source span that supports each fact +4. Mark whether inference was required +5. Do NOT speculate beyond what the text clearly implies +6. Do NOT include facts about the assistant/system, only about the human speaker""" + + +COVERAGE_MATCHING_PROMPT = """You are evaluating how well extracted facts cover the gold standard facts. + +## Your Task + +For each GOLD FACT (what should have been extracted), determine if the EXTRACTED FACTS cover it. + +## Coverage Statuses + +**COVERED**: The gold fact is fully represented in the extraction +- Gold: "User works at Google" +- Extracted: "User is employed at Google as an engineer" +- Status: COVERED (employment at Google is captured, even with added detail) + +**PARTIAL**: Core information present but incomplete +- Gold: "User has a sister who lives in Boston" +- Extracted: "User has a sister" +- Status: PARTIAL (sister exists, but location missing) + +**MISSING**: The gold fact is not present at all +- Gold: "User enjoys hiking" +- Extracted: [no mention of hiking or outdoor activities] +- Status: MISSING + +## Matching Rules + +1. Semantic equivalence counts as COVERED (rephrasing is fine) +2. If extraction is MORE specific, still COVERED +3. If extraction is LESS specific, mark as PARTIAL +4. Consider entailment: If extracted fact entails gold fact, it's COVERED +5. Don't penalize for organization/grouping differences""" + + +QA_GENERATION_PROMPT = """Generate questions that should be answerable if information was fully extracted. + +## Your Task + +Given the source messages, generate questions about the speaker that a complete extraction would answer. + +## Question Types + +1. **Identity Questions**: Who/What is the speaker? + - "What is the speaker's occupation?" + - "Where does the speaker live?" + +2. **Relationship Questions**: Who are the key people in their life? + - "Does the speaker have siblings?" + - "Is the speaker married?" + +3. **Preference Questions**: What do they like/dislike? + - "What are the speaker's hobbies?" + - "What is the speaker's favorite food?" + +4. **Temporal Questions**: When did events occur? + - "When did the speaker start their current job?" + - "How long has the speaker lived in their current city?" + +5. **Behavioral Questions**: What are their habits? + - "What time does the speaker typically wake up?" + - "How does the speaker commute to work?" + +## Rules + +1. Only ask questions answerable from the source +2. Vary difficulty (some obvious, some requiring inference) +3. Cover different fact categories +4. Make questions specific enough to have clear answers""" + + +QA_VERIFICATION_PROMPT = """Verify which questions can be answered from the extracted facts alone. + +## Your Task + +For each question, determine if the EXTRACTED FACTS (not the source!) provide enough information to answer it. + +## Answerability Levels + +**ANSWERABLE**: The extracted facts directly answer or clearly imply the answer +**PARTIAL**: Some relevant information exists but answer is incomplete +**UNANSWERABLE**: The extracted facts don't contain relevant information + +## Rules + +1. Only consider the EXTRACTED FACTS, not your world knowledge +2. The answer must be derivable from the extraction alone +3. If the question requires combining multiple extracted facts, that's fine +4. Partial credit for incomplete answers""" + + +# ============================================================================= +# COVERAGE JUDGE +# ============================================================================= + + +class CoverageJudge: + """ + Evaluates information recall/coverage in fact extraction. + + Based on: + - FActScore: Atomic fact decomposition + - SAFE: F1 scoring with precision and recall + - QuestEval: QA-based coverage verification + - Pyramid: Importance weighting + """ + + def __init__( + self, + llm_client: AsyncAnthropic | AsyncOpenAI, + model: str = "claude-sonnet-4-20250514", + provider: str = "anthropic", + use_qa_verification: bool = True, + verbose: bool = False, + ): + self.llm_client: AsyncAnthropic | AsyncOpenAI = llm_client + self.model: str = model + self.provider: str = provider + self.use_qa_verification: bool = use_qa_verification + self.verbose: bool = verbose + + if verbose: + logger.setLevel(logging.DEBUG) + + async def _call_llm( + self, + system: str, + user: str, + tool_def: dict[str, Any] + ) -> dict[str, Any]: + """Call LLM with structured tool output.""" + try: + if isinstance(self.llm_client, AsyncAnthropic): + resp = await asyncio.wait_for( + self.llm_client.messages.create( + model=self.model, + max_tokens=8192, + temperature=0.0, + system=system, + messages=[{"role": "user", "content": user}], + tools=cast(Any, [tool_def]), + tool_choice={"type": "tool", "name": tool_def["name"]}, + ), + timeout=300.0, + ) + for block in resp.content: + if block.type == "tool_use": + return dict(cast(Any, block).input) + return {} + else: + # OpenAI-compatible + openai_tool = { + "type": "function", + "function": { + "name": tool_def["name"], + "description": tool_def.get("description", ""), + "parameters": tool_def["input_schema"], + }, + } + resp = await asyncio.wait_for( + self.llm_client.chat.completions.create( + model=self.model, + max_tokens=8192, + temperature=0.0, + messages=[ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + tools=cast(Any, [openai_tool]), + tool_choice={"type": "function", "function": {"name": tool_def["name"]}}, + ), + timeout=300.0, + ) + if resp.choices and resp.choices[0].message.tool_calls: + tc = cast(Any, resp.choices[0].message.tool_calls[0]) + func = tc.function + if func.arguments: + return cast(dict[str, Any], json.loads(func.arguments)) + return {} + except TimeoutError: + logger.exception("LLM call timed out after 300s") + return {} + except (anthropic.APIError, openai.APIError): + logger.exception("LLM API error") + return {} + except json.JSONDecodeError: + logger.exception("LLM returned invalid JSON") + return {} + except Exception: + logger.exception("Unexpected error in LLM call") + raise + + async def extract_gold_facts(self, source_messages: list[dict[str, Any]]) -> list[GoldFact]: + """ + Stage 1: Extract all facts that SHOULD be extractable from source. + + This defines the "gold standard" for recall measurement. + """ + + tool_def = { + "name": "submit_gold_facts", + "description": "Submit all extractable facts from source messages", + "input_schema": { + "type": "object", + "properties": { + "facts": { + "type": "array", + "items": { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": "The fact as a standalone statement" + }, + "category": { + "type": "string", + "enum": [c.value for c in FactCategory] + }, + "importance": { + "type": "string", + "enum": [i.value for i in ImportanceLevel] + }, + "source_span": { + "type": "string", + "description": "The text that supports this fact" + }, + "requires_inference": { + "type": "boolean", + "description": "Whether extracting this requires reasoning beyond literal text" + } + }, + "required": ["content", "category", "importance"] + } + } + }, + "required": ["facts"] + } + } + + # Format messages for LLM + messages_text = "\n".join( + f"[{msg.get('speaker', 'user')}]: {msg.get('text', '')}" + for msg in source_messages + if msg.get('speaker', 'user') == 'user' # Only user messages + ) + + result = await self._call_llm( + GOLD_EXTRACTION_PROMPT, + f"Extract all facts from these messages:\n\n{messages_text}", + tool_def + ) + + gold_facts: list[GoldFact] = [] + for item in cast(list[dict[str, Any]], result.get("facts", [])): + try: + gold_facts.append(GoldFact( + content=cast(str, item.get("content", "")), + category=FactCategory(cast(str, item.get("category", "explicit"))), + importance=ImportanceLevel(cast(str, item.get("importance", "important"))), + source_span=cast(str, item.get("source_span", "")), + requires_inference=cast(bool, item.get("requires_inference", False)), + )) + except (ValueError, KeyError) as e: + logger.debug(f"Skipping malformed gold fact: {e}") + continue + + logger.info(f"Extracted {len(gold_facts)} gold facts from source") + return gold_facts + + async def match_coverage( + self, + gold_facts: list[GoldFact], + extracted_facts: list[str] + ) -> list[CoverageMatch]: + """ + Stage 2: Match gold facts against extraction to measure coverage. + """ + + if not gold_facts: + return [] + + tool_def = { + "name": "submit_coverage_matches", + "description": "Submit coverage matching results", + "input_schema": { + "type": "object", + "properties": { + "matches": { + "type": "array", + "items": { + "type": "object", + "properties": { + "gold_index": {"type": "integer"}, + "status": { + "type": "string", + "enum": [s.value for s in CoverageStatus] + }, + "matched_extraction": { + "type": "string", + "description": "The extracted fact that covers this (if any)" + }, + "match_quality": { + "type": "number", + "minimum": 0, + "maximum": 1, + "description": "How well it matches (1.0 = perfect)" + }, + "explanation": {"type": "string"} + }, + "required": ["gold_index", "status"] + } + } + }, + "required": ["matches"] + } + } + + # Format for LLM + gold_text = "\n".join( + f"{i+1}. [{gf.importance.value.upper()}] {gf.content}" + for i, gf in enumerate(gold_facts) + ) + + extracted_text = "\n".join( + f"- {fact}" for fact in extracted_facts + ) if extracted_facts else "(No facts extracted)" + + result = await self._call_llm( + COVERAGE_MATCHING_PROMPT, + ( + f"GOLD FACTS (what should be extracted):\n{gold_text}\n\n" + + f"EXTRACTED FACTS (what was actually extracted):\n{extracted_text}\n\n" + + "For each gold fact, determine its coverage status." + ), + tool_def + ) + + matches: list[CoverageMatch] = [] + for item in cast(list[dict[str, Any]], result.get("matches", [])): + idx = cast(int, item.get("gold_index", 1)) - 1 + if 0 <= idx < len(gold_facts): + try: + matches.append(CoverageMatch( + gold_fact=gold_facts[idx], + status=CoverageStatus(cast(str, item.get("status", "missing"))), + matched_extraction=cast(str, item.get("matched_extraction", "")), + match_quality=cast(float, item.get("match_quality", 0.0)), + explanation=cast(str, item.get("explanation", "")), + )) + except ValueError: + matches.append(CoverageMatch( + gold_fact=gold_facts[idx], + status=CoverageStatus.MISSING, + )) + + # Ensure all gold facts have a match result + matched_indices = {gold_facts.index(m.gold_fact) for m in matches if m.gold_fact in gold_facts} + for i, gf in enumerate(gold_facts): + if i not in matched_indices: + matches.append(CoverageMatch( + gold_fact=gf, + status=CoverageStatus.MISSING, + explanation="No match result returned" + )) + + return matches + + async def generate_qa_pairs( + self, + source_messages: list[dict[str, Any]] + ) -> list[str]: + """ + Stage 3a: Generate questions that should be answerable from complete extraction. + """ + + tool_def = { + "name": "submit_questions", + "description": "Submit questions for QA-based coverage verification", + "input_schema": { + "type": "object", + "properties": { + "questions": { + "type": "array", + "items": { + "type": "object", + "properties": { + "question": {"type": "string"}, + "expected_answer": {"type": "string"}, + "difficulty": { + "type": "string", + "enum": ["easy", "medium", "hard"] + } + }, + "required": ["question"] + } + } + }, + "required": ["questions"] + } + } + + messages_text = "\n".join( + f"[{msg.get('speaker', 'user')}]: {msg.get('text', '')}" + for msg in source_messages + if msg.get('speaker', 'user') == 'user' + ) + + result = await self._call_llm( + QA_GENERATION_PROMPT, + f"Generate questions from these messages:\n\n{messages_text}", + tool_def + ) + + questions: list[str] = [ + cast(str, item.get("question", "")) + for item in cast(list[dict[str, Any]], result.get("questions", [])) + if item.get("question") + ] + + logger.info(f"Generated {len(questions)} QA questions") + return questions + + async def verify_qa_coverage( + self, + questions: list[str], + extracted_facts: list[str] + ) -> tuple[int, int]: + """ + Stage 3b: Verify how many questions can be answered from extraction alone. + + Returns: (answerable_count, total_count) + """ + + if not questions: + return 0, 0 + + tool_def = { + "name": "submit_qa_results", + "description": "Submit QA verification results", + "input_schema": { + "type": "object", + "properties": { + "results": { + "type": "array", + "items": { + "type": "object", + "properties": { + "question_index": {"type": "integer"}, + "answerable": { + "type": "string", + "enum": ["yes", "partial", "no"] + }, + "answer_from_extraction": {"type": "string"}, + "explanation": {"type": "string"} + }, + "required": ["question_index", "answerable"] + } + } + }, + "required": ["results"] + } + } + + questions_text = "\n".join(f"{i+1}. {q}" for i, q in enumerate(questions)) + extracted_text = "\n".join(f"- {f}" for f in extracted_facts) if extracted_facts else "(No facts)" + + result = await self._call_llm( + QA_VERIFICATION_PROMPT, + ( + f"QUESTIONS:\n{questions_text}\n\n" + + f"EXTRACTED FACTS:\n{extracted_text}\n\n" + + "For each question, determine if it can be answered from the extracted facts alone." + ), + tool_def + ) + + answerable: float = 0 + for item in cast(list[dict[str, Any]], result.get("results", [])): + status = cast(str, item.get("answerable", "no")) + if status == "yes": + answerable += 1 + elif status == "partial": + answerable += 0.5 + + return int(answerable), len(questions) + + async def analyze_extraction_quality( + self, + extracted_facts: list[str], + source_messages: list[dict[str, Any]] + ) -> list[ExtractionAnalysis]: + """ + Analyze each extracted fact for grounding and hallucination. + """ + + tool_def = { + "name": "submit_extraction_analysis", + "description": "Analyze extracted facts for grounding", + "input_schema": { + "type": "object", + "properties": { + "analyses": { + "type": "array", + "items": { + "type": "object", + "properties": { + "index": {"type": "integer"}, + "is_grounded": {"type": "boolean"}, + "is_hallucinated": {"type": "boolean"}, + "explanation": {"type": "string"} + }, + "required": ["index", "is_grounded"] + } + } + }, + "required": ["analyses"] + } + } + + messages_text = "\n".join( + f"[{msg.get('speaker', 'user')}]: {msg.get('text', '')}" + for msg in source_messages + if msg.get('speaker', 'user') == 'user' + ) + + extracted_text = "\n".join(f"{i+1}. {f}" for i, f in enumerate(extracted_facts)) + + result = await self._call_llm( + ( + "Verify each extracted fact is grounded in the source messages. " + + "A fact is GROUNDED if the source text supports it. " + + "A fact is HALLUCINATED if it claims something not supported by the source." + ), + ( + f"SOURCE MESSAGES:\n{messages_text}\n\n" + + f"EXTRACTED FACTS:\n{extracted_text}\n\n" + + "Analyze each extracted fact." + ), + tool_def + ) + + analyses: list[ExtractionAnalysis] = [] + for fact in extracted_facts: + analyses.append(ExtractionAnalysis(content=fact)) + + for item in cast(list[dict[str, Any]], result.get("analyses", [])): + idx = cast(int, item.get("index", 1)) - 1 + if 0 <= idx < len(analyses): + analyses[idx].is_grounded = cast(bool, item.get("is_grounded", True)) + analyses[idx].is_hallucinated = cast(bool, item.get("is_hallucinated", False)) + + return analyses + + def compute_metrics(self, report: CoverageReport) -> None: + """Compute all coverage metrics from matches.""" + + if not report.gold_facts: + return + + n_gold = len(report.gold_facts) + + # Count by status + covered = sum(1 for m in report.matches if m.status == CoverageStatus.COVERED) + partial = sum(1 for m in report.matches if m.status == CoverageStatus.PARTIAL) + + # Basic recall + report.recall = covered / n_gold if n_gold > 0 else 0 + report.partial_recall = (covered + 0.5 * partial) / n_gold if n_gold > 0 else 0 + + # Weighted recall (by importance) + importance_weights = { + ImportanceLevel.CRITICAL: 3.0, + ImportanceLevel.IMPORTANT: 2.0, + ImportanceLevel.MINOR: 1.0, + ImportanceLevel.TRIVIAL: 0.5, + } + + total_weight = sum(importance_weights[gf.importance] for gf in report.gold_facts) + covered_weight = sum( + importance_weights[m.gold_fact.importance] + for m in report.matches + if m.status == CoverageStatus.COVERED + ) + partial_weight = sum( + importance_weights[m.gold_fact.importance] * 0.5 + for m in report.matches + if m.status == CoverageStatus.PARTIAL + ) + + report.weighted_recall = (covered_weight + partial_weight) / total_weight if total_weight > 0 else 0 + + # Precision (grounded extractions / total extractions) + if report.extracted_count > 0: + # This would need extraction analysis, simplified here + report.precision = 1.0 # Assume all grounded unless analyzed + + # F1 + if report.recall + report.precision > 0: + report.f1 = 2 * report.recall * report.precision / (report.recall + report.precision) + + # Coverage by category + for category in FactCategory: + cat_facts = [m for m in report.matches if m.gold_fact.category == category] + if cat_facts: + cat_covered = sum(1 for m in cat_facts if m.status == CoverageStatus.COVERED) + report.coverage_by_category[category.value] = cat_covered / len(cat_facts) + + # Coverage by importance + for importance in ImportanceLevel: + imp_facts = [m for m in report.matches if m.gold_fact.importance == importance] + if imp_facts: + imp_covered = sum(1 for m in imp_facts if m.status == CoverageStatus.COVERED) + report.coverage_by_importance[importance.value] = imp_covered / len(imp_facts) + + # Density metrics + if report.source_token_count > 0: + report.extraction_density = report.extracted_count / report.source_token_count + report.gold_density = n_gold / report.source_token_count + report.density_ratio = report.extraction_density / report.gold_density if report.gold_density > 0 else 0 + + # Track missing critical facts + report.missing_critical = [ + m.gold_fact.content + for m in report.matches + if m.status == CoverageStatus.MISSING and m.gold_fact.importance == ImportanceLevel.CRITICAL + ] + + async def evaluate( + self, + extracted_facts: list[str], + source_messages: list[dict[str, Any]], + conversation_id: str = "", + ) -> CoverageReport: + """Run full coverage evaluation pipeline. + + Args: + extracted_facts: Propositions produced by the deriver to evaluate + for recall against gold-standard facts. + source_messages: Raw conversation messages used to derive gold facts. + Each dict should contain ``speaker`` and ``text`` keys. + conversation_id: Optional identifier for the conversation trace. + Defaults to an empty string. + + Returns: + A ``CoverageReport`` containing recall, precision, F1, density + metrics, per-category/importance breakdowns, and optional QA + coverage results. + """ + + # Estimate token count + source_text = " ".join( + msg.get("text", "") for msg in source_messages + if msg.get("speaker", "user") == "user" + ) + source_tokens = len(source_text.split()) # Rough estimate + + report = CoverageReport( + conversation_id=conversation_id, + source_message_count=len([m for m in source_messages if m.get("speaker") == "user"]), + source_token_count=source_tokens, + extracted_facts=extracted_facts, + extracted_count=len(extracted_facts), + ) + + logger.info(f"Evaluating coverage for {conversation_id}...") + + # Stage 1: Extract gold facts + logger.info("Stage 1: Extracting gold facts...") + report.gold_facts = await self.extract_gold_facts(source_messages) + report.gold_fact_count = len(report.gold_facts) + + if not report.gold_facts: + logger.warning(f"No gold facts extracted for {conversation_id}") + return report + + # Stage 2: Match coverage + logger.info(f"Stage 2: Matching {len(extracted_facts)} extracted against {len(report.gold_facts)} gold facts...") + report.matches = await self.match_coverage(report.gold_facts, extracted_facts) + + # Stage 3: QA verification (optional) + if self.use_qa_verification: + logger.info("Stage 3: QA-based verification...") + report.qa_questions = await self.generate_qa_pairs(source_messages) + if report.qa_questions: + answerable, total = await self.verify_qa_coverage(report.qa_questions, extracted_facts) + report.qa_answerable = answerable + report.qa_coverage = answerable / total if total > 0 else 0 + + # Compute all metrics + self.compute_metrics(report) + + logger.info( + f"Coverage complete: recall={report.recall:.1%}, " + + f"partial_recall={report.partial_recall:.1%}, " + + f"weighted_recall={report.weighted_recall:.1%}" + ) + + return report + + +# ============================================================================= +# COMBINED SCORER (Molecular + Coverage) +# ============================================================================= + + +@dataclass +class CombinedScore: + """Combined molecular quality + coverage recall score.""" + + # Individual scores + molecular: float = 0.0 # From MolecularBench + decontextuality: float = 0.0 + minimality: float = 0.0 + + coverage: float = 0.0 # From CoverageBench + weighted_coverage: float = 0.0 + qa_coverage: float = 0.0 + + # Combined scores + f1: float = 0.0 # Harmonic mean of quality and recall + f2: float = 0.0 # F2 weights recall higher + f05: float = 0.0 # F0.5 weights precision higher + + # For training data filtering + passes_threshold: bool = False + + def compute_combined( + self, + quality_weight: float = 1.0, # noqa: ARG002 + coverage_weight: float = 1.5, # noqa: ARG002 - Weight coverage higher by default + ) -> None: + """Compute F-beta scores.""" + _ = quality_weight, coverage_weight # reserved for future use + + quality = self.molecular # Use molecular as quality proxy + recall = self.weighted_coverage or self.coverage + + if quality + recall == 0: + return + + # F1 (balanced) + self.f1 = 2 * quality * recall / (quality + recall) + + # F2 (recall-weighted) + beta = 2 + self.f2 = (1 + beta**2) * quality * recall / (beta**2 * quality + recall) + + # F0.5 (precision-weighted) + beta = 0.5 + self.f05 = (1 + beta**2) * quality * recall / (beta**2 * quality + recall) + + +def compute_combined_score( + molecular_score: float, + decontextuality: float, + minimality: float, + coverage: float, + weighted_coverage: float | None = None, + qa_coverage: float | None = None, + min_coverage: float = 0.5, + min_molecular: float = 0.8, +) -> CombinedScore: + """ + Combine molecular quality with coverage recall. + + Args: + molecular_score: From MolecularBench + decontextuality: Decontextuality score + minimality: Minimality score + coverage: Basic recall score + weighted_coverage: Importance-weighted recall + qa_coverage: QA-based coverage + min_coverage: Minimum coverage threshold + min_molecular: Minimum molecular quality threshold + """ + + score = CombinedScore( + molecular=molecular_score, + decontextuality=decontextuality, + minimality=minimality, + coverage=coverage, + weighted_coverage=weighted_coverage if weighted_coverage is not None else coverage, + qa_coverage=qa_coverage if qa_coverage is not None else 0.0, + ) + + score.compute_combined() + + # Check if passes both thresholds + score.passes_threshold = ( + score.molecular >= min_molecular and + score.coverage >= min_coverage + ) + + return score + + +# ============================================================================= +# OUTPUT +# ============================================================================= + + +def print_report(report: CoverageReport) -> None: + """Print formatted coverage report.""" + + print("\n" + "=" * 70) + print(f"COVERAGE ANALYSIS: {report.conversation_id}") + print("=" * 70) + + print(f"\nSource: {report.source_message_count} messages, ~{report.source_token_count} tokens") + print(f"Gold Facts: {report.gold_fact_count} | Extracted: {report.extracted_count}") + + # Coverage counts + covered = sum(1 for m in report.matches if m.status == CoverageStatus.COVERED) + partial = sum(1 for m in report.matches if m.status == CoverageStatus.PARTIAL) + missing = sum(1 for m in report.matches if m.status == CoverageStatus.MISSING) + + print(f"\nCoverage: {covered} covered, {partial} partial, {missing} missing") + + # Scores + print(f"\n{'RECALL SCORES'}") + print("-" * 40) + print(f"{'Basic Recall:':<25} {report.recall:.1%}") + print(f"{'Partial Recall:':<25} {report.partial_recall:.1%}") + print(f"{'Weighted Recall:':<25} {report.weighted_recall:.1%}") + if report.qa_coverage > 0: + print(f"{'QA Coverage:':<25} {report.qa_coverage:.1%} ({report.qa_answerable}/{len(report.qa_questions)})") + + # Density + print(f"\n{'DENSITY'}") + print("-" * 40) + print(f"{'Extraction Density:':<25} {report.extraction_density:.4f} facts/token") + print(f"{'Gold Density:':<25} {report.gold_density:.4f} facts/token") + print(f"{'Density Ratio:':<25} {report.density_ratio:.1%}") + + # By category + if report.coverage_by_category: + print(f"\n{'COVERAGE BY CATEGORY'}") + print("-" * 40) + for cat, cov in sorted(report.coverage_by_category.items(), key=lambda x: -x[1]): + print(f" {cat:<20} {cov:.1%}") + + # By importance + if report.coverage_by_importance: + print(f"\n{'COVERAGE BY IMPORTANCE'}") + print("-" * 40) + for imp, cov in sorted(report.coverage_by_importance.items(), key=lambda x: -x[1]): + print(f" {imp:<20} {cov:.1%}") + + # Missing critical + if report.missing_critical: + print("\n⚠️ MISSING CRITICAL FACTS:") + for fact in report.missing_critical[:5]: + print(f" • {fact[:60]}...") + + print("=" * 70) + + +# ============================================================================= +# MAIN +# ============================================================================= + + +async def main(): + parser = argparse.ArgumentParser(description="CoverageBench - Information Recall Evaluation") + parser.add_argument("--traces", type=Path, help="JSON/JSONL trace file") + parser.add_argument("--trace-dir", type=Path, help="Directory of trace files") + parser.add_argument("--output-dir", type=Path, default=Path("tests/bench/coverage_results")) + parser.add_argument("--provider", choices=["anthropic", "openai", "openrouter"], default="anthropic") + parser.add_argument("--api-key", type=str, help="API key") + parser.add_argument("--model", default="claude-sonnet-4-20250514") + parser.add_argument("--no-qa", action="store_true", help="Skip QA verification") + parser.add_argument("--verbose", "-v", action="store_true") + parser.add_argument("--limit", type=int, help="Limit number of traces") + parser.add_argument("--concurrency", type=int, default=5, help="Number of concurrent evaluations") + args = parser.parse_args() + + # Initialize client using runner_common helpers + if args.provider == "anthropic": + client: AsyncAnthropic | AsyncOpenAI = create_anthropic_client( + api_key=args.api_key + ) + elif args.provider == "openrouter": + client = create_openai_client( + api_key=args.api_key, + base_url="https://openrouter.ai/api/v1", + env_key_name="OPENROUTER_API_KEY", + ) + else: + client = create_openai_client(api_key=args.api_key) + + judge = CoverageJudge( + client, + args.model, + args.provider, + use_qa_verification=not args.no_qa, + verbose=args.verbose + ) + + # Load traces + all_traces: list[tuple[dict[str, Any], str]] = [] + if args.traces: + traces = load_traces(args.traces) + all_traces.extend((t, args.traces.name) for t in traces) + elif args.trace_dir: + for f in list(args.trace_dir.glob("*.json")) + list(args.trace_dir.glob("*.jsonl")): + traces = load_traces(f) + all_traces.extend((t, f.name) for t in traces) + else: + parser.error("Specify --traces or --trace-dir") + + if args.limit: + all_traces = all_traces[:args.limit] + + print(f"Evaluating {len(all_traces)} traces for coverage with concurrency={args.concurrency}...\n") + + overall_start = time.time() + + # Semaphore to limit concurrent API calls + semaphore = asyncio.Semaphore(args.concurrency) + + async def evaluate_trace(idx: int, trace: dict[str, Any], source: str) -> tuple[int, CoverageReport | None]: + """Evaluate a single trace with concurrency control.""" + async with semaphore: + props = extract_propositions(trace) + msgs = extract_messages(trace) + conv_id = cast(str, trace.get("conversation_id", f"trace_{idx:04d}")) + + try: + report = await judge.evaluate(props, msgs, conv_id) + return idx, report + except TimeoutError: + logger.exception("Timed out evaluating %s (source=%s)", conv_id, source) + return idx, None + except (anthropic.APIError, openai.APIError): + logger.exception("API error evaluating %s (source=%s)", conv_id, source) + return idx, None + except json.JSONDecodeError: + logger.exception("JSON parse error evaluating %s (source=%s)", conv_id, source) + return idx, None + except Exception: + logger.exception("Unexpected error evaluating %s (source=%s)", conv_id, source) + raise + + # Process all traces concurrently with progress bar + tasks = [evaluate_trace(idx, trace, source) for idx, (trace, source) in enumerate(all_traces)] + results_raw: list[tuple[int, CoverageReport]] = [] + + for coro in cast(Any, tqdm).as_completed(tasks, total=len(tasks), desc="Evaluating traces"): + idx: int + report: CoverageReport | None + idx, report = await coro + if report: + results_raw.append((idx, report)) + + # Sort by original index and extract reports + results_raw.sort(key=lambda x: x[0]) + results: list[CoverageReport] = [report for _, report in results_raw] + + # Print detailed reports + if args.verbose: + print("\n" + "=" * 70) + print("DETAILED REPORTS") + print("=" * 70) + for report in results: + print_report(report) + + total_duration = time.time() - overall_start + + # Save results + if results: + args.output_dir.mkdir(parents=True, exist_ok=True) + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + out_file = args.output_dir / f"coverage_{ts}.json" + + agg: dict[str, Any] = { + "timestamp": ts, + "model": args.model, + "count": len(results), + "duration": format_duration(total_duration), + "averages": { + "recall": sum(r.recall for r in results) / len(results), + "partial_recall": sum(r.partial_recall for r in results) / len(results), + "weighted_recall": sum(r.weighted_recall for r in results) / len(results), + "qa_coverage": sum(r.qa_coverage for r in results) / len(results), + "density_ratio": sum(r.density_ratio for r in results) / len(results), + }, + "totals": { + "gold_facts": sum(r.gold_fact_count for r in results), + "extracted_facts": sum(r.extracted_count for r in results), + }, + "results": [r.to_dict() for r in results], + } + + with open(out_file, "w") as f: + json.dump(agg, f, indent=2) + + print(f"\nSaved to {out_file}") + + # Aggregate summary + print("\n" + "=" * 70) + print("AGGREGATE RESULTS") + print("=" * 70) + print(f"Traces: {len(results)} | Duration: {format_duration(total_duration)}") + print(f"Total Gold Facts: {agg['totals']['gold_facts']}") + print(f"Total Extracted: {agg['totals']['extracted_facts']}") + print(f"\nRecall: {agg['averages']['recall']:.1%}") + print(f"Partial Recall: {agg['averages']['partial_recall']:.1%}") + print(f"Weighted Recall: {agg['averages']['weighted_recall']:.1%}") + print(f"QA Coverage: {agg['averages']['qa_coverage']:.1%}") + print(f"Density Ratio: {agg['averages']['density_ratio']:.1%}") + + return 0 + + +if __name__ == "__main__": + exit(asyncio.run(main())) \ No newline at end of file diff --git a/tests/bench/explicit.py b/tests/bench/explicit.py deleted file mode 100644 index 69509c52..00000000 --- a/tests/bench/explicit.py +++ /dev/null @@ -1,1443 +0,0 @@ -""" -ExplicitBench - Explicit Derivation Benchmark -Author: 3un01a (3un01a@plasticlabs.ai) - -A single-file implementation of the 5-axis evaluation system for explicit derivations. - -## To Run: - -1. Save this file to: tests/bench/explicit.py - -2. Set your API key (or pass via --api-key): - export ANTHROPIC_API_KEY=your_key_here - # or - export OPENAI_API_KEY=your_key_here - # or - export OPENROUTER_API_KEY=your_key_here - -3. Run against a JSON or JSONL file of traces: - python -m tests.bench.explicit --traces path/to/traces.json - python -m tests.bench.explicit --traces path/to/traces.jsonl - -4. Run against a directory of JSON/JSONL files: - python -m tests.bench.explicit --trace-dir path/to/traces/ - -5. Optional flags: - --provider anthropic # Provider: anthropic, openai, or openrouter (default: anthropic) - --api-key your_key_here # API key (overrides environment variable) - --model claude-sonnet-4-20250514 # Model for evaluation (default) - --output-dir tests/bench/eval_results # Where to save results - --verbose # Enable detailed logging - --weights '{"coverage": 0.35, "atomicity": 0.15}' # Custom score weights - --limit 10 # Only evaluate first N traces from the file - --sample 0.1 # Randomly sample 10% of traces - --batch-size 5 # Process N traces concurrently (default: 1) - -## Input Format (Trace JSON/JSONL): - -The script accepts two formats: - -1. JSON array: -[ - {"input": {"prompt": "........."}, "output": {"content": {"explicit": [{"content": "prop1"}, ...]}}}, - {"input": {"prompt": "........."}, "output": {"content": {"explicit": [{"content": "prop1"}, ...]}}}, - ... -] - -2. JSONL (one JSON object per line): -{"input": {"prompt": "........."}, "output": {"content": {"explicit": [{"content": "prop1"}, ...]}}} -{"input": {"prompt": "........."}, "output": {"content": {"explicit": [{"content": "prop1"}, ...]}}} -... - -## Output: - -- Console summary with scores and notable issues -- JSON file with detailed evaluation results -""" - -import argparse -import asyncio -import json -import logging -import os -import random -from dataclasses import dataclass, field -from datetime import datetime, timezone -from enum import Enum -from pathlib import Path -from typing import Any - -from anthropic import AsyncAnthropic -from openai import AsyncOpenAI - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(levelname)s - %(message)s", -) -logger = logging.getLogger(__name__) - - -# ============================================================================= -# DATA STRUCTURES -# ============================================================================= - - -class TemporalType(Enum): - STATIC = "static" - DYNAMIC_STATE = "dynamic" - EVENT = "event" - DURATION = "duration" - HABITUAL = "habitual" - UNKNOWN = "unknown" - - -class AtomicityViolation(Enum): - CONJUNCTION = "conjunction" - DISJUNCTION = "disjunction" - CONDITIONAL = "conditional" - EMBEDDED_QUOTE = "embedded_quote" - COMPOUND_PREDICATE = "compound_predicate" - CAUSAL_CHAIN = "causal_chain" - TEMPORAL_SEQUENCE = "temporal_sequence" - - -class FidelityViolation(Enum): - HEDGE_REMOVAL = "hedge_removal" - NEGATION_FLIP = "negation_flip" - TEMPORAL_SHIFT = "temporal_shift" - QUANTITY_CHANGE = "quantity_change" - ATTRIBUTION_ERROR = "attribution_error" - OVERGENERALIZATION = "overgeneralization" - OVERSPECIFICATION = "overspecification" - INFERENCE_AS_EXPLICIT = "inference_as_explicit" - - -class PremiseSuitability(Enum): - EXCELLENT = "excellent" - GOOD = "good" - MARGINAL = "marginal" - POOR = "poor" - UNUSABLE = "unusable" - - -@dataclass -class AtomicityResult: - proposition: str - is_atomic: bool - violations: list[AtomicityViolation] = field(default_factory=list) - suggested_decomposition: list[str] = field(default_factory=list) - reasoning: str = "" - - -@dataclass -class CoverageGap: - missing_claim: str - source_quote: str - source_message_id: str - severity: str - reasoning: str = "" - - -@dataclass -class RedundancyCluster: - propositions: list[str] - canonical_form: str - redundancy_type: str - - -@dataclass -class FidelityResult: - proposition: str - is_faithful: bool - violations: list[tuple[FidelityViolation, str]] = field(default_factory=list) - source_quote: str = "" - severity: str = "none" - - -@dataclass -class DownstreamUtilityResult: - proposition: str - suitability: PremiseSuitability - issues: list[str] = field(default_factory=list) - has_clear_subject: bool = True - has_clear_predicate: bool = True - is_contextually_complete: bool = True - has_stable_truth_value: bool = True - is_composable: bool = True - temporal_handling: str = "appropriate" - reasoning: str = "" - - -@dataclass -class AtomicityReport: - total_propositions: int - atomic_count: int - score: float - violations_by_type: dict[str, int] = field(default_factory=dict) - detailed_results: list[AtomicityResult] = field(default_factory=list) - decomposition_suggestions: int = 0 - estimated_atomic_propositions: int = 0 - - -@dataclass -class CoverageReport: - total_source_claims: int - extracted_claims: int - coverage_score: float - gaps: list[CoverageGap] = field(default_factory=list) - gaps_by_severity: dict[str, int] = field(default_factory=dict) - source_message_count: int = 0 - propositions_per_message: float = 0.0 - - -@dataclass -class FidelityReport: - total_propositions: int - faithful_count: int - fidelity_score: float - violations_by_type: dict[str, int] = field(default_factory=dict) - violations_by_severity: dict[str, int] = field(default_factory=dict) - detailed_results: list[FidelityResult] = field(default_factory=list) - - -@dataclass -class EfficiencyReport: - total_propositions: int - unique_propositions: int - efficiency_score: float - redundancy_clusters: list[RedundancyCluster] = field(default_factory=list) - exact_duplicates: int = 0 - near_duplicates: int = 0 - subsumptions: int = 0 - - -@dataclass -class DownstreamUtilityReport: - total_propositions: int - suitability_distribution: dict[str, int] = field(default_factory=dict) - clarity_score: float = 0.0 - completeness_score: float = 0.0 - stability_score: float = 0.0 - composability_score: float = 0.0 - temporal_score: float = 0.0 - utility_score: float = 0.0 - detailed_results: list[DownstreamUtilityResult] = field(default_factory=list) - - -@dataclass -class EvaluationResult: - conversation_id: str - peer_name: str - proposition_count: int - source_message_count: int - atomicity: AtomicityReport - coverage: CoverageReport - fidelity: FidelityReport - efficiency: EfficiencyReport - downstream_utility: DownstreamUtilityReport - overall_score: float = 0.0 - - def compute_overall_score(self, weights: dict[str, float] | None = None) -> float: - w = weights or { - "atomicity": 0.15, - "coverage": 0.35, - "fidelity": 0.20, - "efficiency": 0.10, - "utility": 0.20, - } - self.overall_score = ( - self.atomicity.score * w.get("atomicity", 0.15) - + self.coverage.coverage_score * w.get("coverage", 0.35) - + self.fidelity.fidelity_score * w.get("fidelity", 0.20) - + self.efficiency.efficiency_score * w.get("efficiency", 0.10) - + self.downstream_utility.utility_score * w.get("utility", 0.20) - ) - return self.overall_score - - def to_dict(self) -> dict[str, Any]: - return { - "conversation_id": self.conversation_id, - "peer_name": self.peer_name, - "proposition_count": self.proposition_count, - "source_message_count": self.source_message_count, - "scores": { - "overall": round(self.overall_score, 4), - "atomicity": round(self.atomicity.score, 4), - "coverage": round(self.coverage.coverage_score, 4), - "fidelity": round(self.fidelity.fidelity_score, 4), - "efficiency": round(self.efficiency.efficiency_score, 4), - "downstream_utility": round(self.downstream_utility.utility_score, 4), - }, - "atomicity_details": { - "atomic_count": self.atomicity.atomic_count, - "total": self.atomicity.total_propositions, - "violations_by_type": self.atomicity.violations_by_type, - }, - "coverage_details": { - "estimated_total": self.coverage.total_source_claims, - "gaps_count": len(self.coverage.gaps), - "gaps_by_severity": self.coverage.gaps_by_severity, - "propositions_per_message": round( - self.coverage.propositions_per_message, 2 - ), - "gaps": [ - {"claim": g.missing_claim, "severity": g.severity} - for g in self.coverage.gaps - ], - }, - "fidelity_details": { - "faithful_count": self.fidelity.faithful_count, - "violations_by_type": self.fidelity.violations_by_type, - }, - "efficiency_details": { - "unique_propositions": self.efficiency.unique_propositions, - "redundancy_clusters": len(self.efficiency.redundancy_clusters), - }, - "utility_details": { - "suitability_distribution": self.downstream_utility.suitability_distribution, - "component_scores": { - "clarity": round(self.downstream_utility.clarity_score, 4), - "completeness": round( - self.downstream_utility.completeness_score, 4 - ), - "stability": round(self.downstream_utility.stability_score, 4), - "composability": round( - self.downstream_utility.composability_score, 4 - ), - "temporal": round(self.downstream_utility.temporal_score, 4), - }, - }, - } - - -# ============================================================================= -# EVALUATION PROMPTS -# ============================================================================= - -ATOMICITY_CRITERIA = """## Atomicity Evaluation - -A proposition is ATOMIC if it contains exactly ONE claim with ONE truth value. - -### Violations: -- CONJUNCTION: Multiple claims joined by "and" ("User has a dog and lives in NYC") -- DISJUNCTION: Alternatives with "or" ("User works at Google or Microsoft") -- CONDITIONAL: If/then structure ("If user gets the job, they will move") -- EMBEDDED_QUOTE: Contains quoted multi-claim content -- COMPOUND_PREDICATE: Multiple predicates ("User studied and worked in Paris") -- CAUSAL_CHAIN: Because/since linking claims ("User is tired because they worked late") -- TEMPORAL_SEQUENCE: Multiple events in sequence - -### Test: Can part of this proposition be false while another part remains true? -If YES → Not atomic, needs decomposition""" - -COVERAGE_CRITERIA = """## Coverage Evaluation - -Coverage measures whether ALL extractable information from source messages is captured. - -### Extract: -- Explicit statements: Direct claims made by the speaker -- Embedded facts: Facts within larger statements ("I walked my dog" → has dog, walked dog) -- Relational info: Relationships between entities -- Temporal info: When things happened or states began -- Quantitative info: Numbers, amounts, frequencies - -### Gap Severity: -- CRITICAL: Core identity info missed (name, location, key relationships) -- IMPORTANT: Significant facts (job, major events, goals) -- MINOR: Supporting details (preferences, minor temporal info)""" - -FIDELITY_CRITERIA = """## Fidelity Evaluation - -Fidelity measures whether propositions faithfully represent source semantics. - -### Violations: -- HEDGE_REMOVAL: "I might get a dog" → "User will get a dog" -- NEGATION_FLIP: "I don't like coffee" → "User likes coffee" -- TEMPORAL_SHIFT: "I used to work at Google" → "User works at Google" -- QUANTITY_CHANGE: "I sometimes go running" → "User runs regularly" -- ATTRIBUTION_ERROR: "My sister loves jazz" → "User loves jazz" -- OVERGENERALIZATION: "I enjoyed that restaurant" → "User enjoys Italian food" -- OVERSPECIFICATION: "I have a pet" → "User has a dog" -- INFERENCE_AS_EXPLICIT: Implied → stated as fact - -### Severity: critical (changes meaning), major (significant), minor (slight imprecision)""" - -UTILITY_CRITERIA = """## Downstream Utility Evaluation - -Evaluates whether propositions can serve as valid logical premises. - -### Good Premise Requirements: -- CLEAR SUBJECT: Unambiguous who/what ("They are excited" fails) -- CLEAR PREDICATE: Unambiguous claim -- CONTEXTUALLY COMPLETE: Standalone ("User is nervous" → about what?) -- STABLE TRUTH VALUE: Definitively T/F ("kind of likes" is fuzzy) -- COMPOSABLE: Can participate in syllogisms (no embedded complexity) -- TEMPORAL CLARITY: When states/events apply - -### Suitability Ratings: -- EXCELLENT: Perfect premise, ideal for reasoning -- GOOD: Minor issues, usable -- MARGINAL: May cause ambiguity -- POOR: Significant issues -- UNUSABLE: Cannot serve as premise""" - - -# ============================================================================= -# JUDGE IMPLEMENTATION -# ============================================================================= - - -class ExplicitJudge: - llm_client: AsyncAnthropic | AsyncOpenAI - model: str - verbose: bool - provider: str - - def __init__( - self, - llm_client: AsyncAnthropic | AsyncOpenAI, - model: str = "claude-sonnet-4-20250514", - verbose: bool = False, - provider: str = "anthropic", - ): - self.llm_client = llm_client - self.model = model - self.verbose = verbose - self.provider = provider - if verbose: - logger.setLevel(logging.DEBUG) - - async def _call_llm( - self, - system: str, - user: str, - tool_def: dict[str, Any], - ) -> dict[str, Any]: - """Call LLM with tool use.""" - try: - if isinstance(self.llm_client, AsyncAnthropic): - resp = await asyncio.wait_for( - self.llm_client.messages.create( - model=self.model, - max_tokens=4000, - temperature=0.0, - system=system, - messages=[{"role": "user", "content": user}], - tools=[tool_def], # pyright: ignore[reportArgumentType] - tool_choice={"type": "tool", "name": tool_def["name"]}, - ), - timeout=120.0, - ) - for block in resp.content: - if block.type == "tool_use": - return dict(block.input) - return {} - - else: # AsyncOpenAI - openai_tool = { - "type": "function", - "function": { - "name": tool_def["name"], - "description": tool_def.get("description", ""), - "parameters": tool_def["input_schema"], - }, - } - resp = await asyncio.wait_for( - self.llm_client.chat.completions.create( - model=self.model, - max_tokens=4000, - temperature=0.0, - messages=[ - {"role": "system", "content": system}, - {"role": "user", "content": user}, - ], - tools=[openai_tool], # pyright: ignore[reportArgumentType] - tool_choice={ - "type": "function", - "function": {"name": tool_def["name"]}, - }, - ), - timeout=120.0, - ) - if resp.choices and resp.choices[0].message.tool_calls: - tool_call = resp.choices[0].message.tool_calls[0] - # Access function.arguments for standard function tool calls - func = getattr(tool_call, "function", None) - if func is not None: - arguments = getattr(func, "arguments", None) - if isinstance(arguments, str): - return json.loads(arguments) - return {} - except Exception as e: - logger.error(f"LLM call failed: {e}") - return {} - - async def evaluate_atomicity(self, propositions: list[str]) -> AtomicityReport: - if not propositions: - return AtomicityReport(0, 0, 1.0) - - tool_def = { - "name": "evaluate_atomicity", - "description": "Submit atomicity evaluation", - "input_schema": { - "type": "object", - "properties": { - "evaluations": { - "type": "array", - "items": { - "type": "object", - "properties": { - "index": {"type": "integer"}, - "is_atomic": {"type": "boolean"}, - "violation_types": { - "type": "array", - "items": { - "type": "string", - "enum": [v.value for v in AtomicityViolation], - }, - }, - "suggested_decomposition": { - "type": "array", - "items": {"type": "string"}, - }, - "reasoning": {"type": "string"}, - }, - "required": ["index", "is_atomic"], - }, - } - }, - "required": ["evaluations"], - }, - } - - props_text = "\n".join(f'{i + 1}. "{p}"' for i, p in enumerate(propositions)) - result = await self._call_llm( - ATOMICITY_CRITERIA, - f"Evaluate atomicity:\n\n{props_text}", - tool_def, - ) - - detailed: list[AtomicityResult] = [] - violations_by_type: dict[str, int] = {} - atomic_count = 0 - total_suggested = 0 - - evaluations: list[Any] = result.get("evaluations", []) - for ev in evaluations: - if not isinstance(ev, dict): - continue - idx_raw: Any = ev.get("index", 0) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(idx_raw, int): - continue - idx: int = idx_raw - 1 - if idx < 0 or idx >= len(propositions): - continue - is_atomic_raw: Any = ev.get("is_atomic", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - is_atomic: bool = ( - bool(is_atomic_raw) if isinstance(is_atomic_raw, bool) else True - ) - if is_atomic: - atomic_count += 1 - violations: list[AtomicityViolation] = [] - violation_types: list[Any] = ev.get("violation_types", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - for v in violation_types: # pyright: ignore[reportUnknownVariableType] - if not isinstance(v, str): - continue - try: - violations.append(AtomicityViolation(v)) - except ValueError as e: - logger.warning( - f"Skipping invalid AtomicityViolation type: {v} - {e}" - ) - continue - for v in violations: - violations_by_type[v.value] = violations_by_type.get(v.value, 0) + 1 - decomp_raw: list[Any] = ev.get("suggested_decomposition", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - decomp: list[str] = [str(d) for d in decomp_raw if isinstance(d, str)] # pyright: ignore[reportUnknownVariableType] - total_suggested += len(decomp) - reasoning_raw: Any = ev.get("reasoning", "") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - reasoning: str = ( - str(reasoning_raw) if isinstance(reasoning_raw, str) else "" - ) - detailed.append( - AtomicityResult( - propositions[idx], is_atomic, violations, decomp, reasoning - ) - ) - - return AtomicityReport( - len(propositions), - atomic_count, - atomic_count / len(propositions) if propositions else 1.0, - violations_by_type, - detailed, - len(propositions) - atomic_count, - atomic_count + total_suggested, - ) - - async def evaluate_coverage( - self, propositions: list[str], messages: list[dict[str, Any]], peer_name: str - ) -> CoverageReport: - user_msgs = [m for m in messages if m.get("speaker", "user") == "user"] - if not user_msgs: - return CoverageReport(0, len(propositions), 1.0, source_message_count=0) - - msgs_text = "\n\n".join( - f"[{i + 1}] {m.get('text', '')}" for i, m in enumerate(user_msgs) - ) - props_text = "\n".join(f'{i + 1}. "{p}"' for i, p in enumerate(propositions)) - - tool_def = { - "name": "evaluate_coverage", - "description": "Submit coverage evaluation", - "input_schema": { - "type": "object", - "properties": { - "estimated_total_claims": {"type": "integer"}, - "gaps": { - "type": "array", - "items": { - "type": "object", - "properties": { - "missing_claim": {"type": "string"}, - "source_quote": {"type": "string"}, - "severity": { - "type": "string", - "enum": ["critical", "important", "minor"], - }, - "reasoning": {"type": "string"}, - }, - "required": ["missing_claim", "severity"], - }, - }, - }, - "required": ["estimated_total_claims", "gaps"], - }, - } - - result = await self._call_llm( - COVERAGE_CRITERIA + f"\n\nPeer name: {peer_name}", - f"SOURCE MESSAGES:\n{msgs_text}\n\nEXTRACTED:\n{props_text}\n\nIdentify gaps.", - tool_def, - ) - - est_total = result.get("estimated_total_claims", len(propositions)) - gaps: list[CoverageGap] = [] - gaps_by_sev: dict[str, int] = {"critical": 0, "important": 0, "minor": 0} - - for g in result.get("gaps", []): - sev = g.get("severity", "minor") - gaps_by_sev[sev] = gaps_by_sev.get(sev, 0) + 1 - gaps.append( - CoverageGap( - g["missing_claim"], - g.get("source_quote", ""), - "0", - sev, - g.get("reasoning", ""), - ) - ) - - sev_weights = {"critical": 1.0, "important": 0.5, "minor": 0.25} - weighted_gaps = sum(sev_weights.get(g.severity, 0.25) for g in gaps) - score = ( - max(0, (est_total - weighted_gaps) / est_total) if est_total > 0 else 1.0 - ) - - return CoverageReport( - est_total, - len(propositions), - score, - gaps, - gaps_by_sev, - len(user_msgs), - len(propositions) / len(user_msgs) if user_msgs else 0, - ) - - async def evaluate_fidelity( - self, propositions: list[str], messages: list[dict[str, Any]] - ) -> FidelityReport: - if not propositions: - return FidelityReport(0, 0, 1.0) - - user_msgs = [m for m in messages if m.get("speaker", "user") == "user"] - msgs_text = "\n\n".join( - f"[{i + 1}] {m.get('text', '')}" for i, m in enumerate(user_msgs) - ) - props_text = "\n".join(f'{i + 1}. "{p}"' for i, p in enumerate(propositions)) - - tool_def = { - "name": "evaluate_fidelity", - "description": "Submit fidelity evaluation", - "input_schema": { - "type": "object", - "properties": { - "evaluations": { - "type": "array", - "items": { - "type": "object", - "properties": { - "index": {"type": "integer"}, - "is_faithful": {"type": "boolean"}, - "violations": { - "type": "array", - "items": { - "type": "object", - "properties": { - "type": { - "type": "string", - "enum": [ - v.value for v in FidelityViolation - ], - }, - "description": {"type": "string"}, - }, - }, - }, - "severity": { - "type": "string", - "enum": ["none", "minor", "major", "critical"], - }, - }, - "required": ["index", "is_faithful"], - }, - } - }, - "required": ["evaluations"], - }, - } - - result = await self._call_llm( - FIDELITY_CRITERIA, - f"SOURCE:\n{msgs_text}\n\nPROPOSITIONS:\n{props_text}", - tool_def, - ) - - detailed: list[FidelityResult] = [] - violations_by_type: dict[str, int] = {} - violations_by_sev: dict[str, int] = { - "none": 0, - "minor": 0, - "major": 0, - "critical": 0, - } - faithful_count = 0 - - evaluations: list[Any] = result.get("evaluations", []) - for ev in evaluations: - if not isinstance(ev, dict): - continue - idx_raw: Any = ev.get("index", 0) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(idx_raw, int): - continue - idx: int = idx_raw - 1 - if idx < 0 or idx >= len(propositions): - continue - is_faithful_raw: Any = ev.get("is_faithful", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - is_faithful: bool = ( - bool(is_faithful_raw) if isinstance(is_faithful_raw, bool) else True - ) - if is_faithful: - faithful_count += 1 - sev_raw: Any = ev.get("severity", "none") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - sev: str = str(sev_raw) if isinstance(sev_raw, str) else "none" - violations_by_sev[sev] = violations_by_sev.get(sev, 0) + 1 - violations: list[tuple[FidelityViolation, str]] = [] - violations_raw: list[Any] = ev.get("violations", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - for v in violations_raw: # pyright: ignore[reportUnknownVariableType] - if not isinstance(v, dict): - continue - try: - vtype_str_raw: Any = v.get("type", "") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(vtype_str_raw, str): - continue - vtype_str: str = vtype_str_raw - vtype = FidelityViolation(vtype_str) - violations_by_type[vtype.value] = ( - violations_by_type.get(vtype.value, 0) + 1 - ) - desc_raw: Any = v.get("description", "") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - desc: str = str(desc_raw) if isinstance(desc_raw, str) else "" - violations.append((vtype, desc)) - except (ValueError, KeyError) as e: - # v is already confirmed to be a dict from the earlier isinstance check - vtype_unknown: Any = v.get("type") # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType] - logger.warning( - f"Skipping invalid FidelityViolation type: {vtype_unknown} - {e}" - ) - continue - detailed.append( - FidelityResult(propositions[idx], is_faithful, violations, "", sev) - ) - - sev_penalties = {"none": 0, "minor": 0.25, "major": 0.5, "critical": 1.0} - penalty = sum(sev_penalties.get(r.severity, 0) for r in detailed) - score = max(0, 1 - penalty / len(propositions)) if propositions else 1.0 - - return FidelityReport( - len(propositions), - faithful_count, - score, - violations_by_type, - violations_by_sev, - detailed, - ) - - async def evaluate_efficiency(self, propositions: list[str]) -> EfficiencyReport: - if len(propositions) <= 1: - return EfficiencyReport(len(propositions), len(propositions), 1.0) - - props_text = "\n".join(f'{i + 1}. "{p}"' for i, p in enumerate(propositions)) - - tool_def = { - "name": "evaluate_efficiency", - "description": "Submit redundancy analysis", - "input_schema": { - "type": "object", - "properties": { - "redundancy_clusters": { - "type": "array", - "items": { - "type": "object", - "properties": { - "proposition_indices": { - "type": "array", - "items": {"type": "integer"}, - }, - "canonical_form": {"type": "string"}, - "redundancy_type": { - "type": "string", - "enum": [ - "exact_duplicate", - "near_duplicate", - "subsumption", - "overlap", - ], - }, - }, - "required": [ - "proposition_indices", - "canonical_form", - "redundancy_type", - ], - }, - }, - "unique_proposition_count": {"type": "integer"}, - }, - "required": ["redundancy_clusters", "unique_proposition_count"], - }, - } - - result = await self._call_llm( - "Identify redundant propositions (exact duplicates, near duplicates, subsumptions, overlaps).", - f"PROPOSITIONS:\n{props_text}", - tool_def, - ) - - clusters: list[RedundancyCluster] = [] - exact = near = subs = 0 - - redundancy_clusters_raw: list[Any] = result.get("redundancy_clusters", []) - for c in redundancy_clusters_raw: - if not isinstance(c, dict): - continue - indices_raw: Any = c.get("proposition_indices", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(indices_raw, list): - continue - indices: list[int] = [i for i in indices_raw if isinstance(i, int)] # pyright: ignore[reportUnknownVariableType] - props: list[str] = [ - propositions[i - 1] for i in indices if 0 < i <= len(propositions) - ] - rtype_raw: Any = c.get("redundancy_type", "overlap") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - rtype: str = str(rtype_raw) if isinstance(rtype_raw, str) else "overlap" - count = len(props) - 1 if props else 0 - if rtype == "exact_duplicate": - exact += count - elif rtype == "near_duplicate": - near += count - elif rtype == "subsumption": - subs += count - canonical_raw: Any = c.get("canonical_form", "") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - canonical: str = ( - str(canonical_raw) if isinstance(canonical_raw, str) else "" - ) - clusters.append(RedundancyCluster(props, canonical, rtype)) - - unique = result.get("unique_proposition_count", len(propositions)) - - return EfficiencyReport( - len(propositions), - unique, - unique / len(propositions) if propositions else 1.0, - clusters, - exact, - near, - subs, - ) - - async def evaluate_utility( - self, propositions: list[str], peer_name: str - ) -> DownstreamUtilityReport: - if not propositions: - return DownstreamUtilityReport(0, utility_score=1.0) - - props_text = "\n".join(f'{i + 1}. "{p}"' for i, p in enumerate(propositions)) - - tool_def = { - "name": "evaluate_utility", - "description": "Submit utility evaluation", - "input_schema": { - "type": "object", - "properties": { - "evaluations": { - "type": "array", - "items": { - "type": "object", - "properties": { - "index": {"type": "integer"}, - "has_clear_subject": {"type": "boolean"}, - "has_clear_predicate": {"type": "boolean"}, - "is_contextually_complete": {"type": "boolean"}, - "has_stable_truth_value": {"type": "boolean"}, - "is_composable": {"type": "boolean"}, - "temporal_handling": { - "type": "string", - "enum": ["appropriate", "missing", "excessive"], - }, - "suitability": { - "type": "string", - "enum": [s.value for s in PremiseSuitability], - }, - "issues": { - "type": "array", - "items": {"type": "string"}, - }, - }, - "required": ["index", "suitability"], - }, - } - }, - "required": ["evaluations"], - }, - } - - result = await self._call_llm( - UTILITY_CRITERIA + f"\n\nPeer: {peer_name}", - f"PROPOSITIONS:\n{props_text}\n\nEvaluate as logical premises.", - tool_def, - ) - - detailed: list[DownstreamUtilityResult] = [] - dist: dict[str, int] = {s.value: 0 for s in PremiseSuitability} - clarity = complete = stable = compos = temporal = 0 - - evaluations: list[Any] = result.get("evaluations", []) - for ev in evaluations: - if not isinstance(ev, dict): - continue - idx_raw: Any = ev.get("index", 0) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(idx_raw, int): - continue - idx: int = idx_raw - 1 - if idx < 0 or idx >= len(propositions): - continue - suit_raw: Any = ev.get("suitability", "marginal") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - suit: str = str(suit_raw) if isinstance(suit_raw, str) else "marginal" - dist[suit] = dist.get(suit, 0) + 1 - - subj_raw: Any = ev.get("has_clear_subject", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - subj: bool = bool(subj_raw) if isinstance(subj_raw, bool) else True - pred_raw: Any = ev.get("has_clear_predicate", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - pred: bool = bool(pred_raw) if isinstance(pred_raw, bool) else True - comp_raw: Any = ev.get("is_contextually_complete", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - comp: bool = bool(comp_raw) if isinstance(comp_raw, bool) else True - stab_raw: Any = ev.get("has_stable_truth_value", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - stab: bool = bool(stab_raw) if isinstance(stab_raw, bool) else True - comb_raw: Any = ev.get("is_composable", True) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - comb: bool = bool(comb_raw) if isinstance(comb_raw, bool) else True - temp_raw: Any = ev.get("temporal_handling", "appropriate") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - temp: str = str(temp_raw) if isinstance(temp_raw, str) else "appropriate" - - if subj and pred: - clarity += 1 - if comp: - complete += 1 - if stab: - stable += 1 - if comb: - compos += 1 - if temp == "appropriate": - temporal += 1 - - issues_raw: Any = ev.get("issues", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(issues_raw, list): - issues_raw = [] - issues: list[str] = [str(i) for i in issues_raw if isinstance(i, str)] # pyright: ignore[reportUnknownVariableType] - detailed.append( - DownstreamUtilityResult( - propositions[idx], - PremiseSuitability(suit), - issues, - subj, - pred, - comp, - stab, - comb, - temp, - ) - ) - - n = len(propositions) - weights = { - "excellent": 1.0, - "good": 0.8, - "marginal": 0.5, - "poor": 0.2, - "unusable": 0.0, - } - score = sum(dist[s] * weights[s] for s in dist) / n if n else 1.0 - - return DownstreamUtilityReport( - n, - dist, - clarity / n if n else 1.0, - complete / n if n else 1.0, - stable / n if n else 1.0, - compos / n if n else 1.0, - temporal / n if n else 1.0, - score, - detailed, - ) - - async def evaluate( - self, - propositions: list[str], - messages: list[dict[str, Any]], - peer_name: str, - conversation_id: str = "", - weights: dict[str, float] | None = None, - ) -> EvaluationResult: - logger.info(f"Evaluating {len(propositions)} propositions...") - - atom, cov, fid, eff, util = await asyncio.gather( - self.evaluate_atomicity(propositions), - self.evaluate_coverage(propositions, messages, peer_name), - self.evaluate_fidelity(propositions, messages), - self.evaluate_efficiency(propositions), - self.evaluate_utility(propositions, peer_name), - ) - - user_msgs = [m for m in messages if m.get("speaker", "user") == "user"] - - result = EvaluationResult( - conversation_id, - peer_name, - len(propositions), - len(user_msgs), - atom, - cov, - fid, - eff, - util, - ) - result.compute_overall_score(weights) - - logger.info(f"Evaluation complete. Overall: {result.overall_score:.2%}") - return result - - -# ============================================================================= -# TRACE PARSING -# ============================================================================= - - -def load_traces_from_json(path: Path) -> list[dict[str, Any]]: - """Load traces from a JSON or JSONL file. - - Supports: - - JSON array: [{"trace": 1}, {"trace": 2}] - - JSON object: {"trace": 1} - - JSONL: One JSON object per line - """ - traces: list[dict[str, Any]] = [] - - # Try JSONL format first (one JSON per line) - try: - with open(path) as f: - first_line = f.readline().strip() - if first_line and not first_line.startswith("["): - # Likely JSONL format - f.seek(0) # Reset to beginning - for line_num, line in enumerate(f, 1): - line = line.strip() - if not line: - continue - try: - trace: Any = json.loads(line) - if isinstance(trace, dict): - traces.append(trace) # pyright: ignore[reportUnknownArgumentType] - except json.JSONDecodeError as e: - logger.warning( - f"Skipping invalid JSON at line {line_num} in {path}: {e}" - ) - - if traces: - logger.info(f"Loaded {len(traces)} traces from JSONL file: {path}") - return traces - except Exception as e: - logger.debug(f"Not JSONL format, trying standard JSON: {e}") - - # Try standard JSON format - try: - with open(path) as f: - data: Any = json.load(f) - - if isinstance(data, list): - return data # pyright: ignore[reportUnknownVariableType] - elif isinstance(data, dict): - # Single trace, wrap in list - return [data] - else: - raise ValueError(f"Unexpected JSON format in {path}") - except json.JSONDecodeError as e: - raise ValueError(f"Failed to parse {path} as JSON or JSONL: {e}") from e - - -def extract_propositions(trace: dict[str, Any]) -> list[str]: - output: Any = trace.get("output", {}) - if not isinstance(output, dict): - return [] - content: Any = output.get("content", {}) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(content, dict): - return [] - explicit: Any = content.get("explicit", []) # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - if not isinstance(explicit, list): - return [] - propositions: list[str] = [] - for item in explicit: # pyright: ignore[reportUnknownVariableType] - if isinstance(item, dict) and "content" in item: - content_val: Any = item["content"] # pyright: ignore[reportUnknownVariableType] - if isinstance(content_val, str): - propositions.append(content_val) - return propositions - - -def extract_messages(trace: dict[str, Any]) -> list[dict[str, Any]]: - input_data: Any = trace.get("input", {}) - if not isinstance(input_data, dict): - return [] - prompt_raw: Any = input_data.get("prompt", "") # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType] - prompt: str = str(prompt_raw) if isinstance(prompt_raw, str) else "" - messages: list[dict[str, Any]] = [] - - if "" in prompt: - section = prompt.split("")[1].split("")[0] - for line in section.strip().split("\n"): - line = line.strip() - if not line: - continue - parts = line.split(" ", 3) - if len(parts) >= 3: - speaker: str = parts[2].rstrip(":") - text: str = parts[3] if len(parts) > 3 else "" - if "->->" in text: - text = text.split("->->")[0].strip() - messages.append({"speaker": speaker, "text": text}) - - return messages - - -def extract_conversation_id(trace: dict[str, Any], index: int) -> str: - """Extract or generate a conversation ID for a trace.""" - # Try common ID fields - if "conversation_id" in trace: - return trace["conversation_id"] - if "id" in trace: - return trace["id"] - # Generate from index - return f"trace_{index:04d}" - - -def extract_peer_name(trace: dict[str, Any]) -> str: - """Extract peer name from trace propositions.""" - for prop in extract_propositions(trace): - prop_lower = prop.lower() - # Pattern: "user's name is Victor" or "User is named Victor" - if "name is" in prop_lower: - parts = prop_lower.split("name is") - if len(parts) > 1: - name = parts[1].strip().rstrip(".").split()[0] - return name.capitalize() - if "is named" in prop_lower: - parts = prop_lower.split("is named") - if len(parts) > 1: - name = parts[1].strip().rstrip(".").split()[0] - return name.capitalize() - return "user" - - -# ============================================================================= -# OUTPUT -# ============================================================================= - - -def print_summary(result: EvaluationResult) -> None: - print("\n" + "=" * 70) - print(f"EVALUATION: {result.conversation_id}") - print("=" * 70) - print( - f"Peer: {result.peer_name} | Props: {result.proposition_count} | Messages: {result.source_message_count}" - ) - - print(f"\n{'OVERALL SCORE:':<20} {result.overall_score:.1%}") - print("-" * 40) - print( - f"{'Atomicity:':<20} {result.atomicity.score:.1%} ({result.atomicity.atomic_count}/{result.atomicity.total_propositions} atomic)" - ) - print( - f"{'Coverage:':<20} {result.coverage.coverage_score:.1%} ({len(result.coverage.gaps)} gaps)" - ) - print( - f"{'Fidelity:':<20} {result.fidelity.fidelity_score:.1%} ({result.fidelity.faithful_count}/{result.fidelity.total_propositions} faithful)" - ) - print( - f"{'Efficiency:':<20} {result.efficiency.efficiency_score:.1%} ({result.efficiency.unique_propositions}/{result.efficiency.total_propositions} unique)" - ) - print(f"{'Utility:':<20} {result.downstream_utility.utility_score:.1%}") - - # Show issues - if result.coverage.gaps: - crit = [g for g in result.coverage.gaps if g.severity == "critical"] - if crit: - print(f"\n⚠️ Critical coverage gaps ({len(crit)}):") - for g in crit[:3]: - print(f" - {g.missing_claim[:60]}...") - - non_atomic = [r for r in result.atomicity.detailed_results if not r.is_atomic] - if non_atomic: - print(f"\n⚠️ Non-atomic propositions ({len(non_atomic)}):") - for r in non_atomic[:2]: - print(f' - "{r.proposition[:50]}..."') - if r.suggested_decomposition: - print(f" → Split into: {r.suggested_decomposition[:2]}") - - print("=" * 70) - - -# ============================================================================= -# MAIN -# ============================================================================= - - -async def main(): - parser = argparse.ArgumentParser(description="Run explicit derivation benchmark") - parser.add_argument( - "--traces", type=Path, help="JSON or JSONL file containing traces" - ) - parser.add_argument( - "--trace-dir", type=Path, help="Directory of JSON/JSONL trace files" - ) - parser.add_argument( - "--output-dir", type=Path, default=Path("tests/bench/eval_results") - ) - parser.add_argument( - "--provider", - choices=["anthropic", "openai", "openrouter"], - default="anthropic", - help="LLM provider to use (default: anthropic)", - ) - parser.add_argument( - "--api-key", type=str, help="API key (overrides environment variable)" - ) - parser.add_argument("--model", default="claude-sonnet-4-20250514") - parser.add_argument("--verbose", "-v", action="store_true") - parser.add_argument("--weights", type=str, help="JSON string of custom weights") - parser.add_argument("--limit", type=int, help="Only evaluate first N traces") - parser.add_argument( - "--sample", type=float, help="Randomly sample this fraction of traces (0.0-1.0)" - ) - parser.add_argument( - "--batch-size", - type=int, - default=1, - help="Number of traces to process concurrently (default: 1)", - ) - args = parser.parse_args() - - # Get API key from argument or environment - api_key = args.api_key - if not api_key: - if args.provider == "anthropic": - api_key = os.getenv("ANTHROPIC_API_KEY") or os.getenv( - "LLM_ANTHROPIC_API_KEY" - ) - if not api_key: - logger.error( - "Set ANTHROPIC_API_KEY or LLM_ANTHROPIC_API_KEY, or use --api-key" - ) - return 1 - elif args.provider == "openai": - api_key = os.getenv("OPENAI_API_KEY") - if not api_key: - logger.error("Set OPENAI_API_KEY or use --api-key") - return 1 - elif args.provider == "openrouter": - api_key = os.getenv("OPENROUTER_API_KEY") - if not api_key: - logger.error("Set OPENROUTER_API_KEY or use --api-key") - return 1 - - # Initialize client based on provider - if args.provider == "anthropic": - client = AsyncAnthropic(api_key=api_key) - elif args.provider == "openai": - client = AsyncOpenAI(api_key=api_key) - elif args.provider == "openrouter": - # OpenRouter uses OpenAI-compatible API - client = AsyncOpenAI( - api_key=api_key, - base_url="https://openrouter.ai/api/v1", - ) - else: - logger.error(f"Unsupported provider: {args.provider}") - return 1 - - judge = ExplicitJudge(client, args.model, args.verbose, args.provider) - - weights = json.loads(args.weights) if args.weights else None - - # Collect all traces - all_traces: list[tuple[dict[str, Any], str]] = [] # (trace, source_file) - - if args.traces: - traces = load_traces_from_json(args.traces) - all_traces.extend((t, args.traces.name) for t in traces) - elif args.trace_dir: - # Support both .json and .jsonl extensions - for json_file in list(args.trace_dir.glob("*.json")) + list( - args.trace_dir.glob("*.jsonl") - ): - traces = load_traces_from_json(json_file) - all_traces.extend((t, json_file.name) for t in traces) - else: - parser.error("Specify --traces or --trace-dir") - - print(f"Loaded {len(all_traces)} trace(s)") - - # Apply sampling/limiting - if args.sample and 0 < args.sample < 1: - sample_size = max(1, int(len(all_traces) * args.sample)) - all_traces = random.sample(all_traces, sample_size) - print(f"Sampled {len(all_traces)} traces ({args.sample:.0%})") - - if args.limit and args.limit < len(all_traces): - all_traces = all_traces[: args.limit] - print(f"Limited to first {args.limit} traces") - - print(f"\nEvaluating {len(all_traces)} trace(s)...\n") - - # Process traces in batches if batch_size > 1 - results: list[EvaluationResult] = [] - - async def process_trace( - idx: int, trace: dict[str, Any], source_file: str - ) -> EvaluationResult | None: - """Process a single trace and return the result.""" - try: - props = extract_propositions(trace) - if not props: - logger.warning( - f"Trace {idx} from {source_file} has no propositions, skipping" - ) - return None - - msgs = extract_messages(trace) - peer = extract_peer_name(trace) - conv_id = extract_conversation_id(trace, idx) - - print( - f"[{idx + 1}/{len(all_traces)}] Evaluating {conv_id} ({len(props)} props)..." - ) - - result = await judge.evaluate(props, msgs, peer, conv_id, weights) - print_summary(result) - return result - - except Exception as e: - logger.error(f"Failed trace {idx} from {source_file}: {e}") - if args.verbose: - import traceback - - traceback.print_exc() - return None - - # Process in batches - if args.batch_size > 1: - print(f"Processing traces in batches of {args.batch_size}...\n") - for i in range(0, len(all_traces), args.batch_size): - batch = all_traces[i : i + args.batch_size] - batch_results = await asyncio.gather( - *[ - process_trace(i + j, trace, source_file) - for j, (trace, source_file) in enumerate(batch) - ], - return_exceptions=True, - ) - for result in batch_results: - if isinstance(result, EvaluationResult): - results.append(result) - elif isinstance(result, Exception): - logger.error(f"Batch processing error: {result}") - else: - # Process sequentially - for idx, (trace, source_file) in enumerate(all_traces): - result = await process_trace(idx, trace, source_file) - if result: - results.append(result) - - if results: - args.output_dir.mkdir(parents=True, exist_ok=True) - ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") - out_file = args.output_dir / f"explicit_{ts}.json" - - averages: dict[str, float] = { - "overall": sum(r.overall_score for r in results) / len(results), - "atomicity": sum(r.atomicity.score for r in results) / len(results), - "coverage": sum(r.coverage.coverage_score for r in results) / len(results), - "fidelity": sum(r.fidelity.fidelity_score for r in results) / len(results), - "efficiency": sum(r.efficiency.efficiency_score for r in results) - / len(results), - "utility": sum(r.downstream_utility.utility_score for r in results) - / len(results), - } - - agg = { - "timestamp": ts, - "model": args.model, - "count": len(results), - "averages": averages, - "results": [r.to_dict() for r in results], - } - - with open(out_file, "w") as f: - json.dump(agg, f, indent=2) - - print(f"\n✅ Results saved to {out_file}") - - if len(results) > 1: - print("\n" + "=" * 70) - print("AGGREGATE RESULTS") - print("=" * 70) - print(f"Traces evaluated: {len(results)}") - print() - for k, v in averages.items(): - print(f" {k:<20} {v:.1%}") - else: - print("\n⚠️ No traces were successfully evaluated") - return 1 - - return 0 - - -if __name__ == "__main__": - exit(asyncio.run(main())) diff --git a/tests/bench/molecular.py b/tests/bench/molecular.py new file mode 100644 index 00000000..ff1a6ca4 --- /dev/null +++ b/tests/bench/molecular.py @@ -0,0 +1,1222 @@ +""" +MolecularBench - Molecular Facts Evaluation Benchmark +Author: 3un01a (3un01a@plasticlabs.ai) + +Based on "Molecular Facts: Desiderata for Decontextualization in LLM Fact Verification" +(Gunjal & Durrett, EMNLP 2024) + +## Core Principle + +Atomic facts are insufficient for verification because they lose context. +Molecular facts balance two competing criteria: + +1. **Decontextuality**: The claim can be fully interpreted without the original context +2. **Minimality**: The minimum information needed to achieve decontextuality + +## Evaluation Pipeline + +For each proposition, we evaluate: + +### Stage 1: Ambiguity Detection +Identify what makes the proposition context-dependent: +- Unresolved references (pronouns, demonstratives) +- Ambiguous entities (names that could refer to multiple people) +- Missing anchors (temporal, spatial, comparative) +- Implicit knowledge requirements + +### Stage 2: Decontextuality Assessment +Can a reader understand this claim in isolation? +- Entity identification: Can we uniquely identify WHO/WHAT? +- Event grounding: Can we identify WHEN/WHERE/HOW? +- Relation clarity: Are relationships explicit? + +### Stage 3: Minimality Assessment +Does the proposition avoid unnecessary elaboration? +- Information density: Is every word necessary? +- Claim independence: Does it contain only ONE verifiable claim? +- Elaboration check: Are qualifiers essential or excessive? + +## Scoring + +- Decontextuality ∈ [0, 1]: Higher = more standalone +- Minimality ∈ [0, 1]: Higher = more concise +- Molecular Score = √(Decontextuality * Minimality): Geometric mean captures balance + +## Usage + +python -m tests.bench.molecular --traces path/to/traces.jsonl +""" + +import argparse +import asyncio +import json +import logging +import random +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Any, cast + +import anthropic +import openai +from anthropic import AsyncAnthropic +from anthropic.types import TextBlock +from dotenv import load_dotenv +from openai import AsyncOpenAI + +from .runner_common import ( + configure_logging, + create_anthropic_client, + create_openai_client, + extract_conversation_id, + extract_messages, + extract_peer_name, + extract_propositions, + format_duration, + load_traces, +) + +# Load .env from bench directory +bench_dir = Path(__file__).parent +load_dotenv(bench_dir / ".env") + +logger = configure_logging(level=logging.INFO, name=__name__) + + +# ============================================================================= +# AMBIGUITY TYPES (Stage 1) +# ============================================================================= + + +class AmbiguityType(Enum): + """Types of ambiguity that prevent standalone interpretation.""" + + # Reference ambiguities + UNRESOLVED_PRONOUN = "unresolved_pronoun" # he, she, they, it + UNRESOLVED_DEMONSTRATIVE = "unresolved_demonstrative" # this, that, these + ANAPHORIC_REFERENCE = "anaphoric_reference" # "the project", "the meeting" + ELLIPSIS = "ellipsis" # Incomplete structure relying on context + + # Entity ambiguities + AMBIGUOUS_PERSON = "ambiguous_person" # Common name without disambiguation + AMBIGUOUS_ORGANIZATION = "ambiguous_organization" # Company name collision + AMBIGUOUS_LOCATION = "ambiguous_location" # City/place name collision + AMBIGUOUS_EVENT = "ambiguous_event" # Event without unique identification + + # Anchor ambiguities + RELATIVE_TIME = "relative_time" # "recently", "last week", "soon" + RELATIVE_LOCATION = "relative_location" # "here", "there", "nearby" + IMPLICIT_COMPARISON = "implicit_comparison" # "better", "more", "preferred" + + # Knowledge ambiguities + DOMAIN_JARGON = "domain_jargon" # Unexpanded acronyms, technical terms + IMPLICIT_CONTEXT = "implicit_context" # Requires world knowledge to interpret + + +class DecontextualityIssue(Enum): + """Issues that prevent a proposition from standing alone.""" + + UNIDENTIFIABLE_SUBJECT = "unidentifiable_subject" # Can't tell WHO + UNIDENTIFIABLE_OBJECT = "unidentifiable_object" # Can't tell WHAT + UNIDENTIFIABLE_TIME = "unidentifiable_time" # Can't tell WHEN + UNIDENTIFIABLE_PLACE = "unidentifiable_place" # Can't tell WHERE + UNCLEAR_RELATION = "unclear_relation" # Can't tell the relationship + VERIFICATION_IMPOSSIBLE = "verification_impossible" # Can't fact-check this + + +class MinimalityIssue(Enum): + """Issues where a proposition contains unnecessary information.""" + + NESTED_CLAIM = "nested_claim" # Contains multiple verifiable facts + EXCESSIVE_QUALIFIER = "excessive_qualifier" # Unnecessary identifying info + REDUNDANT_CONTEXT = "redundant_context" # Context that doesn't aid verification + EXPLANATORY_ADDITION = "explanatory_addition" # Added explanation not in source + TEMPORAL_OVER_PRECISION = "temporal_over_precision" # Unnecessary date detail + BIOGRAPHICAL_OVERLOAD = "biographical_overload" # Too much background info + + +# ============================================================================= +# DATA STRUCTURES +# ============================================================================= + + +@dataclass +class AmbiguityAnalysis: + """Result of ambiguity detection for a proposition.""" + + proposition: str + ambiguities: list[tuple[AmbiguityType, str, str]] = field(default_factory=list) + # Each tuple: (type, problematic_span, explanation) + ambiguity_count: int = 0 + primary_subject: str = "" # Who/what is this about? + requires_context: bool = False + disambiguation_needed: list[str] = field(default_factory=list) + + +@dataclass +class DecontextualityAnalysis: + """Result of decontextuality assessment.""" + + proposition: str + score: float # 0-1, higher = more decontextualized + is_standalone: bool + issues: list[tuple[DecontextualityIssue, str]] = field(default_factory=list) + + # Specific checks + subject_identifiable: bool = True + object_identifiable: bool = True + time_anchored: bool = True + place_anchored: bool = True + relation_clear: bool = True + + # Suggested fix + suggested_decontextualization: str = "" + + +@dataclass +class MinimalityAnalysis: + """Result of minimality assessment.""" + + proposition: str + score: float # 0-1, higher = more minimal + is_minimal: bool + issues: list[tuple[MinimalityIssue, str]] = field(default_factory=list) + + # Metrics + token_count: int = 0 + claim_count: int = 1 # Should be 1 for molecular + excess_tokens: int = 0 + + # Nested claims found + nested_claims: list[str] = field(default_factory=list) + + # Suggested reduction + suggested_reduction: str = "" + + +@dataclass +class MolecularAnalysis: + """Complete molecular analysis for a single proposition.""" + + proposition: str + ambiguity: AmbiguityAnalysis + decontextuality: DecontextualityAnalysis + minimality: MinimalityAnalysis + + # Molecular score: geometric mean of decontextuality and minimality + molecular_score: float = 0.0 + + # Classification + is_molecular: bool = False # True if both decontextualized AND minimal + classification: str = "" # "molecular", "too_atomic", "too_verbose", "problematic" + + def compute_molecular_score(self) -> float: + """Compute the molecular score as geometric mean.""" + d = self.decontextuality.score + m = self.minimality.score + self.molecular_score = (d * m) ** 0.5 + + # Classification based on scores + if d >= 0.8 and m >= 0.8: + self.is_molecular = True + self.classification = "molecular" + elif d < 0.5: + self.classification = "too_atomic" # Needs more context + elif m < 0.5: + self.classification = "too_verbose" # Over-elaborated + else: + self.classification = "borderline" + + return self.molecular_score + + +@dataclass +class MolecularReport: + """Aggregate report for all propositions in a trace.""" + + conversation_id: str + peer_name: str + proposition_count: int + source_message_count: int + + # Aggregate scores + avg_decontextuality: float = 0.0 + avg_minimality: float = 0.0 + avg_molecular: float = 0.0 + + # Classification counts + molecular_count: int = 0 + too_atomic_count: int = 0 + too_verbose_count: int = 0 + borderline_count: int = 0 + + # Common issues + ambiguity_distribution: dict[str, int] = field(default_factory=dict) + decontextuality_issues: dict[str, int] = field(default_factory=dict) + minimality_issues: dict[str, int] = field(default_factory=dict) + + # Detailed results + analyses: list[MolecularAnalysis] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "conversation_id": self.conversation_id, + "peer_name": self.peer_name, + "proposition_count": self.proposition_count, + "source_message_count": self.source_message_count, + "scores": { + "molecular": round(self.avg_molecular, 4), + "decontextuality": round(self.avg_decontextuality, 4), + "minimality": round(self.avg_minimality, 4), + }, + "classification": { + "molecular": self.molecular_count, + "too_atomic": self.too_atomic_count, + "too_verbose": self.too_verbose_count, + "borderline": self.borderline_count, + "molecular_rate": round( + self.molecular_count / self.proposition_count, 3 + ) + if self.proposition_count > 0 + else 0, + }, + "issues": { + "ambiguity_types": self.ambiguity_distribution, + "decontextuality_issues": self.decontextuality_issues, + "minimality_issues": self.minimality_issues, + }, + "propositions": [ + { + "text": a.proposition, + "molecular_score": round(a.molecular_score, 3), + "decontextuality": round(a.decontextuality.score, 3), + "minimality": round(a.minimality.score, 3), + "classification": a.classification, + "is_molecular": a.is_molecular, + } + for a in self.analyses + ], + } + + +# ============================================================================= +# EVALUATION PROMPTS +# ============================================================================= + + +AMBIGUITY_DETECTION_PROMPT = """You are analyzing propositions for ambiguity - elements that prevent standalone interpretation. + +## Ambiguity Types + +**Reference Ambiguities** (require prior context to resolve): +- UNRESOLVED_PRONOUN: "he", "she", "they", "it" without clear antecedent +- UNRESOLVED_DEMONSTRATIVE: "this", "that", "these" without clear referent +- ANAPHORIC_REFERENCE: "the project", "the meeting" - definite reference without introduction +- ELLIPSIS: Grammatically incomplete, relies on context + +**Entity Ambiguities** (multiple possible referents): +- AMBIGUOUS_PERSON: Common name (e.g., "John Smith", "Maria Garcia") without disambiguation +- AMBIGUOUS_ORGANIZATION: Company name that could refer to multiple entities +- AMBIGUOUS_LOCATION: Place name collision (e.g., "Springfield", "Cambridge") +- AMBIGUOUS_EVENT: Event without unique identification + +**Anchor Ambiguities** (relative without absolute reference): +- RELATIVE_TIME: "recently", "last week", "tomorrow" without date anchor +- RELATIVE_LOCATION: "here", "there", "nearby" without location anchor +- IMPLICIT_COMPARISON: "better", "more", "preferred" without baseline + +**Knowledge Ambiguities** (require external knowledge): +- DOMAIN_JARGON: Unexpanded acronyms, technical terms +- IMPLICIT_CONTEXT: Requires unstated world knowledge + +## Your Task + +For each proposition, identify ALL ambiguities that would prevent a reader from fully understanding the claim in isolation (without the original conversation). + +For each ambiguity, specify: +1. The type (from the enum above) +2. The problematic span (the actual text causing ambiguity) +3. Brief explanation of why it's ambiguous + +Also identify: +- The primary subject (who/what is this proposition about?) +- What disambiguation would be needed to make it standalone""" + + +DECONTEXTUALITY_PROMPT = """You are evaluating whether propositions can stand alone without their original context. + +## Decontextuality Criteria + +A proposition is DECONTEXTUALIZED if a reader can fully understand it WITHOUT: +- Reading the original conversation +- Knowing who the speakers are +- Having any prior context + +### Checks + +**Subject Identifiable**: Can we uniquely identify WHO the proposition is about? +- Bad: "He is a software engineer" (who is "he"?) +- Good: "Victor Chen is a software engineer" +- Good: "The user is a software engineer" (if "user" is the established convention) + +**Object Identifiable**: Can we uniquely identify WHAT is being described? +- Bad: "User likes it" (what is "it"?) +- Good: "User likes Python programming" + +**Time Anchored**: If temporal, can we identify WHEN? +- Bad: "User started recently" (when is recently?) +- Good: "User started in January 2024" +- Acceptable: "User currently works at X" (present tense is self-anchoring) + +**Place Anchored**: If spatial, can we identify WHERE? +- Bad: "User lives there" (where is there?) +- Good: "User lives in San Francisco" + +**Relation Clear**: Is the relationship/predicate unambiguous? +- Bad: "User has a connection to Google" (what kind of connection?) +- Good: "User works at Google" + +### Scoring Guide + +- 1.0: Fully standalone, no context needed +- 0.8: Minor ambiguity but interpretable +- 0.6: Some context would help but core meaning clear +- 0.4: Significant ambiguity, meaning partially unclear +- 0.2: Major context needed +- 0.0: Unintelligible without context + +### Issue Types +- UNIDENTIFIABLE_SUBJECT: Can't determine who/what this is about +- UNIDENTIFIABLE_OBJECT: Can't determine the object/target +- UNIDENTIFIABLE_TIME: Temporal reference unclear +- UNIDENTIFIABLE_PLACE: Spatial reference unclear +- UNCLEAR_RELATION: The relationship/action is ambiguous +- VERIFICATION_IMPOSSIBLE: Cannot be fact-checked as written""" + + +MINIMALITY_PROMPT = """You are evaluating whether propositions are MINIMAL - containing only the information necessary for standalone interpretation. + +## Minimality Criteria + +A proposition is MINIMAL if: +1. It contains exactly ONE verifiable claim +2. Every word contributes to that claim's interpretability +3. No unnecessary elaboration or explanation is added + +### The Molecular Sweet Spot + +Too Atomic (needs more context): +- "Ann won a medal" → Which Ann? What medal? When? + +Too Verbose (excessive elaboration): +- "Ann Jansson, the Swedish footballer born on 6 May 1957 who played for Hammarby IF DFF and later became a coach, won a bronze medal at the 1984 European Athletics Championships, an event organized by the European Athletics Association held in Prague, Czechoslovakia" + +Molecular (just right): +- "Ann Jansson, the Swedish footballer born 1957, won a medal at the 1984 European Athletics Championships" + +### Issue Types + +**NESTED_CLAIM**: Contains multiple independently verifiable facts +- "User's sister, who is a doctor, lives in Boston" (3 claims: has sister, sister is doctor, sister lives in Boston) + +**EXCESSIVE_QUALIFIER**: Unnecessary identifying information +- "John Smith, the 45-year-old Caucasian male software engineer" (when just "John Smith, software engineer at X" would suffice for disambiguation) + +**REDUNDANT_CONTEXT**: Context that doesn't aid verification +- "User, who we've been discussing cooking with, likes Italian food" + +**EXPLANATORY_ADDITION**: Added explanation not from source +- "User likes Python, which is a popular programming language" + +**TEMPORAL_OVER_PRECISION**: Unnecessary time detail +- "On Tuesday, March 5th, 2024 at 3:47 PM Pacific Standard Time, user mentioned..." + +**BIOGRAPHICAL_OVERLOAD**: Excessive background for simple claims +- "Victor Chen, born in 1990 in Shanghai, educated at MIT, currently residing in San Francisco, enjoys hiking" + +### Metrics + +- Token count: Ideal molecular range is 8-25 tokens +- Claim count: Should be exactly 1 +- Excess tokens: Tokens that could be removed without losing meaning + +### Scoring Guide + +- 1.0: Perfectly minimal, one claim, no excess +- 0.8: Slight excess but acceptable +- 0.6: Some unnecessary information +- 0.4: Multiple claims or significant bloat +- 0.2: Heavy over-elaboration +- 0.0: Massively verbose, many nested claims""" + + +# ============================================================================= +# MOLECULAR JUDGE +# ============================================================================= + + +class MolecularJudge: + """ + Evaluates propositions against the Molecular Facts criteria. + + Based on Gunjal & Durrett (2024): The key insight is that purely atomic + facts lose context needed for verification, but over-decontextualization + adds verifiable claims that could be wrong. Molecular facts are the sweet spot. + """ + + def __init__( + self, + llm_client: AsyncAnthropic | AsyncOpenAI, + model: str = "claude-sonnet-4-20250514", + verbose: bool = False, + provider: str = "anthropic", + ): + self.llm_client: AsyncAnthropic | AsyncOpenAI = llm_client + self.model: str = model + self.verbose: bool = verbose + self.provider: str = provider + if verbose: + logger.setLevel(logging.DEBUG) + + async def _call_llm( + self, system: str, user: str, response_schema: dict[str, Any] + ) -> dict[str, Any]: + """Call LLM without tool use, requesting JSON response directly.""" + try: + # Add JSON schema instruction to user prompt + schema_instruction = f"\n\nRespond with a JSON object matching this schema:\n```json\n{json.dumps(response_schema, indent=2)}\n```" + full_user_prompt = user + schema_instruction + + if isinstance(self.llm_client, AsyncAnthropic): + resp = await asyncio.wait_for( + self.llm_client.messages.create( + model=self.model, + max_tokens=4096, + temperature=0.0, + system=system, + messages=[{"role": "user", "content": full_user_prompt}], + ), + timeout=180.0, + ) + # Extract text content + text_content = "" + for block in resp.content: + if isinstance(block, TextBlock): + text_content += block.text + + # Try to extract JSON from the response + return self._extract_json(text_content) + else: + resp = await asyncio.wait_for( + self.llm_client.chat.completions.create( + model=self.model, + max_tokens=4096, + temperature=0.0, + messages=[ + {"role": "system", "content": system}, + {"role": "user", "content": full_user_prompt}, + ], + response_format={"type": "json_object"}, + ), + timeout=180.0, + ) + if resp.choices and resp.choices[0].message.content: + return json.loads(resp.choices[0].message.content) + return {} + except TimeoutError: + logger.exception( + "LLM call timed out (model=%s, provider=%s)", + self.model, + self.provider, + ) + return {} + except (anthropic.APIError, openai.APIError): + logger.exception( + "LLM API error (model=%s, provider=%s)", + self.model, + self.provider, + ) + return {} + except json.JSONDecodeError: + logger.exception( + "Failed to parse LLM response as JSON (model=%s, provider=%s)", + self.model, + self.provider, + ) + return {} + except Exception: + logger.exception( + "Unexpected error in LLM call (model=%s, provider=%s)", + self.model, + self.provider, + ) + raise + + def _extract_json(self, text: str) -> dict[str, Any]: + """Extract JSON from text response, handling markdown code blocks.""" + # Try direct JSON parsing first + try: + return json.loads(text.strip()) + except json.JSONDecodeError: + pass + + # Try extracting from markdown code block + import re + + json_match = re.search(r"```(?:json)?\s*\n(.*?)\n```", text, re.DOTALL) + if json_match: + try: + return json.loads(json_match.group(1)) + except json.JSONDecodeError: + pass + + # Try finding JSON object in text + brace_start = text.find("{") + if brace_start >= 0: + try: + return json.loads(text[brace_start:]) + except json.JSONDecodeError: + pass + + logger.warning(f"Failed to extract JSON from response: {text[:200]}") + return {} + + async def analyze_ambiguity( + self, propositions: list[str], source_context: str + ) -> list[AmbiguityAnalysis]: + """Stage 1: Detect ambiguities in each proposition.""" + + if not propositions: + return [] + + response_schema = { + "type": "object", + "properties": { + "analyses": { + "type": "array", + "items": { + "type": "object", + "properties": { + "index": {"type": "integer"}, + "ambiguities": { + "type": "array", + "items": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": [t.value for t in AmbiguityType], + }, + "span": {"type": "string"}, + "explanation": {"type": "string"}, + }, + "required": ["type", "span", "explanation"], + }, + }, + "primary_subject": {"type": "string"}, + "requires_context": {"type": "boolean"}, + "disambiguation_needed": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": [ + "index", + "ambiguities", + "primary_subject", + "requires_context", + ], + }, + } + }, + "required": ["analyses"], + } + + props_text = "\n".join(f'{i+1}. "{p}"' for i, p in enumerate(propositions)) + + result = await self._call_llm( + AMBIGUITY_DETECTION_PROMPT, + ( + f"SOURCE CONTEXT (for reference only):\n{source_context}\n\n" + + f"PROPOSITIONS TO ANALYZE:\n{props_text}\n\n" + + "Identify all ambiguities in each proposition." + ), + response_schema, + ) + + analyses: list[AmbiguityAnalysis] = [] + for prop in propositions: + analyses.append(AmbiguityAnalysis(proposition=prop)) + + for item in cast(list[dict[str, Any]], result.get("analyses", [])): + idx: int = int(item.get("index", 0)) - 1 + if 0 <= idx < len(analyses): + a: AmbiguityAnalysis = analyses[idx] + for amb in cast(list[dict[str, Any]], item.get("ambiguities", [])): + try: + atype = AmbiguityType(amb["type"]) + a.ambiguities.append( + (atype, amb.get("span", ""), amb.get("explanation", "")) + ) + except (ValueError, KeyError): + continue + a.ambiguity_count = len(a.ambiguities) + a.primary_subject = item.get("primary_subject", "") + a.requires_context = item.get("requires_context", False) + a.disambiguation_needed = item.get("disambiguation_needed", []) + + return analyses + + async def analyze_decontextuality( + self, propositions: list[str], source_context: str + ) -> list[DecontextualityAnalysis]: + """Stage 2: Assess how well each proposition stands alone.""" + + if not propositions: + return [] + + response_schema = { + "type": "object", + "properties": { + "analyses": { + "type": "array", + "items": { + "type": "object", + "properties": { + "index": {"type": "integer"}, + "score": {"type": "number", "minimum": 0, "maximum": 1}, + "is_standalone": {"type": "boolean"}, + "issues": { + "type": "array", + "items": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": [ + t.value for t in DecontextualityIssue + ], + }, + "description": {"type": "string"}, + }, + }, + }, + "subject_identifiable": {"type": "boolean"}, + "object_identifiable": {"type": "boolean"}, + "time_anchored": {"type": "boolean"}, + "place_anchored": {"type": "boolean"}, + "relation_clear": {"type": "boolean"}, + "suggested_decontextualization": {"type": "string"}, + }, + "required": ["index", "score", "is_standalone"], + }, + } + }, + "required": ["analyses"], + } + + props_text = "\n".join(f'{i+1}. "{p}"' for i, p in enumerate(propositions)) + + result = await self._call_llm( + DECONTEXTUALITY_PROMPT, + ( + f"SOURCE CONTEXT (the original conversation - propositions should be understandable WITHOUT this):\n{source_context}\n\n" + + f"PROPOSITIONS TO EVALUATE:\n{props_text}\n\n" + + "For each proposition, assess whether it can be understood in COMPLETE ISOLATION." + ), + response_schema, + ) + + analyses: list[DecontextualityAnalysis] = [] + for prop in propositions: + analyses.append( + DecontextualityAnalysis( + proposition=prop, score=0.5, is_standalone=False + ) + ) + + for item in cast(list[dict[str, Any]], result.get("analyses", [])): + idx: int = int(item.get("index", 0)) - 1 + if 0 <= idx < len(analyses): + a: DecontextualityAnalysis = analyses[idx] + a.score = item.get("score", 0.5) + a.is_standalone = item.get("is_standalone", False) + for issue in cast(list[dict[str, Any]], item.get("issues", [])): + try: + itype = DecontextualityIssue(issue["type"]) + a.issues.append((itype, issue.get("description", ""))) + except (ValueError, KeyError): + continue + a.subject_identifiable = item.get("subject_identifiable", True) + a.object_identifiable = item.get("object_identifiable", True) + a.time_anchored = item.get("time_anchored", True) + a.place_anchored = item.get("place_anchored", True) + a.relation_clear = item.get("relation_clear", True) + a.suggested_decontextualization = item.get( + "suggested_decontextualization", "" + ) + + return analyses + + async def analyze_minimality( + self, propositions: list[str], source_context: str + ) -> list[MinimalityAnalysis]: + """Stage 3: Assess whether each proposition avoids excessive elaboration.""" + + if not propositions: + return [] + + response_schema = { + "type": "object", + "properties": { + "analyses": { + "type": "array", + "items": { + "type": "object", + "properties": { + "index": {"type": "integer"}, + "score": {"type": "number", "minimum": 0, "maximum": 1}, + "is_minimal": {"type": "boolean"}, + "issues": { + "type": "array", + "items": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": [t.value for t in MinimalityIssue], + }, + "description": {"type": "string"}, + }, + }, + }, + "token_count": {"type": "integer"}, + "claim_count": {"type": "integer"}, + "excess_tokens": {"type": "integer"}, + "nested_claims": { + "type": "array", + "items": {"type": "string"}, + }, + "suggested_reduction": {"type": "string"}, + }, + "required": [ + "index", + "score", + "is_minimal", + "token_count", + "claim_count", + ], + }, + } + }, + "required": ["analyses"], + } + + props_text = "\n".join(f'{i+1}. "{p}"' for i, p in enumerate(propositions)) + + result = await self._call_llm( + MINIMALITY_PROMPT, + ( + f"SOURCE CONTEXT (for comparison):\n{source_context}\n\n" + + f"PROPOSITIONS TO EVALUATE:\n{props_text}\n\n" + + "For each proposition, assess whether it is MINIMAL (one claim, no excess)." + ), + response_schema, + ) + + analyses: list[MinimalityAnalysis] = [] + for prop in propositions: + analyses.append( + MinimalityAnalysis(proposition=prop, score=0.5, is_minimal=False) + ) + + for item in cast(list[dict[str, Any]], result.get("analyses", [])): + idx: int = int(item.get("index", 0)) - 1 + if 0 <= idx < len(analyses): + a: MinimalityAnalysis = analyses[idx] + a.score = item.get("score", 0.5) + a.is_minimal = item.get("is_minimal", False) + for issue in cast(list[dict[str, Any]], item.get("issues", [])): + try: + itype = MinimalityIssue(issue["type"]) + a.issues.append((itype, issue.get("description", ""))) + except (ValueError, KeyError): + continue + a.token_count = item.get("token_count", 0) + a.claim_count = item.get("claim_count", 1) + a.excess_tokens = item.get("excess_tokens", 0) + a.nested_claims = item.get("nested_claims", []) + a.suggested_reduction = item.get("suggested_reduction", "") + + return analyses + + async def evaluate( + self, + propositions: list[str], + messages: list[dict[str, Any]], + peer_name: str, + conversation_id: str = "", + ) -> MolecularReport: + """Run full molecular evaluation pipeline. + + Args: + propositions: Extracted fact strings to evaluate for molecular + quality (decontextuality, minimality, and ambiguity). + messages: Raw conversation messages used as source context. + Each dict should contain ``speaker`` and ``text`` keys. + peer_name: Name of the peer whose propositions are being evaluated. + conversation_id: Optional identifier for the conversation trace. + Defaults to an empty string. + + Returns: + A ``MolecularReport`` containing per-proposition molecular + analyses (ambiguity, decontextuality, minimality scores and + issues), aggregate scores, classification distribution, and + common issues found across propositions. + """ + + if not propositions: + return MolecularReport( + conversation_id=conversation_id, + peer_name=peer_name, + proposition_count=0, + source_message_count=len(messages), + ) + + # Build source context string + user_msgs = [m for m in messages if m.get("speaker", "user") == "user"] + source_context = "\n".join(f"- {m.get('text', '')}" for m in user_msgs) + + logger.info( + f"Evaluating {len(propositions)} propositions (3-stage molecular pipeline)..." + ) + + # Run all three stages in parallel + ambiguity_results, decontext_results, minimal_results = await asyncio.gather( + self.analyze_ambiguity(propositions, source_context), + self.analyze_decontextuality(propositions, source_context), + self.analyze_minimality(propositions, source_context), + ) + + # Combine into molecular analyses + analyses: list[MolecularAnalysis] = [] + for i, prop in enumerate(propositions): + analysis = MolecularAnalysis( + proposition=prop, + ambiguity=ambiguity_results[i] + if i < len(ambiguity_results) + else AmbiguityAnalysis(proposition=prop), + decontextuality=decontext_results[i] + if i < len(decontext_results) + else DecontextualityAnalysis( + proposition=prop, score=0.5, is_standalone=False + ), + minimality=minimal_results[i] + if i < len(minimal_results) + else MinimalityAnalysis(proposition=prop, score=0.5, is_minimal=False), + ) + analysis.compute_molecular_score() + analyses.append(analysis) + + # Build aggregate report + report = MolecularReport( + conversation_id=conversation_id, + peer_name=peer_name, + proposition_count=len(propositions), + source_message_count=len(user_msgs), + analyses=analyses, + ) + + # Compute averages + report.avg_decontextuality = sum( + a.decontextuality.score for a in analyses + ) / len(analyses) + report.avg_minimality = sum(a.minimality.score for a in analyses) / len( + analyses + ) + report.avg_molecular = sum(a.molecular_score for a in analyses) / len(analyses) + + # Count classifications + for a in analyses: + if a.classification == "molecular": + report.molecular_count += 1 + elif a.classification == "too_atomic": + report.too_atomic_count += 1 + elif a.classification == "too_verbose": + report.too_verbose_count += 1 + else: + report.borderline_count += 1 + + # Aggregate issue distributions + for a in analyses: + for amb_type, _, _ in a.ambiguity.ambiguities: + report.ambiguity_distribution[amb_type.value] = ( + report.ambiguity_distribution.get(amb_type.value, 0) + 1 + ) + for issue_type, _ in a.decontextuality.issues: + report.decontextuality_issues[issue_type.value] = ( + report.decontextuality_issues.get(issue_type.value, 0) + 1 + ) + for issue_type, _ in a.minimality.issues: + report.minimality_issues[issue_type.value] = ( + report.minimality_issues.get(issue_type.value, 0) + 1 + ) + + logger.info( + f"Evaluation complete. Molecular: {report.avg_molecular:.2%} " + + f"(D: {report.avg_decontextuality:.2%}, M: {report.avg_minimality:.2%})" + ) + + return report + + +# ============================================================================= +# OUTPUT +# ============================================================================= + + +def print_report(report: MolecularReport) -> None: + """Print formatted molecular report.""" + + print("\n" + "=" * 70) + print(f"MOLECULAR ANALYSIS: {report.conversation_id}") + print("=" * 70) + print( + f"Peer: {report.peer_name} | Propositions: {report.proposition_count} | Messages: {report.source_message_count}" + ) + + print(f"\n{'MOLECULAR SCORE:':<20} {report.avg_molecular:.1%}") + print("-" * 40) + print(f"{'Decontextuality:':<20} {report.avg_decontextuality:.1%}") + print(f"{'Minimality:':<20} {report.avg_minimality:.1%}") + + print("\n" + "-" * 40) + print("CLASSIFICATION DISTRIBUTION:") + total = report.proposition_count + print( + f" ✓ Molecular: {report.molecular_count:3d} ({report.molecular_count/total*100:5.1f}%)" + ) + print( + f" ⚠ Too Atomic: {report.too_atomic_count:3d} ({report.too_atomic_count/total*100:5.1f}%)" + ) + print( + f" ⚠ Too Verbose: {report.too_verbose_count:3d} ({report.too_verbose_count/total*100:5.1f}%)" + ) + print( + f" ~ Borderline: {report.borderline_count:3d} ({report.borderline_count/total*100:5.1f}%)" + ) + + # Show top issues + if report.ambiguity_distribution: + print("\nTop Ambiguity Types:") + for k, v in sorted(report.ambiguity_distribution.items(), key=lambda x: -x[1])[ + :3 + ]: + print(f" - {k}: {v}") + + if report.decontextuality_issues: + print("\nTop Decontextuality Issues:") + for k, v in sorted(report.decontextuality_issues.items(), key=lambda x: -x[1])[ + :3 + ]: + print(f" - {k}: {v}") + + if report.minimality_issues: + print("\nTop Minimality Issues:") + for k, v in sorted(report.minimality_issues.items(), key=lambda x: -x[1])[:3]: + print(f" - {k}: {v}") + + # Show example problems + too_atomic = [a for a in report.analyses if a.classification == "too_atomic"] + if too_atomic: + print("\nExample Too Atomic (needs more context):") + for a in too_atomic[:2]: + print(f' - "{a.proposition[:60]}..."') + if a.decontextuality.suggested_decontextualization: + print( + f' → Suggest: "{a.decontextuality.suggested_decontextualization[:60]}..."' + ) + + too_verbose = [a for a in report.analyses if a.classification == "too_verbose"] + if too_verbose: + print("\nExample Too Verbose (over-elaborated):") + for a in too_verbose[:2]: + print( + f' - "{a.proposition[:60]}..." [{a.minimality.token_count} tokens, {a.minimality.claim_count} claims]' + ) + if a.minimality.suggested_reduction: + print(f' → Suggest: "{a.minimality.suggested_reduction[:60]}..."') + + print("=" * 70) + + +# ============================================================================= +# MAIN +# ============================================================================= + + +async def main(): + parser = argparse.ArgumentParser( + description="MolecularBench - Molecular Facts Evaluation" + ) + parser.add_argument("--traces", type=Path, help="JSON/JSONL trace file") + parser.add_argument("--trace-dir", type=Path, help="Directory of trace files") + parser.add_argument( + "--output-dir", type=Path, default=Path("tests/bench/molecular_results") + ) + parser.add_argument( + "--provider", choices=["anthropic", "openai", "openrouter"], default="anthropic" + ) + parser.add_argument("--api-key", type=str, help="API key") + parser.add_argument("--model", default="claude-sonnet-4-20250514") + parser.add_argument("--verbose", "-v", action="store_true") + parser.add_argument("--limit", type=int, help="Limit number of traces") + parser.add_argument("--sample", type=float, help="Sample fraction (0-1)") + parser.add_argument( + "--batch-size", type=int, default=1, help="Concurrent batch size" + ) + args = parser.parse_args() + + # Initialize client using runner_common helpers + if args.provider == "anthropic": + client: AsyncAnthropic | AsyncOpenAI = create_anthropic_client( + api_key=args.api_key + ) + elif args.provider == "openrouter": + client = create_openai_client( + api_key=args.api_key, + base_url="https://openrouter.ai/api/v1", + env_key_name="OPENROUTER_API_KEY", + ) + else: + client = create_openai_client(api_key=args.api_key) + + judge = MolecularJudge(client, args.model, args.verbose, args.provider) + + # Load traces + all_traces: list[tuple[dict[str, Any], str]] = [] + if args.traces: + traces = load_traces(args.traces) + all_traces.extend((t, args.traces.name) for t in traces) + elif args.trace_dir: + for f in list(args.trace_dir.glob("*.json")) + list( + args.trace_dir.glob("*.jsonl") + ): + traces = load_traces(f) + all_traces.extend((t, f.name) for t in traces) + else: + parser.error("Specify --traces or --trace-dir") + + print(f"Loaded {len(all_traces)} traces") + + if args.sample and 0 < args.sample < 1: + all_traces = random.sample( + all_traces, max(1, int(len(all_traces) * args.sample)) + ) + if args.limit: + all_traces = all_traces[: args.limit] + + print(f"\nEvaluating {len(all_traces)} traces with Molecular Facts criteria...\n") + + overall_start = time.time() + results: list[MolecularReport] = [] + + async def process_trace( + idx: int, trace: dict[str, Any], _source: str + ) -> MolecularReport | None: + props = extract_propositions(trace) + if not props: + logger.warning(f"Trace {idx} has no propositions") + return None + + msgs = extract_messages(trace) + peer = extract_peer_name(trace) + conv_id = extract_conversation_id(trace, idx) + + print(f"[{idx+1}/{len(all_traces)}] {conv_id} ({len(props)} props)...") + + try: + report = await judge.evaluate(props, msgs, peer, conv_id) + print_report(report) + return report + except Exception as e: + logger.error(f"Failed {conv_id}: {e}") + return None + + # Process + if args.batch_size > 1: + for i in range(0, len(all_traces), args.batch_size): + batch = all_traces[i : i + args.batch_size] + batch_results = await asyncio.gather( + *[process_trace(i + j, t, s) for j, (t, s) in enumerate(batch)], + return_exceptions=True, + ) + for r in batch_results: + if isinstance(r, MolecularReport): + results.append(r) + else: + for idx, (trace, source) in enumerate(all_traces): + r = await process_trace(idx, trace, source) + if r: + results.append(r) + + total_duration = time.time() - overall_start + + # Save results + if results: + args.output_dir.mkdir(parents=True, exist_ok=True) + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + out_file = args.output_dir / f"molecular_{ts}.json" + + agg: dict[str, Any] = { + "timestamp": ts, + "model": args.model, + "count": len(results), + "duration": format_duration(total_duration), + "averages": { + "molecular": sum(r.avg_molecular for r in results) / len(results), + "decontextuality": sum(r.avg_decontextuality for r in results) + / len(results), + "minimality": sum(r.avg_minimality for r in results) / len(results), + }, + "classification_totals": { + "molecular": sum(r.molecular_count for r in results), + "too_atomic": sum(r.too_atomic_count for r in results), + "too_verbose": sum(r.too_verbose_count for r in results), + "borderline": sum(r.borderline_count for r in results), + }, + "results": [r.to_dict() for r in results], + } + + with open(out_file, "w") as f: + json.dump(agg, f, indent=2) + + print(f"\nSaved to {out_file}") + + if len(results) > 1: + print("\n" + "=" * 70) + print("AGGREGATE RESULTS") + print("=" * 70) + print( + f"Traces: {len(results)} | Duration: {format_duration(total_duration)}" + ) + print(f"\nMolecular Score: {agg['averages']['molecular']:.1%}") + print(f"Decontextuality: {agg['averages']['decontextuality']:.1%}") + print(f"Minimality: {agg['averages']['minimality']:.1%}") + print("\nClassification:") + total_props = sum(r.proposition_count for r in results) + for k, v in agg["classification_totals"].items(): + print(f" {k:<15} {v:4d} ({v/total_props*100:5.1f}%)") + else: + print(f"Duration: {format_duration(total_duration)}") + else: + print("\nNo traces evaluated") + return 1 + + return 0 + + +if __name__ == "__main__": + exit(asyncio.run(main())) diff --git a/tests/bench/obex.py b/tests/bench/obex.py deleted file mode 100644 index abd784e8..00000000 --- a/tests/bench/obex.py +++ /dev/null @@ -1,643 +0,0 @@ -""" -Deriver Observation Extraction Benchmark. - -Evaluates the deriver's ability to extract observations from messages by comparing -extracted observations against ground truth using embedding similarity. - -Test data sources (915 total cases): -- LoCoMo: 543 cases with human-curated observations -- LongMem: 170 cases with Sonnet-generated observations -- BEAM: 202 cases with Sonnet-generated observations - -Usage: - uv run python tests/bench/obex.py # Run full evaluation - uv run python tests/bench/obex.py --limit 10 # Test with limited cases - uv run python tests/bench/obex.py --source locomo # Evaluate specific source - uv run python tests/bench/obex.py --threshold 0.80 # Custom similarity threshold -""" - -import argparse -import asyncio -import json -import sys -import time -from dataclasses import asdict, dataclass, field -from datetime import datetime -from pathlib import Path -from typing import Any - -import numpy as np - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - -from src.config import settings -from src.deriver.prompts import minimal_deriver_prompt -from src.embedding_client import EmbeddingClient -from src.utils.clients import honcho_llm_call -from src.utils.representation import PromptRepresentation - -CANDIDATES_DIR = Path(__file__).parent / "obexeval_data" / "candidates" -EVAL_RESULTS_DIR = Path(__file__).parent / "eval_results" - - -# ============================================================================= -# Data Classes -# ============================================================================= - - -def f1_score(precision: float, recall: float) -> float: - """Calculate F1 score from precision and recall.""" - if precision + recall == 0: - return 0.0 - return 2 * (precision * recall) / (precision + recall) - - -@dataclass -class CaseResult: - """Result for a single test case.""" - - case_id: str - source: str - difficulty: str - explicit_precision: float - explicit_recall: float - # deductive_precision: float - # deductive_recall: float - num_extracted: int - num_expected: int - # Actual model output - extracted_explicit: list[str] = field(default_factory=list) - # extracted_deductive: list[str] = field(default_factory=list) - error: str | None = None - - @property - def explicit_f1(self) -> float: - return f1_score(self.explicit_precision, self.explicit_recall) - - # @property - # def deductive_f1(self) -> float: - # return f1_score(self.deductive_precision, self.deductive_recall) - - -@dataclass -class SourceStats: - """Statistics for a source or difficulty grouping.""" - - count: int - explicit_precision: float - explicit_recall: float - explicit_f1: float - # deductive_precision: float - # deductive_recall: float - # deductive_f1: float - - -@dataclass -class AggregateResults: - """Aggregate results across all cases.""" - - total_cases: int - explicit_precision: float - explicit_recall: float - explicit_f1: float - # deductive_precision: float - # deductive_recall: float - # deductive_f1: float - by_source: dict[str, SourceStats] = field(default_factory=dict) - by_difficulty: dict[str, SourceStats] = field(default_factory=dict) - - -# ============================================================================= -# Data Loading -# ============================================================================= - - -TestCase = dict[str, Any] - - -def load_test_cases(source: str | None = None) -> list[TestCase]: - """Load test cases from candidate files.""" - all_cases: list[TestCase] = [] - for json_file in CANDIDATES_DIR.glob("*_candidates.json"): - with open(json_file) as f: - data: dict[str, Any] = json.load(f) - all_cases.extend(data.get("candidates", [])) - - if source: - all_cases = [tc for tc in all_cases if tc.get("source") == source] - - return all_cases - - -# ============================================================================= -# Deriver Execution -# ============================================================================= - - -def format_messages(messages: list[dict[str, Any]]) -> str: - """Format messages for the deriver prompt.""" - formatted: list[str] = [] - for msg in messages: - author: str = msg.get("author", "unknown") - content: str = msg.get("content", "") - timestamp: str = msg.get("timestamp", "") - - if timestamp: - try: - dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) - time_str = dt.strftime("%Y-%m-%d %H:%M") - formatted.append(f"[{time_str}] {author}: {content}") - except ValueError: - formatted.append(f"{author}: {content}") - else: - formatted.append(f"{author}: {content}") - - return "\n".join(formatted) - - -async def run_deriver( - messages: list[dict[str, Any]], target_peer: str -) -> PromptRepresentation: - """Run the deriver on messages and return extracted observations.""" - formatted = format_messages(messages) - prompt = minimal_deriver_prompt( - peer_id=target_peer, - messages=formatted, - ) - - response = await honcho_llm_call( - llm_settings=settings.DERIVER, - prompt=prompt, - max_tokens=settings.DERIVER.MAX_OUTPUT_TOKENS or 2000, - track_name="Deriver Eval", - response_model=PromptRepresentation, - json_mode=True, - stop_seqs=[" \n", "\n\n\n\n"], - enable_retry=True, - retry_attempts=3, - ) - - return response.content - - -# ============================================================================= -# Embedding-based Matching -# ============================================================================= - - -class ObservationMatcher: - """Matches observations using embedding similarity.""" - - threshold: float - client: EmbeddingClient - - def __init__(self, threshold: float = 0.85): - self.threshold = threshold - self.client = EmbeddingClient() - self._cache: dict[str, list[float]] = {} - - async def _get_embeddings(self, texts: list[str]) -> list[list[float]]: - """Get embeddings with caching.""" - uncached = [t for t in texts if t not in self._cache] - if uncached: - embeddings = await self.client.simple_batch_embed(uncached) - for text, emb in zip(uncached, embeddings, strict=True): - self._cache[text] = emb - return [self._cache[t] for t in texts] - - def _cosine_sim(self, a: list[float], b: list[float]) -> float: - """Compute cosine similarity.""" - a_np, b_np = np.array(a), np.array(b) - return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np))) - - async def match( - self, extracted: list[str], expected: list[str] - ) -> tuple[float, float]: - """ - Match extracted to expected using greedy similarity matching. - Returns: (precision, recall) - """ - if not extracted and not expected: - return 1.0, 1.0 - if not extracted: - return 0.0, 0.0 - if not expected: - return 0.0, 1.0 - - # Get embeddings - all_texts = extracted + expected - all_embs = await self._get_embeddings(all_texts) - ext_embs = all_embs[: len(extracted)] - exp_embs = all_embs[len(extracted) :] - - # Build similarity matrix - sim_matrix = np.zeros((len(extracted), len(expected))) - for i, e1 in enumerate(ext_embs): - for j, e2 in enumerate(exp_embs): - sim_matrix[i, j] = self._cosine_sim(e1, e2) - - # Greedy matching - matches = 0 - used_ext: set[int] = set() - used_exp: set[int] = set() - pairs = sorted( - [ - (sim_matrix[i, j], i, j) - for i in range(len(extracted)) - for j in range(len(expected)) - ], - reverse=True, - ) - - for sim, i, j in pairs: - if sim < self.threshold: - break - if i in used_ext or j in used_exp: - continue - matches += 1 - used_ext.add(i) - used_exp.add(j) - - precision = matches / len(extracted) - recall = matches / len(expected) - return precision, recall - - -# ============================================================================= -# Evaluation -# ============================================================================= - - -async def evaluate_case(case: TestCase, matcher: ObservationMatcher) -> CaseResult: - """Evaluate a single test case.""" - case_id: str = case["id"] - source: str = case.get("source", "unknown") - difficulty: str = case.get("difficulty", "unknown") - messages: list[dict[str, Any]] = case.get("messages", []) - target_peer: str = case.get("target_peer", "user") - expected: dict[str, Any] = case.get("expected_observations", {}) - - exp_explicit: list[str] = [o["content"] for o in expected.get("explicit", [])] - # exp_deductive: list[str] = [o["conclusion"] for o in expected.get("deductive", [])] - - ext_explicit: list[str] - # ext_deductive: list[str] - error: str | None - try: - result = await run_deriver(messages, target_peer) - ext_explicit = [o.content for o in result.explicit] - # ext_deductive = [o.conclusion for o in result.deductive] - error = None - except Exception as e: - ext_explicit = [] # , ext_deductive = [], [] - error = str(e) - - # Match explicit - exp_p, exp_r = await matcher.match(ext_explicit, exp_explicit) - - # Match deductive - # ded_p, ded_r = await matcher.match(ext_deductive, exp_deductive) - - return CaseResult( - case_id=case_id, - source=source, - difficulty=difficulty, - explicit_precision=exp_p, - explicit_recall=exp_r, - # deductive_precision=ded_p, - # deductive_recall=ded_r, - num_extracted=len(ext_explicit), # + len(ext_deductive), - num_expected=len(exp_explicit), # + len(exp_deductive), - extracted_explicit=ext_explicit, - # extracted_deductive=ext_deductive, - error=error, - ) - - -def aggregate_results(results: list[CaseResult]) -> AggregateResults: - """Compute aggregate statistics.""" - if not results: - return AggregateResults(0, 0, 0, 0) # , 0, 0, 0) - - n = len(results) - exp_p = sum(r.explicit_precision for r in results) / n - exp_r = sum(r.explicit_recall for r in results) / n - # ded_p = sum(r.deductive_precision for r in results) / n - # ded_r = sum(r.deductive_recall for r in results) / n - agg = AggregateResults( - total_cases=n, - explicit_precision=exp_p, - explicit_recall=exp_r, - explicit_f1=f1_score(exp_p, exp_r), - # deductive_precision=ded_p, - # deductive_recall=ded_r, - # deductive_f1=f1_score(ded_p, ded_r), - ) - - # By source - by_source: dict[str, list[CaseResult]] = {} - for r in results: - by_source.setdefault(r.source, []).append(r) - for source, rs in by_source.items(): - src_exp_p = sum(r.explicit_precision for r in rs) / len(rs) - src_exp_r = sum(r.explicit_recall for r in rs) / len(rs) - # src_ded_p = sum(r.deductive_precision for r in rs) / len(rs) - # src_ded_r = sum(r.deductive_recall for r in rs) / len(rs) - agg.by_source[source] = SourceStats( - count=len(rs), - explicit_precision=src_exp_p, - explicit_recall=src_exp_r, - explicit_f1=f1_score(src_exp_p, src_exp_r), - # deductive_precision=src_ded_p, - # deductive_recall=src_ded_r, - # deductive_f1=f1_score(src_ded_p, src_ded_r), - ) - - # By difficulty - by_diff: dict[str, list[CaseResult]] = {} - for r in results: - by_diff.setdefault(r.difficulty, []).append(r) - for diff, rs in by_diff.items(): - diff_exp_p = sum(r.explicit_precision for r in rs) / len(rs) - diff_exp_r = sum(r.explicit_recall for r in rs) / len(rs) - # diff_ded_p = sum(r.deductive_precision for r in rs) / len(rs) - # diff_ded_r = sum(r.deductive_recall for r in rs) / len(rs) - agg.by_difficulty[diff] = SourceStats( - count=len(rs), - explicit_precision=diff_exp_p, - explicit_recall=diff_exp_r, - explicit_f1=f1_score(diff_exp_p, diff_exp_r), - # deductive_precision=diff_ded_p, - # deductive_recall=diff_ded_r, - # deductive_f1=f1_score(diff_ded_p, diff_ded_r), - ) - - return agg - - -def print_results(agg: AggregateResults, results: list[CaseResult]) -> None: - """Print evaluation results.""" - print("\n" + "=" * 70) - print("DERIVER OBSERVATION EXTRACTION EVALUATION") - print("=" * 70) - - print(f"\nTotal test cases: {agg.total_cases}") - - print(f"\n{'EXPLICIT OBSERVATIONS':^35}") - print(f" Precision: {agg.explicit_precision:.3f}") - print(f" Recall: {agg.explicit_recall:.3f}") - print(f" F1: {agg.explicit_f1:.3f}") - - # print(f"\n{'DEDUCTIVE OBSERVATIONS':^35}") - # print(f" Precision: {agg.deductive_precision:.3f}") - # print(f" Recall: {agg.deductive_recall:.3f}") - # print(f" F1: {agg.deductive_f1:.3f}") - - if agg.by_source: - print(f"\n{'BY SOURCE':^35}") - for source, data in sorted(agg.by_source.items()): - print( - f" {source:12} n={data.count:3} exp_f1={data.explicit_f1:.3f}" # ded_f1={data.deductive_f1:.3f}" - ) - - if agg.by_difficulty: - print(f"\n{'BY DIFFICULTY':^35}") - for diff, data in sorted(agg.by_difficulty.items()): - print( - f" {diff:12} n={data.count:3} exp_f1={data.explicit_f1:.3f}" # ded_f1={data.deductive_f1:.3f}" - ) - - # Worst cases (by average of precision and recall) - worst = sorted( - results, - key=lambda r: ( - r.explicit_precision + r.explicit_recall - # + r.deductive_precision - # + r.deductive_recall - ) - / 2, # / 4, - )[:5] - print(f"\n{'WORST PERFORMING CASES':^35}") - for r in worst: - avg = ( - ( - r.explicit_precision + r.explicit_recall - # + r.deductive_precision - # + r.deductive_recall - ) - / 2, - ) # / 4, - print(f" {r.case_id[:45]:45} avg={avg:.3f}") - - # Errors - errors = [r for r in results if r.error] - if errors: - print(f"\nErrors: {len(errors)} cases failed") - - -def generate_json_summary( - results: list[CaseResult], - agg: AggregateResults, - source_filter: str | None, - threshold: float, - concurrency: int, - total_elapsed_seconds: float, - output_file: Path, -) -> None: - """ - Generate a comprehensive JSON summary of test results for analytics. - - Args: - results: List of case results - agg: Aggregate results - source_filter: Source filter used (if any) - threshold: Similarity threshold used - concurrency: Concurrency level used - total_elapsed_seconds: Total elapsed time for all tests - output_file: Path to write JSON output to - """ - errors = [r for r in results if r.error] - - # Calculate per-case average scores - case_scores: list[float] = [] - for r in results: - avg = (r.explicit_precision + r.explicit_recall) / 2 - # + r.deductive_precision - # + r.deductive_recall - # / 4, - case_scores.append(avg) - - output_data: dict[str, Any] = { - "metadata": { - "benchmark": "obex", - "description": "Observation Extraction Benchmark - evaluates deriver's ability to extract observations from messages", - "execution_timestamp": datetime.now().isoformat(), - "runner_version": "1.0.0", - "deriver_settings": settings.DERIVER.model_dump(), - }, - "config": { - "source_filter": source_filter, - "threshold": threshold, - "concurrency": concurrency, - "total_cases": len(results), - }, - "timing": { - "total_elapsed_seconds": total_elapsed_seconds, - "average_seconds_per_case": total_elapsed_seconds / len(results) - if results - else 0, - }, - "summary_statistics": { - "total_cases": agg.total_cases, - "cases_with_errors": len(errors), - "explicit_precision": agg.explicit_precision, - "explicit_recall": agg.explicit_recall, - "explicit_f1": agg.explicit_f1, - # "deductive_precision": agg.deductive_precision, - # "deductive_recall": agg.deductive_recall, - # "deductive_f1": agg.deductive_f1, - "mean_case_score": sum(case_scores) / len(case_scores) - if case_scores - else 0, - "min_case_score": min(case_scores) if case_scores else 0, - "max_case_score": max(case_scores) if case_scores else 0, - }, - "statistics_by_source": {k: asdict(v) for k, v in agg.by_source.items()}, - "statistics_by_difficulty": { - k: asdict(v) for k, v in agg.by_difficulty.items() - }, - "detailed_results": [ - { - "case_id": r.case_id, - "source": r.source, - "difficulty": r.difficulty, - "explicit_precision": r.explicit_precision, - "explicit_recall": r.explicit_recall, - "explicit_f1": r.explicit_f1, - # "deductive_precision": r.deductive_precision, - # "deductive_recall": r.deductive_recall, - # "deductive_f1": r.deductive_f1, - "num_extracted": r.num_extracted, - "num_expected": r.num_expected, - "average_f1": r.explicit_f1, # (r.explicit_f1 + r.deductive_f1) / 2, - "model_output": { - "explicit": r.extracted_explicit, - # "deductive": r.extracted_deductive, - }, - "error": r.error, - } - for r in results - ], - } - - # Ensure output directory exists - output_file.parent.mkdir(parents=True, exist_ok=True) - - with open(output_file, "w") as f: - json.dump(output_data, f, indent=2) - print(f"\nResults saved to {output_file}") - - -async def run_evaluation( - source: str | None = None, - limit: int | None = None, - threshold: float = 0.85, - output: str | None = None, - concurrency: int = 10, -) -> None: - """Run the evaluation pipeline.""" - start_time = time.time() - - print("Loading test cases...") - cases = load_test_cases(source) - - if limit: - cases = cases[:limit] - - if not cases: - print("No test cases found!") - return - - print( - f"Evaluating {len(cases)} cases (model={settings.DERIVER.MODEL}, threshold={threshold}, concurrency={concurrency})..." - ) - - matcher = ObservationMatcher(threshold=threshold) - completed = 0 - semaphore = asyncio.Semaphore(concurrency) - - async def evaluate_with_semaphore(case: TestCase) -> CaseResult: - nonlocal completed - async with semaphore: - result = await evaluate_case(case, matcher) - completed += 1 - if completed % 10 == 0 or completed == 1: - print(f" Completed {completed}/{len(cases)}...") - return result - - # Run all evaluations concurrently with semaphore limiting - tasks = [evaluate_with_semaphore(case) for case in cases] - results = await asyncio.gather(*tasks) - - end_time = time.time() - total_elapsed = end_time - start_time - - agg = aggregate_results(list(results)) - print_results(agg, list(results)) - - print( - f"\nTotal time: {total_elapsed:.1f}s ({total_elapsed / len(cases):.2f}s per case)" - ) - - # Determine output file path - if output: - output_file = Path(output) - else: - # Generate default timestamped filename - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - source_suffix = f"_{source}" if source else "" - output_file = EVAL_RESULTS_DIR / f"obex{source_suffix}_{timestamp}.json" - - # Always save results - generate_json_summary( - list(results), - agg, - source, - threshold, - concurrency, - total_elapsed, - output_file, - ) - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Deriver observation extraction benchmark" - ) - parser.add_argument( - "--source", type=str, help="Filter by source (locomo, longmem, beam)" - ) - parser.add_argument("--limit", type=int, help="Limit number of test cases") - parser.add_argument( - "--threshold", - type=float, - default=0.85, - help="Similarity threshold (default: 0.85)", - ) - parser.add_argument("--output", type=str, help="Output path for results JSON") - parser.add_argument( - "--concurrency", - type=int, - default=10, - help="Number of concurrent LLM calls (default: 10)", - ) - args = parser.parse_args() - asyncio.run( - run_evaluation( - args.source, args.limit, args.threshold, args.output, args.concurrency - ) - ) - - -if __name__ == "__main__": - main() diff --git a/tests/bench/runner_common.py b/tests/bench/runner_common.py index bd1826a2..093027df 100644 --- a/tests/bench/runner_common.py +++ b/tests/bench/runner_common.py @@ -1,12 +1,14 @@ """ -Shared utilities for Honcho benchmark test runners. +Shared utilities for Honcho benchmark test runners and evaluators. Contains the BaseRunner abstract class and RunnerConfig dataclass that provide -a common framework for all benchmark runners (longmem, beam, locomo). +a common framework for all benchmark runners (longmem, beam, locomo), as well +as shared utilities for trace-based evaluators (molecular, coverage). """ import argparse import asyncio +import json import logging import os import time @@ -15,7 +17,7 @@ from dataclasses import dataclass, field from datetime import datetime from logging import Logger from pathlib import Path -from typing import Any, Generic, Literal, TypeVar +from typing import Any, Generic, Literal, TypeVar, cast from anthropic import AsyncAnthropic from honcho import Honcho @@ -24,6 +26,8 @@ from openai import AsyncOpenAI from src.telemetry.metrics_collector import MetricsCollector +_logger = logging.getLogger(__name__) + # Valid reasoning levels for dialectic chat ReasoningLevel = Literal["minimal", "low", "medium", "high", "max"] REASONING_LEVELS: list[str] = ["minimal", "low", "medium", "high", "max"] @@ -195,22 +199,27 @@ def validate_common_arguments(args: argparse.Namespace) -> str | None: return None -def configure_logging() -> logging.Logger: +def configure_logging( + level: int = logging.WARNING, + name: str | None = None, +) -> logging.Logger: """ Configure logging for benchmark runners. - Sets up logging with WARNING level and suppresses HTTP request logs. + Sets up logging at the specified level and suppresses HTTP request logs. + + Args: + level: Logging level (default: WARNING for runners, use INFO for evaluators) + name: Logger name (default: caller's __name__) Returns: - Logger instance for the calling module + Logger instance """ - logging.basicConfig( - level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(message)s" - ) + logging.basicConfig(level=level, format="%(asctime)s - %(levelname)s - %(message)s") # Suppress HTTP request logs from the Honcho SDK logging.getLogger("httpx").setLevel(logging.ERROR) logging.getLogger("httpcore").setLevel(logging.ERROR) - return logging.getLogger(__name__) + return logging.getLogger(name or __name__) def create_anthropic_client(api_key: str | None = None) -> AsyncAnthropic: @@ -277,6 +286,127 @@ def format_duration(seconds: float) -> str: return f"{hours}h {minutes}m" +# --------------------------------------------------------------------------- +# Trace parsing utilities (shared by trace-based evaluators) +# --------------------------------------------------------------------------- + + +def load_traces(path: Path) -> list[dict[str, Any]]: + """Load traces from JSON or JSONL file. + + Attempts JSONL (one object per line) first; falls back to standard JSON + (array or single object). Returns an empty list when the file cannot be + read or parsed. + """ + traces: list[dict[str, Any]] = [] + + # --- Attempt 1: try JSONL (one JSON object per line) --- + try: + with open(path) as f: + first_line = f.readline().strip() + if first_line and not first_line.startswith("["): + f.seek(0) + for line_no, line in enumerate(f, 1): + line = line.strip() + if not line: + continue + try: + traces.append(cast(dict[str, Any], json.loads(line))) + except json.JSONDecodeError: + _logger.debug( + "%s:%d: skipping malformed JSON line", path, line_no + ) + continue + if traces: + return traces + except json.JSONDecodeError: + _logger.warning("Failed to parse %s as JSONL", path) + except OSError: + _logger.exception("Could not read trace file %s", path) + return [] + + # --- Attempt 2: try standard JSON (array or single object) --- + try: + with open(path) as f: + data = cast(dict[str, Any] | list[dict[str, Any]], json.load(f)) + except json.JSONDecodeError: + _logger.exception("Failed to parse %s as JSON", path) + return [] + except OSError: + _logger.exception("Could not read trace file %s", path) + return [] + + if isinstance(data, list): + return data + return [data] + + +def extract_propositions(trace: dict[str, Any]) -> list[str]: + """Extract propositions from trace output.""" + output = trace.get("output") + if not isinstance(output, dict): + return [] + output_d = cast(dict[str, Any], output) + content = output_d.get("content") + if not isinstance(content, dict): + return [] + content_d = cast(dict[str, Any], content) + explicit = content_d.get("explicit") + if not isinstance(explicit, list): + return [] + results: list[str] = [] + for raw_item in cast(list[Any], explicit): + item = cast(dict[str, Any], raw_item) + if isinstance(raw_item, dict) and "content" in item: + results.append(str(item["content"])) + return results + + +def extract_messages(trace: dict[str, Any]) -> list[dict[str, Any]]: + """Extract source messages from trace input.""" + input_data = trace.get("input") + if not isinstance(input_data, dict): + return [] + input_d = cast(dict[str, Any], input_data) + prompt = input_d.get("prompt", "") + if not isinstance(prompt, str): + return [] + + messages: list[dict[str, Any]] = [] + if "" in prompt: + section = prompt.split("")[1].split("")[0] + for line in section.strip().split("\n"): + line = line.strip() + if not line: + continue + parts = line.split(" ", 3) + if len(parts) >= 3: + speaker = parts[2].rstrip(":") + text = parts[3] if len(parts) > 3 else "" + if "->->" in text: + text = text.split("->->")[0].strip() + messages.append({"speaker": speaker, "text": text}) + return messages + + +def extract_peer_name(trace: dict[str, Any]) -> str: + """Extract peer name from propositions.""" + for prop in extract_propositions(trace): + prop_lower = prop.lower() + for pattern in ["name is", "is named", "called"]: + if pattern in prop_lower: + parts = prop_lower.split(pattern) + if len(parts) > 1: + name = parts[1].strip().rstrip(".").split()[0] + return name.capitalize() + return "user" + + +def extract_conversation_id(trace: dict[str, Any], index: int) -> str: + """Extract or generate conversation ID.""" + return trace.get("conversation_id", trace.get("id", f"trace_{index:04d}")) + + class BaseRunner(ABC, Generic[ResultT]): """ Abstract base class for benchmark runners.