From 10ef7b96a81d4ae3a04ddef586a5b9ab5951d6c8 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Wed, 25 Feb 2026 15:08:09 -0500 Subject: [PATCH] Add Stricter limits to Summary & Peer Card (#400) * fix: Add bounds to gemini client * fix: Prevent empty summaries from being saved to DB (HONCHO-M7) Raise LLMError on blocked Gemini responses (SAFETY, RECITATION, etc.) so retry/backup-provider logic triggers. Treat empty LLM responses in the summarizer as fallback instead of persisting empty strings. Co-Authored-By: Claude Opus 4.6 * feat: Summary Eval via Locomo * fix: Code Rabbit Comments * fix: Code Rabbit Comments --------- Co-authored-by: Claude Opus 4.6 --- src/dreamer/specialists.py | 22 +- src/utils/agent_tools.py | 75 +++- src/utils/clients.py | 58 ++- src/utils/summarizer.py | 18 +- tests/bench/coverage.py | 645 +++++++++++++++++++------------- tests/bench/locomo_summary.py | 577 ++++++++++++++++++++++++++++ tests/utils/test_agent_tools.py | 83 ++++ tests/utils/test_clients.py | 145 ++++++- tests/utils/test_summarizer.py | 219 +++++++++++ 9 files changed, 1539 insertions(+), 303 deletions(-) create mode 100644 tests/bench/locomo_summary.py create mode 100644 tests/utils/test_summarizer.py diff --git a/src/dreamer/specialists.py b/src/dreamer/specialists.py index cb690420..70d51c53 100644 --- a/src/dreamer/specialists.py +++ b/src/dreamer/specialists.py @@ -61,7 +61,9 @@ class BaseSpecialist(ABC): name: str = "base" # Subclasses can override to customize the peer card update instruction - peer_card_update_instruction: str = "Update this with `update_peer_card` if needed." + peer_card_update_instruction: str = ( + "Only update this with durable profile facts via `update_peer_card`." + ) @abstractmethod def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]: @@ -108,6 +110,7 @@ class BaseSpecialist(ABC): {facts} {self.peer_card_update_instruction} +If you update it, send the full deduplicated list and remove stale entries. """ @@ -289,7 +292,7 @@ class DeductionSpecialist(BaseSpecialist): """ name: str = "deduction" - peer_card_update_instruction: str = "Update this with `update_peer_card` if you discover new biographical information." + peer_card_update_instruction: str = "Update this with `update_peer_card` only for stable biographical/profile facts." def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]: if peer_card_enabled: @@ -324,13 +327,16 @@ The peer card is a summary of stable biographical facts. You MUST update it when - Standing instructions ("call me X", "don't mention Y") - Core preferences and traits +Never add temporary event summaries, one-off conclusions, reasoning traces, or contradiction notes. + Format entries as: - Plain facts: "Name: Alice", "Works at Google", "Lives in NYC" - `INSTRUCTION: ...` for standing instructions - `PREFERENCE: ...` for preferences - `TRAIT: ...` for personality traits -Call `update_peer_card` with the complete updated list when you have new biographical info.""" +Call `update_peer_card` with the complete updated list when you have new biographical info. +Keep it concise (max 40 entries), deduplicated, and current.""" return f"""You are a deductive reasoning agent analyzing observations about {observed}. @@ -429,9 +435,7 @@ class InductionSpecialist(BaseSpecialist): """ name: str = "induction" - peer_card_update_instruction: str = ( - "Update this with `update_peer_card` if you identify new patterns or traits." - ) + peer_card_update_instruction: str = "Only add highly stable profile traits/preferences; do not copy transient conclusions." def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]: if peer_card_enabled: @@ -460,12 +464,14 @@ class InductionSpecialist(BaseSpecialist): ## PEER CARD (REQUIRED) -After identifying patterns, update the peer card with high-confidence traits and tendencies: +After identifying patterns, only update the peer card for durable profile-level traits/preferences: - `TRAIT: Analytical thinker` - `TRAIT: Tends to reschedule when stressed` - `PREFERENCE: Prefers detailed explanations` -Call `update_peer_card` with the complete list when you identify new patterns.""" +Do NOT add temporary patterns, episode-specific conclusions, or reasoning summaries. +Call `update_peer_card` with the complete deduplicated list only when a durable profile update is warranted. +Keep it concise (max 40 entries).""" return f"""You are an inductive reasoning agent identifying patterns about {observed}. diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index b93cdecc..e4b38255 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -27,6 +27,9 @@ from src.utils.types import get_current_iteration logger = logging.getLogger(__name__) +# Hard cap to prevent unbounded peer card growth from repeated agent updates. +MAX_PEER_CARD_FACTS = 40 + def _safe_int(value: Any, default: int) -> int: """Coerce a tool input value to int, returning default on failure. @@ -247,13 +250,20 @@ TOOLS: dict[str, dict[str, Any]] = { }, "update_peer_card": { "name": "update_peer_card", - "description": "Update the peer card with facts about the observed peer. The peer card is a summary of key information about the peer.", + "description": ( + "Update the peer card with durable profile facts about the observed peer. " + + "Only include stable biographical facts, standing instructions, and long-lived preferences/traits. " + + "Do not include one-off conclusions, temporary events, or duplicate entries." + ), "input_schema": { "type": "object", "properties": { "content": { "type": "array", - "description": "List of facts about the peer", + "description": ( + "Complete deduplicated peer card list (max 40 entries). " + + "Each entry should be a concise standalone profile fact." + ), "items": {"type": "string"}, }, }, @@ -1106,12 +1116,58 @@ async def _handle_update_peer_card(ctx: ToolContext, tool_input: dict[str, Any]) "Peer card creation is disabled for this workspace/session configuration." ) - peer_card_content = tool_input["content"] + raw_peer_card_content = tool_input.get("content") + + # Guard against None or empty content — keep the existing peer card. + if raw_peer_card_content is None: + logger.warning( + "Peer card update called with None content for %s, keeping existing card", + ctx.workspace_name, + ) + return "Peer card content was empty, no update performed." + + # Normalize and deduplicate to keep peer cards bounded and stable. + normalized_peer_card: list[str] = [] + seen: set[str] = set() + items = ( + cast(list[str], raw_peer_card_content) + if isinstance(raw_peer_card_content, list) + else [str(raw_peer_card_content)] + ) + for item in items: + line = str(item).strip() + if not line: + continue + + # Case-insensitive dedupe with whitespace normalization. + normalized_key = " ".join(line.lower().split()) + if normalized_key in seen: + continue + seen.add(normalized_key) + normalized_peer_card.append(line) + + # Don't clear the peer card if all content normalized to empty. + if not normalized_peer_card: + logger.warning( + "Peer card update normalized to empty for %s, keeping existing card", + ctx.workspace_name, + ) + return "Peer card content was empty after normalization, no update performed." + + if len(normalized_peer_card) > MAX_PEER_CARD_FACTS: + logger.warning( + "Peer card update exceeded max facts (%s), truncating from %s to %s", + MAX_PEER_CARD_FACTS, + len(normalized_peer_card), + MAX_PEER_CARD_FACTS, + ) + normalized_peer_card = normalized_peer_card[:MAX_PEER_CARD_FACTS] + async with ctx.db_lock: await crud.set_peer_card( ctx.db, workspace_name=ctx.workspace_name, - peer_card=peer_card_content, + peer_card=normalized_peer_card, observer=ctx.observer, observed=ctx.observed, ) @@ -1122,15 +1178,6 @@ async def _handle_update_peer_card(ctx: ToolContext, tool_input: dict[str, Any]) # Emit telemetry event if context is available if ctx.run_id and ctx.agent_type and ctx.parent_category: # Count facts in peer card (content is a list of strings per tool schema) - facts_count: int - if isinstance(peer_card_content, list): - content_list = cast(list[str], peer_card_content) - facts_count = len([line for line in content_list if line.strip()]) - else: - # Fallback for string (defensive) - facts_count = len( - [line for line in str(peer_card_content).split("\n") if line.strip()] - ) emit( AgentToolPeerCardUpdatedEvent( run_id=ctx.run_id, @@ -1140,7 +1187,7 @@ async def _handle_update_peer_card(ctx: ToolContext, tool_input: dict[str, Any]) workspace_name=ctx.workspace_name, observer=ctx.observer, observed=ctx.observed, - facts_count=facts_count, + facts_count=len(normalized_peer_card), ) ) diff --git a/src/utils/clients.py b/src/utils/clients.py index 99eefe2e..1c042bff 100644 --- a/src/utils/clients.py +++ b/src/utils/clients.py @@ -23,6 +23,7 @@ from sentry_sdk.ai.monitoring import ai_track from tenacity import retry, stop_after_attempt, wait_exponential from src.config import LLMComponentSettings, settings +from src.exceptions import LLMError from src.telemetry.logging import conditional_observe from src.telemetry.reasoning_traces import log_reasoning_trace from src.utils.json_parser import validate_and_repair_json @@ -32,6 +33,16 @@ from src.utils.types import SupportedProviders, set_current_iteration logger = logging.getLogger(__name__) +# Gemini finish reasons that indicate the response was blocked by safety or policy +# filters. When these occur, the response typically has no usable text content and +# retrying with a backup provider is appropriate. +GEMINI_BLOCKED_FINISH_REASONS = { + "SAFETY", + "RECITATION", + "PROHIBITED_CONTENT", + "BLOCKLIST", +} + @dataclass class IterationData: @@ -2062,6 +2073,9 @@ async def honcho_llm_call_inner( # Build config for Gemini gemini_config: dict[str, Any] = {} + # Gemini uses max_output_tokens, not max_tokens. + gemini_config["max_output_tokens"] = params["max_tokens"] + if temperature is not None: gemini_config["temperature"] = temperature @@ -2199,6 +2213,19 @@ async def honcho_llm_call_inner( else "stop" ) + # Raise on blocked responses so retry/backup-provider logic kicks in + if ( + not text_content + and not gemini_tool_calls + and finish_reason in GEMINI_BLOCKED_FINISH_REASONS + ): + raise LLMError( + f"Gemini response blocked (finish_reason={finish_reason})", + provider="google", + model=model, + finish_reason=finish_reason, + ) + return HonchoLLMCallResponse( content=text_content, input_tokens=input_token_count, @@ -2234,6 +2261,18 @@ async def honcho_llm_call_inner( else "stop" ) + # Raise on blocked responses before checking parsed content + if ( + not gemini_response.parsed + and finish_reason in GEMINI_BLOCKED_FINISH_REASONS + ): + raise LLMError( + f"Gemini response blocked (finish_reason={finish_reason})", + provider="google", + model=model, + finish_reason=finish_reason, + ) + # Validate that parsed content matches the response model if not isinstance(gemini_response.parsed, response_model): raise ValueError( @@ -2449,23 +2488,25 @@ async def handle_streaming_response( case genai.Client(): prompt_text = params["messages"][0]["content"] if params["messages"] else "" + stream_config: GenerateContentConfigDict = { + "max_output_tokens": cast(int, params["max_tokens"]), + } if response_model is not None: + stream_config["response_mime_type"] = "application/json" + stream_config["response_schema"] = response_model response_stream = await client.aio.models.generate_content_stream( model=params["model"], contents=prompt_text, - config={ - "response_mime_type": "application/json", - "response_schema": response_model, - }, + config=stream_config, ) else: + if json_mode: + stream_config["response_mime_type"] = "application/json" response_stream = await client.aio.models.generate_content_stream( model=params["model"], contents=prompt_text, - config={ - "response_mime_type": "application/json" if json_mode else None, - }, + config=stream_config, ) final_chunk = None @@ -2474,6 +2515,9 @@ async def handle_streaming_response( yield HonchoLLMCallStreamChunk(content=chunk.text) final_chunk = chunk + # NOTE: Blocked-response check is intentionally omitted for streaming. + # Exceptions mid-iteration in an async generator won't be caught by + # the tenacity retry wrapper in honcho_llm_call. finish_reason = "stop" # Default fallback gemini_output_tokens: int | None = None if ( diff --git a/src/utils/summarizer.py b/src/utils/summarizer.py index 46415a3a..ca1965b9 100644 --- a/src/utils/summarizer.py +++ b/src/utils/summarizer.py @@ -119,7 +119,7 @@ Return only the summary without any explanation or meta-commentary. {formatted_messages} -Produce as thorough a summary as possible in {output_words} words or less. +Hard limit: {output_words} words maximum. If needed, drop lower-priority detail to stay within the limit. """) @@ -153,7 +153,7 @@ Return only the summary without any explanation or meta-commentary. {formatted_messages} -Produce as thorough a summary as possible in {output_words} words or less. +Hard limit: {output_words} words maximum. If needed, drop lower-priority detail to stay within the limit. """) @@ -565,8 +565,18 @@ async def _create_summary( # Detect potential issues with the summary if not summary_text.strip(): logger.error( - "Generated summary is empty! This may indicate a token limit issue." + "Generated summary is empty (finish_reasons=%s). Falling back to basic summary.", + response.finish_reasons, ) + is_fallback = True + summary_text = ( + f"Conversation with {message_count} messages about {last_message_content_preview}..." + if message_count > 0 + else "" + ) + summary_tokens = estimate_tokens(summary_text) if summary_text else 0 + llm_input_tokens = 0 + llm_output_tokens = 0 except Exception: logger.exception("Error generating summary!") # Fallback to a basic summary in case of error @@ -575,7 +585,7 @@ async def _create_summary( if message_count > 0 else "" ) - summary_tokens = 50 + summary_tokens = 0 is_fallback = True return ( diff --git a/tests/bench/coverage.py b/tests/bench/coverage.py index 26734ccd..d5baa829 100644 --- a/tests/bench/coverage.py +++ b/tests/bench/coverage.py @@ -80,29 +80,34 @@ logger = configure_logging(level=logging.INFO, name=__name__) 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" + + 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 + + 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 + + CRITICAL = "critical" # Core identifying information + IMPORTANT = "important" # Significant details + MINOR = "minor" # Nice-to-have details + TRIVIAL = "trivial" # Marginal information # ============================================================================= @@ -113,75 +118,79 @@ class ImportanceLevel(Enum): @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 + 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 + matched_extraction: str = "" # Which extracted fact covers this (if any) + match_quality: float = 0.0 # 0-1, how well it matches explanation: str = "" -@dataclass +@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 + 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 - + 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 - + 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, @@ -190,9 +199,15 @@ class CoverageReport: "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), + "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), @@ -200,15 +215,21 @@ class CoverageReport: "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, + "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()}, + "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], @@ -381,14 +402,14 @@ For each question, determine if the EXTRACTED FACTS (not the source!) provide en 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, @@ -402,15 +423,12 @@ class CoverageJudge: 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] + self, system: str, user: str, tool_def: dict[str, Any] ) -> dict[str, Any]: """Call LLM with structured tool output.""" try: @@ -451,7 +469,10 @@ class CoverageJudge: {"role": "user", "content": user}, ], tools=cast(Any, [openai_tool]), - tool_choice={"type": "function", "function": {"name": tool_def["name"]}}, + tool_choice={ + "type": "function", + "function": {"name": tool_def["name"]}, + }, ), timeout=300.0, ) @@ -474,13 +495,15 @@ class CoverageJudge: logger.exception("Unexpected error in LLM call") raise - async def extract_gold_facts(self, source_messages: list[dict[str, Any]]) -> list[GoldFact]: + 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", @@ -494,56 +517,64 @@ class CoverageJudge: "properties": { "content": { "type": "string", - "description": "The fact as a standalone statement" + "description": "The fact as a standalone statement", }, "category": { "type": "string", - "enum": [c.value for c in FactCategory] + "enum": [c.value for c in FactCategory], }, "importance": { "type": "string", - "enum": [i.value for i in ImportanceLevel] + "enum": [i.value for i in ImportanceLevel], }, "source_span": { "type": "string", - "description": "The text that supports this fact" + "description": "The text that supports this fact", }, "requires_inference": { "type": "boolean", - "description": "Whether extracting this requires reasoning beyond literal text" - } + "description": "Whether extracting this requires reasoning beyond literal text", + }, }, - "required": ["content", "category", "importance"] - } + "required": ["content", "category", "importance"], + }, } }, - "required": ["facts"] - } + "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 + 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 + 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)), - )) + 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 @@ -552,17 +583,15 @@ class CoverageJudge: return gold_facts async def match_coverage( - self, - gold_facts: list[GoldFact], - extracted_facts: list[str] + 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", @@ -577,38 +606,40 @@ class CoverageJudge: "gold_index": {"type": "integer"}, "status": { "type": "string", - "enum": [s.value for s in CoverageStatus] + "enum": [s.value for s in CoverageStatus], }, "matched_extraction": { "type": "string", - "description": "The extracted fact that covers this (if any)" + "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)" + "description": "How well it matches (1.0 = perfect)", }, - "explanation": {"type": "string"} + "explanation": {"type": "string"}, }, - "required": ["gold_index", "status"] - } + "required": ["gold_index", "status"], + }, } }, - "required": ["matches"] - } + "required": ["matches"], + }, } - + # Format for LLM gold_text = "\n".join( - f"{i+1}. [{gf.importance.value.upper()}] {gf.content}" + 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)" - + + 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, ( @@ -616,47 +647,57 @@ class CoverageJudge: + f"EXTRACTED FACTS (what was actually extracted):\n{extracted_text}\n\n" + "For each gold fact, determine its coverage status." ), - tool_def + tool_def, ) - + matches: list[CoverageMatch] = [] + matched_indices: set[int] = set() 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): + matched_indices.add(idx) 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", "")), - )) + 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, - )) - + 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" - )) - + 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]] + 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", @@ -672,52 +713,50 @@ class CoverageJudge: "expected_answer": {"type": "string"}, "difficulty": { "type": "string", - "enum": ["easy", "medium", "hard"] - } + "enum": ["easy", "medium", "hard"], + }, }, - "required": ["question"] - } + "required": ["question"], + }, } }, - "required": ["questions"] - } + "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' + 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 + 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] + 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", @@ -732,22 +771,26 @@ class CoverageJudge: "question_index": {"type": "integer"}, "answerable": { "type": "string", - "enum": ["yes", "partial", "no"] + "enum": ["yes", "partial", "no"], }, "answer_from_extraction": {"type": "string"}, - "explanation": {"type": "string"} + "explanation": {"type": "string"}, }, - "required": ["question_index", "answerable"] - } + "required": ["question_index", "answerable"], + }, } }, - "required": ["results"] - } + "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)" - + + 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, ( @@ -755,9 +798,9 @@ class CoverageJudge: + f"EXTRACTED FACTS:\n{extracted_text}\n\n" + "For each question, determine if it can be answered from the extracted facts alone." ), - tool_def + tool_def, ) - + answerable: float = 0 for item in cast(list[dict[str, Any]], result.get("results", [])): status = cast(str, item.get("answerable", "no")) @@ -765,18 +808,16 @@ class CoverageJudge: 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]] + 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", @@ -791,24 +832,26 @@ class CoverageJudge: "index": {"type": "integer"}, "is_grounded": {"type": "boolean"}, "is_hallucinated": {"type": "boolean"}, - "explanation": {"type": "string"} + "explanation": {"type": "string"}, }, - "required": ["index", "is_grounded"] - } + "required": ["index", "is_grounded"], + }, } }, - "required": ["analyses"] - } + "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' + if msg.get("speaker", "user") == "user" ) - - extracted_text = "\n".join(f"{i+1}. {f}" for i, f in enumerate(extracted_facts)) - + + 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. " @@ -820,37 +863,39 @@ class CoverageJudge: + f"EXTRACTED FACTS:\n{extracted_text}\n\n" + "Analyze each extracted fact." ), - tool_def + 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)) - + 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, @@ -858,8 +903,10 @@ class CoverageJudge: ImportanceLevel.MINOR: 1.0, ImportanceLevel.TRIVIAL: 0.5, } - - total_weight = sum(importance_weights[gf.importance] for gf in report.gold_facts) + + 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 @@ -870,43 +917,67 @@ class CoverageJudge: 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 - + + 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) - + 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) - + 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] + 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) - + 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.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 - + 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 + if m.status == CoverageStatus.MISSING + and m.gold_fact.importance == ImportanceLevel.CRITICAL ] async def evaluate( @@ -930,55 +1001,62 @@ class CoverageJudge: metrics, per-category/importance breakdowns, and optional QA coverage results. """ - + # Estimate token count source_text = " ".join( - msg.get("text", "") for msg in source_messages + 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_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...") + 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) + 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 @@ -990,24 +1068,24 @@ class CoverageJudge: @dataclass class CombinedScore: """Combined molecular quality + coverage recall score.""" - + # Individual scores - molecular: float = 0.0 # From MolecularBench + molecular: float = 0.0 # From MolecularBench decontextuality: float = 0.0 minimality: float = 0.0 - - coverage: float = 0.0 # From CoverageBench + + 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 - + 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 @@ -1018,17 +1096,17 @@ class CombinedScore: 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) @@ -1046,7 +1124,7 @@ def compute_combined_score( ) -> CombinedScore: """ Combine molecular quality with coverage recall. - + Args: molecular_score: From MolecularBench decontextuality: Decontextuality score @@ -1057,24 +1135,25 @@ def compute_combined_score( 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, + 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 + score.molecular >= min_molecular and score.coverage >= min_coverage ) - + return score @@ -1085,21 +1164,23 @@ def compute_combined_score( 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"\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) @@ -1107,35 +1188,41 @@ def print_report(report: CoverageReport) -> None: 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)})") - + 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]): + 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]): + 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) @@ -1145,19 +1232,27 @@ def print_report(report: CoverageReport) -> None: async def main(): - parser = argparse.ArgumentParser(description="CoverageBench - Information Recall Evaluation") + 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( + "--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") + 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( @@ -1171,38 +1266,44 @@ async def main(): ) else: client = create_openai_client(api_key=args.api_key) - + judge = CoverageJudge( - client, - args.model, + client, + args.model, args.provider, use_qa_verification=not args.no_qa, - verbose=args.verbose + 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")): + 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] + all_traces = all_traces[: args.limit] - print(f"Evaluating {len(all_traces)} traces for coverage with concurrency={args.concurrency}...\n") + 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]: + 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) @@ -1219,17 +1320,26 @@ async def main(): 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) + 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) + 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)] + 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"): + for coro in cast(Any, tqdm).as_completed( + tasks, total=len(tasks), desc="Evaluating traces" + ): idx: int report: CoverageReport | None idx, report = await coro @@ -1247,7 +1357,7 @@ async def main(): print("=" * 70) for report in results: print_report(report) - + total_duration = time.time() - overall_start # Save results @@ -1264,7 +1374,8 @@ async def main(): "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), + "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), }, @@ -1297,4 +1408,4 @@ async def main(): if __name__ == "__main__": - exit(asyncio.run(main())) \ No newline at end of file + exit(asyncio.run(main())) diff --git a/tests/bench/locomo_summary.py b/tests/bench/locomo_summary.py new file mode 100644 index 00000000..9b046076 --- /dev/null +++ b/tests/bench/locomo_summary.py @@ -0,0 +1,577 @@ +""" +Honcho LoCoMo Summary Evaluation Runner + +Evaluates Honcho's summary-making process by using only session summaries +as context for a base model to answer LoCoMo questions. + +This isolates the quality of Honcho's summarization: instead of using the +full dialectic agent or raw conversation context, we retrieve the generated +summaries and feed them to a base LLM. The resulting scores measure how much +information the summaries retain. + +## Comparison points + +- `locomo_baseline.py`: Raw conversation as context (upper bound for the model) +- `locomo_summary.py` (this file): Honcho summaries only as context +- `locomo.py`: Full Honcho memory system (dialectic agent) + +## To use + +0. Set up env: +``` +uv sync +source .venv/bin/activate +``` + +1. Run the test harness: +``` +python -m tests.bench.harness +``` + +2. Run this file with the LoCoMo dataset: +``` +python -m tests.bench.locomo_summary --data-file tests/bench/locomo_data/locomo10.json +``` + +Optional arguments: +``` +--timeout: Timeout for deriver queue to empty in seconds (default: 10 minutes) +--base-api-port: Base port for Honcho API instances (default: 8000) +--pool-size: Number of Honcho instances in the pool (default: 1) +--batch-size: Number of conversations to run concurrently in each batch (default: 1) +--json-output: Path to write JSON summary results for analytics +--cleanup-workspace: Delete workspace after executing each conversation (default: False) +--sample-id: Run only the conversation with this sample_id (skips all others) +--test-count: Number of conversations to run (default: all) +--question-count: Number of questions per conversation to run (default: all) +--model: Model to use for answering questions given summary context (default: anthropic/claude-haiku-4.5) +``` + +## Other notes +- Summaries are enabled (overriding the base runner default) so the deriver generates them +- Uses OpenRouter API for the base model (configured via LLM_OPENAI_COMPATIBLE_API_KEY) +- Evaluation uses the same GPT-4o-mini judge as the standard LoCoMo runner +""" + +import argparse +import os +import time +from datetime import datetime +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv +from honcho.api_types import ( + MessageCreateParams, + SessionConfiguration, + SummaryConfiguration, +) +from honcho.session import SessionPeerConfig +from openai import AsyncOpenAI + +from src.config import settings + +from .locomo import format_message_with_image +from .locomo_common import ( + CATEGORY_NAMES, + ConversationResult, + QuestionResult, + calculate_category_scores, + calculate_tokens, + extract_sessions, + filter_questions, + generate_json_summary, + get_evidence_context, + judge_response, + load_locomo_data, + parse_locomo_date, + print_summary, +) +from .runner_common import ( + BaseRunner, + ItemContext, + RunnerConfig, + add_common_arguments, + create_openai_client, + validate_common_arguments, +) + +# Load .env from bench directory +bench_dir = Path(__file__).parent +load_dotenv(bench_dir / ".env") + +# Default model for answering questions given summary context +DEFAULT_MODEL = "anthropic/claude-haiku-4.5" + + +class LoCoMoSummaryRunner(BaseRunner[ConversationResult]): + """ + Evaluates Honcho summary quality by using only summaries as context + for a base model to answer LoCoMo questions. + """ + + def __init__( + self, + config: RunnerConfig, + data_file: Path, + sample_id: str | None = None, + test_count: int | None = None, + question_count: int | None = None, + model: str = DEFAULT_MODEL, + ): + """ + Initialize the LoCoMo summary evaluation runner. + + Args: + config: Common runner configuration + data_file: Path to the LoCoMo JSON file + sample_id: Optional sample_id to run only that conversation + test_count: Optional number of conversations to run + question_count: Optional limit on questions per conversation + model: Model to use for answering questions given summary context + """ + self.data_file: Path = data_file + self.sample_id_filter: str | None = sample_id + self.test_count: int | None = test_count + self.question_count: int | None = question_count + self.model: str = model + + # Initialize base class + super().__init__(config) + + # Initialize OpenRouter client for the model under test + openrouter_base_url = os.getenv( + "LLM_OPENAI_COMPATIBLE_BASE_URL", "https://openrouter.ai/api/v1" + ) + self.openrouter_client: AsyncOpenAI = create_openai_client( + base_url=openrouter_base_url, + env_key_name="LLM_OPENAI_COMPATIBLE_API_KEY", + ) + + # Initialize OpenAI client for judging responses + self.openai_client: AsyncOpenAI = create_openai_client() + + def get_metrics_prefix(self) -> str: + return "locomo_summary" + + def _get_session_configuration(self) -> SessionConfiguration: + """Enable summaries so the deriver generates them during processing.""" + return SessionConfiguration(summary=SummaryConfiguration(enabled=True)) + + def load_items(self) -> list[Any]: + """Load conversations from the data file.""" + conversations = load_locomo_data(self.data_file) + + # Filter by sample_id if specified + if self.sample_id_filter is not None: + conversations = [ + c for c in conversations if c.get("sample_id") == self.sample_id_filter + ] + if not conversations: + print( + f"Error: No conversation found with sample_id '{self.sample_id_filter}'" + ) + return [] + print(f"Filtering to sample_id '{self.sample_id_filter}'") + + # Limit by test_count + if self.test_count is not None and self.test_count > 0: + conversations = conversations[: self.test_count] + print(f"Limiting to {len(conversations)} conversations") + + return conversations + + def get_workspace_id(self, item: Any) -> str: + """Return workspace ID for a conversation.""" + sample_id = item.get("sample_id", "unknown") + return f"locomo_summary_{sample_id}" + + def get_session_id(self, item: Any, workspace_id: str) -> str: + """Return session ID for a conversation.""" + return f"{workspace_id}_session" + + async def setup_peers(self, ctx: ItemContext, item: Any) -> None: + """Create peers using speaker names as IDs.""" + conversation = item.get("conversation", {}) + speaker_a = conversation.get("speaker_a", "User") + speaker_b = conversation.get("speaker_b", "Assistant") + + ctx.peers["speaker_a"] = await ctx.honcho_client.aio.peer(id=speaker_a) + ctx.peers["speaker_b"] = await ctx.honcho_client.aio.peer(id=speaker_b) + ctx.peers["_speaker_a_name"] = speaker_a + ctx.peers["_speaker_b_name"] = speaker_b + + async def setup_session(self, ctx: ItemContext, item: Any) -> None: + """Create and configure session - observe BOTH peers.""" + peer_a = ctx.peers["speaker_a"] + peer_b = ctx.peers["speaker_b"] + + ctx.session = await ctx.honcho_client.aio.session( + id=ctx.session_id, configuration=self._get_session_configuration() + ) + + # Observe both peers since questions ask about both speakers + await ctx.session.aio.add_peers( + [ + (peer_a, SessionPeerConfig(observe_me=True, observe_others=False)), + (peer_b, SessionPeerConfig(observe_me=True, observe_others=False)), + ] + ) + + async def ingest_messages(self, ctx: ItemContext, item: Any) -> int: + """Ingest conversation messages into the session.""" + conversation = item.get("conversation", {}) + speaker_a = ctx.peers["_speaker_a_name"] + speaker_b = ctx.peers["_speaker_b_name"] + peer_a = ctx.peers["speaker_a"] + peer_b = ctx.peers["speaker_b"] + + # Extract and ingest all sessions + sessions = extract_sessions(conversation) + + messages: list[MessageCreateParams] = [] + total_tokens = 0 + + for date_str, session_messages in sessions: + session_date = parse_locomo_date(date_str) if date_str else None + + for msg in session_messages: + speaker = msg.get("speaker", "") + content, metadata = format_message_with_image(msg) + total_tokens += calculate_tokens(content) + + # Map speaker to peer by name + if speaker == speaker_a: + messages.append( + peer_a.message( + content, metadata=metadata, created_at=session_date + ) + ) + elif speaker == speaker_b: + messages.append( + peer_b.message( + content, metadata=metadata, created_at=session_date + ) + ) + + # Store counts for results + ctx.peers["_total_tokens"] = total_tokens + ctx.peers["_total_sessions"] = len(sessions) + ctx.peers["_total_turns"] = len(messages) + + # Add messages in batches of 100 + for i in range(0, len(messages), 100): + batch = messages[i : i + 100] + await ctx.session.aio.add_messages(batch) + + return len(messages) + + def get_dream_observers(self, item: Any) -> list[str]: + """Return both speaker names - LoCoMo triggers dreams for both.""" + conversation = item.get("conversation", {}) + speaker_a = conversation.get("speaker_a", "User") + speaker_b = conversation.get("speaker_b", "Assistant") + return [speaker_a, speaker_b] + + async def _retrieve_summary_context(self, ctx: ItemContext) -> str: + """ + Retrieve summaries from the session and format them as context. + + Collects and concatenates both long and short summaries (long first, + then short), including token counts in headings. + + Returns: + Formatted summary context string, or a note if no summaries available. + """ + workspace_id = ctx.workspace_id + + summaries = await ctx.session.aio.summaries() + + parts: list[str] = [] + + if summaries.long_summary: + parts.append( + f"[Long Summary ({summaries.long_summary.token_count} tokens)]" + ) + parts.append(summaries.long_summary.content) + print( + f" [{workspace_id}] Retrieved long summary: {summaries.long_summary.token_count} tokens" + ) + + if summaries.short_summary: + parts.append( + f"\n[Short Summary ({summaries.short_summary.token_count} tokens)]" + ) + parts.append(summaries.short_summary.content) + print( + f" [{workspace_id}] Retrieved short summary: {summaries.short_summary.token_count} tokens" + ) + + if not parts: + print(f" [{workspace_id}] WARNING: No summaries available!") + return "(No summaries were generated for this conversation.)" + + return "\n".join(parts) + + async def execute_questions( + self, ctx: ItemContext, item: Any + ) -> ConversationResult: + """Execute all questions using only summary context.""" + start_time = time.time() + sample_id = item.get("sample_id", "unknown") + conversation = item.get("conversation", {}) + qa_list = item.get("qa", []) + workspace_id = ctx.workspace_id + + speaker_a = ctx.peers["_speaker_a_name"] + speaker_b = ctx.peers["_speaker_b_name"] + + result: ConversationResult = { + "sample_id": sample_id, + "speaker_a": speaker_a, + "speaker_b": speaker_b, + "total_sessions": ctx.peers.get("_total_sessions", 0), + "total_turns": ctx.peers.get("_total_turns", 0), + "total_tokens": ctx.peers.get("_total_tokens", 0), + "question_results": [], + "category_scores": {}, + "overall_score": 0.0, + "error": None, + "start_time": start_time, + "end_time": 0.0, + "duration_seconds": 0.0, + } + + # Retrieve summaries as context + print(f"[{workspace_id}] Retrieving summaries...") + summary_context = await self._retrieve_summary_context(ctx) + + # Build system prompt with summary context + system_prompt = ( + f"You are a helpful assistant with memory of past conversations " + f"between {speaker_a} and {speaker_b}.\n\n" + f"Below are summaries of their past conversations. Use these summaries " + f"to answer the user's question as accurately as possible.\n\n" + f"=== CONVERSATION SUMMARIES ===\n" + f"{summary_context}\n" + f"=== END SUMMARIES ===" + ) + + # Filter questions + filtered_qa = filter_questions( + qa_list, + exclude_adversarial=True, + test_count=self.question_count, + ) + + print(f"[{workspace_id}] Executing {len(filtered_qa)} questions...") + + # Execute questions + for q_idx, qa in enumerate(filtered_qa): + question = qa.get("question", "") + expected_answer = qa.get("answer", "") + category = qa.get("category", 0) + evidence = qa.get("evidence", []) + category_name = CATEGORY_NAMES.get(category, f"category_{category}") + + print(f" Q{q_idx + 1} [{category_name}]: {question[:80]}...") + + try: + # Ask the base model with summary context + response = await self.openrouter_client.chat.completions.create( + model=self.model, + max_tokens=settings.DIALECTIC.MAX_OUTPUT_TOKENS, + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": question}, + ], + ) + + if not response.choices or not response.choices[0].message.content: + actual_response = "" + else: + actual_response = response.choices[0].message.content + + # Get evidence context for the judge + evidence_context = get_evidence_context(conversation, evidence) + + # Judge the response + judgment = await judge_response( + self.openai_client, + question, + str(expected_answer), + actual_response, + evidence_context=evidence_context, + ) + + passed = judgment.get("passed", False) + + question_result = QuestionResult( + question_id=q_idx, + question=question, + expected_answer=str(expected_answer), + actual_response=actual_response, + category=category, + category_name=category_name, + evidence=evidence, + judgment=judgment, + passed=passed, + ) + + result["question_results"].append(question_result) + + status = "PASS" if passed else "FAIL" + print(f" [{status}]") + if not passed: + print(f" Expected: {expected_answer}") + print(f" Got: {actual_response[:200]}...") + + except Exception as e: + self.logger.exception(f"Error executing question {q_idx}: {e}") + question_result = QuestionResult( + question_id=q_idx, + question=question, + expected_answer=str(expected_answer), + actual_response=f"ERROR: {e}", + category=category, + category_name=category_name, + evidence=evidence, + judgment={"passed": False, "reasoning": str(e)}, + passed=False, + ) + result["question_results"].append(question_result) + + # Calculate category scores + result["category_scores"] = calculate_category_scores( + result["question_results"] + ) + + # Calculate overall score (pass rate) + if result["question_results"]: + passed_count = sum(1 for qr in result["question_results"] if qr["passed"]) + result["overall_score"] = passed_count / len(result["question_results"]) + + result["end_time"] = time.time() + result["duration_seconds"] = result["end_time"] - result["start_time"] + + print(f"\nOverall Score: {result['overall_score']:.3f}") + + return result + + def print_summary( + self, results: list[ConversationResult], total_duration: float + ) -> None: + """Print summary using the common function.""" + print_summary(results, total_duration) + + def generate_output( + self, results: list[ConversationResult], total_duration: float + ) -> None: + """Generate JSON output file.""" + if self.config.json_output: + output_file = self.config.json_output + else: + output_file = ( + bench_dir + / f"eval_results/locomo_summary_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + ) + + generate_json_summary( + results, + total_duration, + output_file, + metadata_extra={ + "data_file": str(self.data_file), + "runner_type": "summary_context", + "model": self.model, + "base_api_port": self.config.base_api_port, + "pool_size": self.config.pool_size, + "timeout_seconds": self.config.timeout_seconds, + "deriver_settings": settings.DERIVER.model_dump(), + "summary_settings": settings.SUMMARY.model_dump(), + "dream_settings": settings.DREAM.model_dump(), + }, + ) + + +def main() -> int: + """Main entry point for the LoCoMo summary evaluation runner.""" + parser = argparse.ArgumentParser( + description="Evaluate Honcho summary quality using LoCoMo questions", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s --data-file tests/bench/locomo_data/locomo10.json + %(prog)s --data-file locomo10.json --pool-size 4 + %(prog)s --data-file locomo10.json --sample-id "conv-26" + %(prog)s --data-file locomo10.json --model anthropic/claude-sonnet-4.5 + %(prog)s --data-file locomo10.json --test-count 5 --question-count 20 + """, + ) + + parser.add_argument( + "--data-file", + type=Path, + required=True, + help="Path to LoCoMo JSON file (required)", + ) + + # Add common arguments shared across all runners + add_common_arguments(parser) + + # Summary-eval-specific arguments + parser.add_argument( + "--sample-id", + type=str, + help="Run only the conversation with this sample_id (skips all others)", + ) + + parser.add_argument( + "--test-count", + type=int, + help="Number of conversations to run from the data file (default: all)", + ) + + parser.add_argument( + "--question-count", + type=int, + help="Number of questions per conversation to run (default: all)", + ) + + parser.add_argument( + "--model", + type=str, + default=DEFAULT_MODEL, + help=f"Model to use for answering questions given summary context (default: {DEFAULT_MODEL})", + ) + + args = parser.parse_args() + + # Validate common arguments + error = validate_common_arguments(args) + if error: + print(error) + return 1 + + # Validate locomo-specific arguments + if not args.data_file.exists(): + print(f"Error: Data file {args.data_file} does not exist") + return 1 + + # Create config and runner + config = RunnerConfig.from_args(args, default_timeout=600) + + runner = LoCoMoSummaryRunner( + config=config, + data_file=args.data_file, + sample_id=args.sample_id, + test_count=args.test_count, + question_count=args.question_count, + model=args.model, + ) + + return runner.run_and_summarize() + + +if __name__ == "__main__": + exit(main()) diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 4fa7e160..69e47086 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src import crud, models, schemas from src.config import settings from src.utils.agent_tools import ( + MAX_PEER_CARD_FACTS, ObservationsCreatedResult, ToolContext, _handle_create_observations, # pyright: ignore[reportPrivateUsage] @@ -788,6 +789,88 @@ class TestUpdatePeerCard: assert peer_card is not None assert "Name: John" in peer_card + async def test_deduplicates_and_caps_peer_card( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """Normalizes peer card updates to avoid unbounded growth.""" + workspace, peer1, peer2, _, _, _ = tool_test_data + ctx = make_tool_context() + + oversized = ["Name: John", " Name: John ", "", " "] + oversized.extend([f"Fact {i}" for i in range(MAX_PEER_CARD_FACTS + 5)]) + + await _handle_update_peer_card(ctx, {"content": oversized}) + + peer_card = await crud.get_peer_card( + db_session, + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + ) + assert peer_card is not None + assert len(peer_card) == MAX_PEER_CARD_FACTS + assert all(line.strip() for line in peer_card) + assert peer_card.count("Name: John") == 1 + + async def test_none_content_preserves_existing_card( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """None content should not overwrite the existing peer card.""" + workspace, peer1, peer2, _, _, _ = tool_test_data + ctx = make_tool_context() + + # First, create a valid peer card + await _handle_update_peer_card( + ctx, {"content": ["Name: Alice", "Location: NYC"]} + ) + + # Now attempt to update with None — should be a no-op + result = await _handle_update_peer_card(ctx, {"content": None}) + assert "empty" in result.lower() + + # Verify original card is preserved + peer_card = await crud.get_peer_card( + db_session, + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + ) + assert peer_card is not None + assert "Name: Alice" in peer_card + + async def test_empty_list_preserves_existing_card( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """Empty list should not clear the existing peer card.""" + workspace, peer1, peer2, _, _, _ = tool_test_data + ctx = make_tool_context() + + # First, create a valid peer card + await _handle_update_peer_card(ctx, {"content": ["Name: Bob", "Age: 30"]}) + + # Now attempt to update with empty list — should be a no-op + result = await _handle_update_peer_card(ctx, {"content": []}) + assert "empty" in result.lower() + + # Verify original card is preserved + peer_card = await crud.get_peer_card( + db_session, + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + ) + assert peer_card is not None + assert "Name: Bob" in peer_card + @pytest.mark.asyncio class TestGetPeerCard: diff --git a/tests/utils/test_clients.py b/tests/utils/test_clients.py index bf92ff37..f76cb27c 100644 --- a/tests/utils/test_clients.py +++ b/tests/utils/test_clients.py @@ -26,6 +26,7 @@ from openai.types.completion_usage import CompletionUsage from pydantic import BaseModel, Field from src.config import settings +from src.exceptions import LLMError from src.utils.clients import ( CLIENTS, HonchoLLMCallResponse, @@ -566,6 +567,11 @@ class TestGoogleClient: assert response.output_tokens == 5 assert response.finish_reasons == ["STOP"] + # Verify max output token cap is passed through + mock_aio.models.generate_content.assert_called_once() + call_args = mock_aio.models.generate_content.call_args + assert call_args.kwargs["config"]["max_output_tokens"] == 100 + async def test_google_json_mode(self): """Test Google/Gemini with JSON mode""" from google import genai @@ -606,9 +612,9 @@ class TestGoogleClient: # Verify JSON mode was set in config mock_aio.models.generate_content.assert_called_once() call_args = mock_aio.models.generate_content.call_args - assert ( - call_args.kwargs["config"]["response_mime_type"] == "application/json" - ) + config = call_args.kwargs["config"] + assert config["response_mime_type"] == "application/json" + assert config["max_output_tokens"] == 100 async def test_google_response_model(self): """Test Google/Gemini with structured output""" @@ -651,6 +657,7 @@ class TestGoogleClient: config = call_args.kwargs["config"] assert config["response_mime_type"] == "application/json" assert config["response_schema"] == SampleTestModel + assert config["max_output_tokens"] == 100 async def test_google_streaming(self): """Test Google/Gemini streaming response""" @@ -705,6 +712,11 @@ class TestGoogleClient: assert chunks[2].is_done is True assert chunks[2].finish_reasons == ["STOP"] + # Verify streaming call includes max output token cap + mock_aio.models.generate_content_stream.assert_called_once() + call_args = mock_aio.models.generate_content_stream.call_args + assert call_args.kwargs["config"]["max_output_tokens"] == 100 + async def test_google_no_candidates_fallback(self): """Test Google/Gemini fallback when no candidates""" from google import genai @@ -732,6 +744,133 @@ class TestGoogleClient: assert response.output_tokens == 0 # Fallback value assert response.finish_reasons == ["stop"] # Default fallback + @pytest.mark.parametrize( + "finish_reason", ["SAFETY", "RECITATION", "PROHIBITED_CONTENT", "BLOCKLIST"] + ) + async def test_google_blocked_response_raises_error(self, finish_reason: str): + """Test that blocked Gemini responses raise LLMError for retry/failover.""" + from google import genai + + mock_client = Mock(spec=genai.Client) + mock_response = Mock() + # Blocked responses typically have candidates with no content + mock_finish_reason = Mock() + mock_finish_reason.name = finish_reason + mock_candidate = Mock() + mock_candidate.content = None + mock_candidate.finish_reason = mock_finish_reason + mock_response.candidates = [mock_candidate] + mock_usage_metadata = Mock() + mock_usage_metadata.prompt_token_count = 10 + mock_usage_metadata.candidates_token_count = 0 + mock_response.usage_metadata = mock_usage_metadata + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio + + with ( + patch.dict(CLIENTS, {"google": mock_client}), + pytest.raises(LLMError, match=f"finish_reason={finish_reason}"), + ): + await honcho_llm_call_inner( + provider="google", + model="gemini-2.5-flash", + prompt="Summarize this", + max_tokens=1000, + ) + + async def test_google_max_tokens_empty_does_not_raise(self): + """Test that MAX_TOKENS with empty content returns normally (not a blocked response).""" + from google import genai + + mock_client = Mock(spec=genai.Client) + mock_response = Mock() + mock_finish_reason = Mock() + mock_finish_reason.name = "MAX_TOKENS" + mock_candidate = Mock() + mock_candidate.content = None + mock_candidate.finish_reason = mock_finish_reason + mock_response.candidates = [mock_candidate] + mock_usage_metadata = Mock() + mock_usage_metadata.prompt_token_count = 10 + mock_usage_metadata.candidates_token_count = 0 + mock_response.usage_metadata = mock_usage_metadata + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio + + with patch.dict(CLIENTS, {"google": mock_client}): + response = await honcho_llm_call_inner( + provider="google", + model="gemini-2.5-flash", + prompt="Hello", + max_tokens=100, + ) + # MAX_TOKENS is not a blocked reason — returns empty content without raising + assert response.content == "" + assert response.finish_reasons == ["MAX_TOKENS"] + + async def test_google_blocked_response_model_raises_error(self): + """Test that blocked responses in the response_model path raise LLMError.""" + from google import genai + + mock_client = Mock(spec=genai.Client) + mock_response = Mock() + mock_response.parsed = None + mock_finish_reason = Mock() + mock_finish_reason.name = "SAFETY" + mock_response.candidates = [Mock(finish_reason=mock_finish_reason)] + mock_usage_metadata = Mock() + mock_usage_metadata.prompt_token_count = 10 + mock_usage_metadata.candidates_token_count = 0 + mock_response.usage_metadata = mock_usage_metadata + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio + + with ( + patch.dict(CLIENTS, {"google": mock_client}), + pytest.raises(LLMError, match="finish_reason=SAFETY"), + ): + await honcho_llm_call_inner( + provider="google", + model="gemini-2.5-flash", + prompt="Generate a person", + max_tokens=100, + response_model=SampleTestModel, + ) + + async def test_google_blocked_finish_reason_with_valid_parsed_does_not_raise(self): + """Blocked finish_reason should not raise if parsed content is valid.""" + from google import genai + + mock_client = Mock(spec=genai.Client) + mock_response = Mock() + mock_response.parsed = SampleTestModel(name="Alice", age=30, active=True) + mock_finish_reason = Mock() + mock_finish_reason.name = "SAFETY" + mock_response.candidates = [Mock(finish_reason=mock_finish_reason)] + mock_usage_metadata = Mock() + mock_usage_metadata.prompt_token_count = 10 + mock_usage_metadata.candidates_token_count = 5 + mock_response.usage_metadata = mock_usage_metadata + mock_aio = Mock() + mock_aio.models.generate_content = AsyncMock(return_value=mock_response) + mock_client.aio = mock_aio + + with patch.dict(CLIENTS, {"google": mock_client}): + response = await honcho_llm_call_inner( + provider="google", + model="gemini-2.5-flash", + prompt="Generate a person", + max_tokens=100, + response_model=SampleTestModel, + ) + + assert isinstance(response.content, SampleTestModel) + assert response.content.name == "Alice" + assert response.finish_reasons == ["SAFETY"] + @pytest.mark.asyncio class TestGroqClient: diff --git a/tests/utils/test_summarizer.py b/tests/utils/test_summarizer.py new file mode 100644 index 00000000..3e8f8dc9 --- /dev/null +++ b/tests/utils/test_summarizer.py @@ -0,0 +1,219 @@ +""" +Tests for src/utils/summarizer.py + +Covers the _create_summary function's handling of empty, blocked, and +normal LLM responses, ensuring fallback logic prevents empty summaries +from being persisted. +""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from src.utils.clients import HonchoLLMCallResponse +from src.utils.summarizer import ( + Summary, + SummaryType, + _create_summary, # pyright: ignore[reportPrivateUsage] +) + +# Common test arguments for _create_summary +_FORMATTED_MESSAGES = "user: hello\nassistant: hi there" +_INPUT_TOKENS = 100 +_MESSAGE_PUBLIC_ID = "msg_abc123" +_LAST_MESSAGE_ID = 42 +_LAST_MESSAGE_CONTENT_PREVIEW = "hello there how are you" +_MESSAGE_COUNT = 5 + + +async def _call_create_summary( + summary_type: SummaryType, + *, + message_count: int = _MESSAGE_COUNT, + input_tokens: int = _INPUT_TOKENS, +) -> tuple[Summary, bool, int, int]: + """Helper to call _create_summary with standard test arguments.""" + return await _create_summary( + formatted_messages=_FORMATTED_MESSAGES, + previous_summary_text=None, + summary_type=summary_type, + input_tokens=input_tokens, + message_public_id=_MESSAGE_PUBLIC_ID, + last_message_id=_LAST_MESSAGE_ID, + last_message_content_preview=_LAST_MESSAGE_CONTENT_PREVIEW, + message_count=message_count, + ) + + +@pytest.mark.asyncio +class TestCreateSummary: + """Tests for the _create_summary function.""" + + async def test_normal_response_succeeds(self): + """Normal LLM response with content is preserved as-is.""" + mock_response = HonchoLLMCallResponse( + content="User greeted the assistant and asked about the weather.", + input_tokens=100, + output_tokens=15, + finish_reasons=["STOP"], + ) + + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + return_value=mock_response, + ): + ( + summary, + is_fallback, + input_tokens, + output_tokens, + ) = await _call_create_summary(SummaryType.SHORT) + + assert is_fallback is False + assert ( + summary["content"] + == "User greeted the assistant and asked about the weather." + ) + assert input_tokens == 100 + assert output_tokens == 15 + + async def test_empty_response_uses_fallback(self): + """Empty LLM response triggers fallback text instead of saving empty string.""" + mock_response = HonchoLLMCallResponse( + content="", + input_tokens=100, + output_tokens=0, + finish_reasons=["SAFETY"], + ) + + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + return_value=mock_response, + ): + ( + summary, + is_fallback, + input_tokens, + output_tokens, + ) = await _call_create_summary(SummaryType.SHORT) + + assert is_fallback is True + assert "Conversation with 5 messages" in summary["content"] + assert summary["content"] != "" + assert input_tokens == 0 + assert output_tokens == 0 + + async def test_whitespace_response_uses_fallback(self): + """Whitespace-only LLM response is treated as empty.""" + mock_response = HonchoLLMCallResponse( + content=" \n \t ", + input_tokens=100, + output_tokens=3, + finish_reasons=["STOP"], + ) + + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + return_value=mock_response, + ): + ( + summary, + is_fallback, + input_tokens, + output_tokens, + ) = await _call_create_summary(SummaryType.SHORT) + + assert is_fallback is True + assert "Conversation with 5 messages" in summary["content"] + assert input_tokens == 0 + assert output_tokens == 0 + + async def test_exception_uses_fallback(self): + """LLM exception triggers the existing fallback path.""" + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + side_effect=RuntimeError("API timeout"), + ): + ( + summary, + is_fallback, + input_tokens, + output_tokens, + ) = await _call_create_summary(SummaryType.SHORT) + + assert is_fallback is True + assert "Conversation with 5 messages" in summary["content"] + assert input_tokens == 0 + assert output_tokens == 0 + + async def test_long_type_routes_to_long_summary(self): + """SummaryType.LONG calls create_long_summary, not create_short_summary.""" + mock_response = HonchoLLMCallResponse( + content="A comprehensive summary of the conversation.", + input_tokens=100, + output_tokens=10, + finish_reasons=["STOP"], + ) + + with ( + patch( + "src.utils.summarizer.create_long_summary", + new_callable=AsyncMock, + return_value=mock_response, + ) as mock_long, + patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + ) as mock_short, + ): + summary, is_fallback, _, _ = await _call_create_summary(SummaryType.LONG) + + assert is_fallback is False + assert summary["content"] == "A comprehensive summary of the conversation." + mock_long.assert_called_once() + mock_short.assert_not_called() + + async def test_non_stop_finish_with_content_keeps_content(self): + """Non-STOP finish reason with actual content is preserved (not a false positive).""" + mock_response = HonchoLLMCallResponse( + content="User discussed their project deadlines and asked for help prioritizing", + input_tokens=100, + output_tokens=12, + finish_reasons=["MAX_TOKENS"], + ) + + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + return_value=mock_response, + ): + summary, is_fallback, _, _ = await _call_create_summary(SummaryType.SHORT) + + assert is_fallback is False + assert "project deadlines" in summary["content"] + + async def test_zero_message_count_empty_fallback(self): + """Empty response with zero messages produces empty fallback text.""" + mock_response = HonchoLLMCallResponse( + content="", + input_tokens=0, + output_tokens=0, + finish_reasons=["SAFETY"], + ) + + with patch( + "src.utils.summarizer.create_short_summary", + new_callable=AsyncMock, + return_value=mock_response, + ): + summary, is_fallback, _, _ = await _call_create_summary( + SummaryType.SHORT, message_count=0, input_tokens=0 + ) + + assert is_fallback is True + assert summary["content"] == "" + assert summary["token_count"] == 0