diff --git a/scripts/dialectic_cost_calculator.py b/scripts/dialectic_cost_calculator.py index 0c014628..72ea33cf 100644 --- a/scripts/dialectic_cost_calculator.py +++ b/scripts/dialectic_cost_calculator.py @@ -5,6 +5,8 @@ Dialectic Cost Calculator Calculates the maximum potential cost for each dialectic reasoning level based on configured settings and model pricing. +Supports both single-model and two-phase (search + synthesis) configurations. + Usage: uv run python scripts/dialectic_cost_calculator.py """ @@ -60,13 +62,29 @@ MODEL_PRICING: dict[str, dict[str, float]] = { "output": 5.00, "cached": 0.10, }, + "claude-sonnet-4-5": { + "input": 3.00, + "output": 15.00, + "cached": 0.30, + }, "claude-opus-4-5": { "input": 5.00, "output": 25.00, "cached": 0.50, }, + # OpenRouter GLM-4.7 Flash + "z-ai/glm-4.7-flash": { + "input": 0.07, + "output": 0.40, + "cached": 0.007, # Assuming 10% of input price for cached + }, } +# Text serialization overhead for synthesis phase +# When converting search messages to text format ([USER]: ..., [TOOL CALL: ...], etc.) +# there's ~20% overhead compared to structured message tokens +TEXT_SERIALIZATION_OVERHEAD = 1.20 + @dataclass class TokenEstimates: @@ -129,12 +147,12 @@ class TokenEstimates: return self.tool_result_per_iter + self.assistant_message_per_iter -def calculate_level_cost( +def calculate_single_model_cost( level_name: ReasoningLevel, base_estimates: TokenEstimates, ) -> dict[str, Any]: """ - Calculate the maximum potential cost for a reasoning level. + Calculate cost for single-model (non-two-phase) dialectic. Returns dict with all cost components, including both worst-case and realistic estimates. """ @@ -250,10 +268,14 @@ def calculate_level_cost( return { "level": level_name, + "two_phase": False, "provider": provider, "model": model, + "synthesis_provider": None, + "synthesis_model": None, "max_iterations": max_iterations, "thinking_tokens": thinking_budget, + "synthesis_thinking_tokens": 0, "first_iter_input": first_iter_input, "total_input_tokens": total_input_tokens, "total_cached_tokens": total_cached_tokens, @@ -269,9 +291,260 @@ def calculate_level_cost( # Shared input costs "input_cost": input_cost, "cached_cost": cached_cost, + # Phase breakdown (for two-phase display) + "search_cost_realistic": None, + "synthesis_cost_realistic": None, } +def calculate_two_phase_cost( + level_name: ReasoningLevel, + base_estimates: TokenEstimates, +) -> dict[str, Any]: + """ + Calculate cost for two-phase (search + synthesis) dialectic. + + In two-phase mode: + - Search phase: uses cheaper model for tool calling iterations + - Synthesis phase: uses smarter model for final response generation + (receives text-serialized search context, NO tool definitions) + + Returns dict with all cost components, including phase breakdown. + """ + level_config = settings.DIALECTIC.LEVELS[level_name] + synthesis_config = level_config.SYNTHESIS + assert synthesis_config is not None # Caller ensures this + + # Use full tools for search phase (minimal level doesn't use two-phase) + num_tools = NUM_DIALECTIC_TOOLS + prefetch = PREFETCH_OBSERVATIONS_FULL + + # Search phase settings + search_model = level_config.MODEL + search_provider = level_config.PROVIDER + search_max_iterations = level_config.MAX_TOOL_ITERATIONS + search_thinking_budget = level_config.THINKING_BUDGET_TOKENS + search_max_output = ( + level_config.MAX_OUTPUT_TOKENS + if level_config.MAX_OUTPUT_TOKENS is not None + else 1024 # Lower default for search + ) + + # Synthesis phase settings + synthesis_model = synthesis_config.MODEL + synthesis_provider = synthesis_config.PROVIDER + synthesis_thinking_budget = synthesis_config.THINKING_BUDGET_TOKENS + synthesis_max_output = ( + synthesis_config.MAX_OUTPUT_TOKENS + if synthesis_config.MAX_OUTPUT_TOKENS is not None + else base_estimates.max_output_tokens + ) + + # Get pricing for both models + search_pricing = MODEL_PRICING.get( + search_model, {"input": 0, "output": 0, "cached": 0} + ) + synthesis_pricing = MODEL_PRICING.get( + synthesis_model, {"input": 0, "output": 0, "cached": 0} + ) + + estimates = TokenEstimates( + system_prompt=base_estimates.system_prompt, + num_tools=num_tools, + peer_cards=base_estimates.peer_cards, + prefetched_observations=prefetch, + user_query=base_estimates.user_query, + session_history_max=base_estimates.session_history_max, + tool_result_per_iter=base_estimates.tool_result_per_iter, + assistant_message_per_iter=base_estimates.assistant_message_per_iter, + max_output_tokens=synthesis_max_output, + max_input_tokens=base_estimates.max_input_tokens, + realistic_tool_call_output=base_estimates.realistic_tool_call_output, + realistic_thinking_per_tool=base_estimates.realistic_thinking_per_tool, + realistic_final_answer=min( + synthesis_max_output, base_estimates.realistic_final_answer + ), + ) + + # Calculate search phase tokens + first_iter_input = min(estimates.first_iteration_input, estimates.max_input_tokens) + cacheable = estimates.cacheable_tokens + growth_per_iter = estimates.subsequent_iteration_growth() + cache_hit_rate = 0.90 + + # Search phase: only tool calling, no final response + search_total_input = 0 + search_total_cached = 0 + search_total_uncached = 0 + search_total_output_worst = 0 + search_total_output_realistic = 0 + + # For realistic output: use actual thinking budget (could be 0 for models without thinking) + # Don't assume 400 tokens of thinking if the model has THINKING_BUDGET_TOKENS=0 + realistic_thinking_per_tool = min( + estimates.realistic_thinking_per_tool, search_thinking_budget + ) + tool_iter_output = ( + realistic_thinking_per_tool + estimates.realistic_tool_call_output + ) + search_output_per_iter_worst = search_thinking_budget + search_max_output + + for i in range(search_max_iterations): + if i == 0: + iter_input = first_iter_input + cached = 0 + uncached = iter_input + else: + iter_input = min( + first_iter_input + (i * growth_per_iter), estimates.max_input_tokens + ) + cached = int(cacheable * cache_hit_rate) + uncached = iter_input - cached + + search_total_input += iter_input + search_total_cached += cached + search_total_uncached += uncached + search_total_output_worst += search_output_per_iter_worst + search_total_output_realistic += tool_iter_output + + # Search phase costs + search_input_cost = (search_total_uncached / 1_000_000) * search_pricing["input"] + search_cached_cost = (search_total_cached / 1_000_000) * search_pricing["cached"] + search_output_cost_worst = (search_total_output_worst / 1_000_000) * search_pricing[ + "output" + ] + search_output_cost_realistic = ( + search_total_output_realistic / 1_000_000 + ) * search_pricing["output"] + + search_cost_worst = ( + search_input_cost + search_cached_cost + search_output_cost_worst + ) + search_cost_realistic = ( + search_input_cost + search_cached_cost + search_output_cost_realistic + ) + + # Synthesis phase: single call with text-serialized search context + # The implementation serializes search conversation to text format: + # - System prompt (preserved) + # - Text-serialized search conversation ([USER]: ..., [TOOL CALL: ...], [TOOL RESULT: ...]) + # - Synthesis instruction (~100 tokens) + # NOTE: NO tool definitions (tools=None for synthesis call) + # + # Synthesis input components: + # - System prompt + # - Peer cards + # - Session history + # - Prefetched observations + # - User query + # - Search conversation (assistant messages + tool results) with text serialization overhead + # - Synthesis instruction + search_conversation_tokens = search_max_iterations * growth_per_iter + synthesis_input_base = ( + estimates.system_prompt + # NO tool definitions - synthesis has tools=None + + estimates.peer_cards + + estimates.session_history_max + + estimates.prefetched_observations + + estimates.user_query + ) + # Apply text serialization overhead to search conversation + serialized_search_tokens = int( + search_conversation_tokens * TEXT_SERIALIZATION_OVERHEAD + ) + synthesis_instruction_tokens = 100 + + synthesis_input = min( + synthesis_input_base + serialized_search_tokens + synthesis_instruction_tokens, + estimates.max_input_tokens, + ) + # No caching benefit for synthesis (different model, cache miss expected) + synthesis_input_cost = (synthesis_input / 1_000_000) * synthesis_pricing["input"] + + # Synthesis output + synthesis_output_worst = synthesis_thinking_budget + synthesis_max_output + synthesis_output_realistic = ( + synthesis_thinking_budget + estimates.realistic_final_answer + ) + + synthesis_output_cost_worst = ( + synthesis_output_worst / 1_000_000 + ) * synthesis_pricing["output"] + synthesis_output_cost_realistic = ( + synthesis_output_realistic / 1_000_000 + ) * synthesis_pricing["output"] + + synthesis_cost_worst = synthesis_input_cost + synthesis_output_cost_worst + synthesis_cost_realistic = synthesis_input_cost + synthesis_output_cost_realistic + + # Combined totals + total_input_tokens = search_total_input + synthesis_input + total_output_tokens_worst = search_total_output_worst + synthesis_output_worst + total_output_tokens_realistic = ( + search_total_output_realistic + synthesis_output_realistic + ) + + total_cost_worst = search_cost_worst + synthesis_cost_worst + total_cost_realistic = search_cost_realistic + synthesis_cost_realistic + + return { + "level": level_name, + "two_phase": True, + "provider": search_provider, + "model": search_model, + "synthesis_provider": synthesis_provider, + "synthesis_model": synthesis_model, + "max_iterations": search_max_iterations, + "thinking_tokens": search_thinking_budget, + "synthesis_thinking_tokens": synthesis_thinking_budget, + "first_iter_input": first_iter_input, + "total_input_tokens": total_input_tokens, + "total_cached_tokens": search_total_cached, + "total_uncached_tokens": search_total_uncached + synthesis_input, + # Worst-case output + "total_output_tokens": total_output_tokens_worst, + "output_cost": search_output_cost_worst + synthesis_output_cost_worst, + "total_cost": total_cost_worst, + # Realistic output + "total_output_tokens_realistic": total_output_tokens_realistic, + "output_cost_realistic": search_output_cost_realistic + + synthesis_output_cost_realistic, + "total_cost_realistic": total_cost_realistic, + # Shared input costs + "input_cost": search_input_cost + synthesis_input_cost, + "cached_cost": search_cached_cost, + # Phase breakdown + "search_cost_realistic": search_cost_realistic, + "synthesis_cost_realistic": synthesis_cost_realistic, + # Additional phase details + "search_input_tokens": search_total_input, + "search_output_tokens_realistic": search_total_output_realistic, + "synthesis_input_tokens": synthesis_input, + "synthesis_output_tokens_realistic": synthesis_output_realistic, + } + + +def calculate_level_cost( + level_name: ReasoningLevel, + base_estimates: TokenEstimates, +) -> dict[str, Any]: + """ + Calculate the maximum potential cost for a reasoning level. + + Automatically detects whether to use single-model or two-phase calculation + based on whether SYNTHESIS is configured. + + Returns dict with all cost components, including both worst-case and realistic estimates. + """ + level_config = settings.DIALECTIC.LEVELS[level_name] + + # Check if two-phase mode is enabled (synthesis config exists and not minimal) + if level_config.SYNTHESIS is not None and level_name != "minimal": + return calculate_two_phase_cost(level_name, base_estimates) + else: + return calculate_single_model_cost(level_name, base_estimates) + + def main(): console = Console() @@ -332,19 +605,23 @@ def main(): # Create summary table table = Table(title="Cost by Reasoning Level", show_lines=True) table.add_column("Level", style="cyan", no_wrap=True) - table.add_column("Model", style="dim", no_wrap=True) + table.add_column("Mode", style="dim", no_wrap=True) + table.add_column("Search Model", style="dim", no_wrap=True) + table.add_column("Synthesis Model", style="dim", no_wrap=True) table.add_column("Iters", justify="right") - table.add_column("Think", justify="right") table.add_column("Target", justify="right", style="dim") table.add_column("Realistic", justify="right", style="bold green") table.add_column("Worst Case", justify="right", style="yellow") for r in results: + mode = "2-phase" if r["two_phase"] else "single" + synthesis_model = r.get("synthesis_model") or "-" table.add_row( r["level"], + mode, r["model"], + synthesis_model, str(r["max_iterations"]), - f"{r['thinking_tokens']:,}", f"${TARGET_COSTS.get(r['level'], 0):.3f}", f"${r['total_cost_realistic']:.4f}", f"${r['total_cost']:.4f}", @@ -362,14 +639,28 @@ def main(): detail_table.add_column("Cached $", justify="right", style="dim") detail_table.add_column("Output $", justify="right") detail_table.add_column("Total $", justify="right", style="bold green") + detail_table.add_column("Search $", justify="right", style="dim") + detail_table.add_column("Synthesis $", justify="right", style="dim") for r in results: + search_cost = ( + f"${r['search_cost_realistic']:.4f}" + if r.get("search_cost_realistic") is not None + else "-" + ) + synthesis_cost = ( + f"${r['synthesis_cost_realistic']:.4f}" + if r.get("synthesis_cost_realistic") is not None + else "-" + ) detail_table.add_row( r["level"], f"${r['input_cost']:.4f}", f"${r['cached_cost']:.4f}", f"${r['output_cost_realistic']:.4f}", f"${r['total_cost_realistic']:.4f}", + search_cost, + synthesis_cost, ) console.print(detail_table) @@ -377,31 +668,78 @@ def main(): # Print detailed breakdown for max level console.print("\n[bold]Detailed Breakdown for 'max' Level:[/bold]") max_result = results[-1] - console.print(f" Model: {max_result['model']} ({max_result['provider']})") - console.print(f" Max iterations: {max_result['max_iterations']}") - console.print(f" Thinking budget per iteration: {max_result['thinking_tokens']:,}") - console.print(f" First iteration input: {max_result['first_iter_input']:,} tokens") - console.print( - f" Total input tokens (all iterations): {max_result['total_input_tokens']:,}" - ) - console.print( - f" - Uncached: {max_result['total_uncached_tokens']:,} @ ${MODEL_PRICING[max_result['model']]['input']}/1M" - ) - console.print( - f" - Cached: {max_result['total_cached_tokens']:,} @ ${MODEL_PRICING[max_result['model']]['cached']}/1M" - ) - console.print(" Output tokens:") - console.print( - f" - Realistic: {max_result['total_output_tokens_realistic']:,} " - + f"(9 tool calls × {estimates.realistic_thinking_per_tool + estimates.realistic_tool_call_output} + final {max_result['thinking_tokens'] + estimates.realistic_final_answer})" - ) - console.print( - f" - Worst case: {max_result['total_output_tokens']:,} " - + f"(10 × {max_result['thinking_tokens'] + estimates.max_output_tokens})" - ) - console.print( - f" - Output rate: ${MODEL_PRICING[max_result['model']]['output']}/1M" - ) + + if max_result["two_phase"]: + console.print(" [cyan]Mode: Two-phase (Search + Synthesis)[/cyan]") + console.print( + f" Search model: {max_result['model']} ({max_result['provider']})" + ) + console.print( + f" Synthesis model: {max_result['synthesis_model']} ({max_result['synthesis_provider']})" + ) + console.print(f" Search iterations: {max_result['max_iterations']}") + console.print( + f" Search thinking budget: {max_result['thinking_tokens']:,} tokens" + ) + console.print( + f" Synthesis thinking budget: {max_result['synthesis_thinking_tokens']:,} tokens" + ) + console.print() + console.print(" [dim]Search Phase:[/dim]") + console.print( + f" Input tokens: {max_result.get('search_input_tokens', 'N/A'):,}" + ) + console.print( + f" Output tokens (realistic): {max_result.get('search_output_tokens_realistic', 'N/A'):,}" + ) + console.print( + f" Cost (realistic): ${max_result['search_cost_realistic']:.4f}" + ) + console.print() + console.print(" [dim]Synthesis Phase:[/dim]") + console.print( + f" Input tokens: {max_result.get('synthesis_input_tokens', 'N/A'):,}" + ) + console.print( + f" Output tokens (realistic): {max_result.get('synthesis_output_tokens_realistic', 'N/A'):,}" + ) + console.print( + f" Cost (realistic): ${max_result['synthesis_cost_realistic']:.4f}" + ) + else: + console.print(" [cyan]Mode: Single-model[/cyan]") + console.print(f" Model: {max_result['model']} ({max_result['provider']})") + console.print(f" Max iterations: {max_result['max_iterations']}") + console.print( + f" Thinking budget per iteration: {max_result['thinking_tokens']:,}" + ) + console.print( + f" First iteration input: {max_result['first_iter_input']:,} tokens" + ) + console.print( + f" Total input tokens (all iterations): {max_result['total_input_tokens']:,}" + ) + if max_result["model"] in MODEL_PRICING: + console.print( + f" - Uncached: {max_result['total_uncached_tokens']:,} @ ${MODEL_PRICING[max_result['model']]['input']}/1M" + ) + console.print( + f" - Cached: {max_result['total_cached_tokens']:,} @ ${MODEL_PRICING[max_result['model']]['cached']}/1M" + ) + console.print(" Output tokens:") + console.print( + f" - Realistic: {max_result['total_output_tokens_realistic']:,} " + + f"({max_result['max_iterations'] - 1} tool calls × {estimates.realistic_thinking_per_tool + estimates.realistic_tool_call_output} + final {max_result['thinking_tokens'] + estimates.realistic_final_answer})" + ) + console.print( + f" - Worst case: {max_result['total_output_tokens']:,} " + + f"({max_result['max_iterations']} × {max_result['thinking_tokens'] + estimates.max_output_tokens})" + ) + if max_result["model"] in MODEL_PRICING: + console.print( + f" - Output rate: ${MODEL_PRICING[max_result['model']]['output']}/1M" + ) + console.print( f"\n [bold green]Realistic cost: ${max_result['total_cost_realistic']:.4f}[/bold green]" ) @@ -428,10 +766,14 @@ def main(): console.print(pricing_table) console.print( - "\n[dim]Note: 'Realistic' assumes tool calls use ~550 output tokens each " - + "(400 thinking + 150 JSON), with full budget only on final answer.\n" + "\n[dim]Note: 'Realistic' estimates tool call output as thinking_budget + 150 JSON tokens.\n" + + "Models with THINKING_BUDGET_TOKENS=0 only output ~150 tokens per tool call.\n" + "'Worst case' assumes max output tokens on every iteration. " - + "Actual costs may be even lower due to early termination.[/dim]\n" + + "Actual costs may be even lower due to early termination.\n\n" + + "Two-phase mode (when SYNTHESIS is configured):\n" + + "- Search phase: cheaper model handles tool calling (with tool definitions)\n" + + "- Synthesis phase: smarter model generates final response (NO tool definitions)\n" + + f"- Synthesis input includes {int(TEXT_SERIALIZATION_OVERHEAD * 100 - 100)}% overhead for text serialization of search context[/dim]\n" ) diff --git a/src/config.py b/src/config.py index 5e365c8b..e079097f 100644 --- a/src/config.py +++ b/src/config.py @@ -299,8 +299,61 @@ REASONING_LEVELS: list[ReasoningLevel] = [ ] +class SynthesisModelSettings(BaseModel): + """Settings for the synthesis model in two-phase dialectic. + + When configured, dialectic runs in two phases: + 1. Search phase: Uses the parent level's model for tool calling + 2. Synthesis phase: Uses this model for final response generation + """ + + model_config = SettingsConfigDict(populate_by_name=True) # pyright: ignore + + PROVIDER: Annotated[SupportedProviders, Field(validation_alias="provider")] + MODEL: Annotated[str, Field(validation_alias="model")] + BACKUP_PROVIDER: Annotated[ + SupportedProviders | None, Field(validation_alias="backup_provider") + ] = None + BACKUP_MODEL: Annotated[str | None, Field(validation_alias="backup_model")] = None + THINKING_BUDGET_TOKENS: Annotated[ + int, Field(ge=0, le=100_000, validation_alias="thinking_budget_tokens") + ] = 0 + MAX_OUTPUT_TOKENS: Annotated[ + int | None, Field(ge=1, le=100_000, validation_alias="max_output_tokens") + ] = None # None means use global DIALECTIC.MAX_OUTPUT_TOKENS + + @model_validator(mode="after") + def _validate_backup_configuration(self) -> "SynthesisModelSettings": + """Ensure both backup fields are set together or both are None.""" + if (self.BACKUP_PROVIDER is None) != (self.BACKUP_MODEL is None): + raise ValueError( + "BACKUP_PROVIDER and BACKUP_MODEL must both be set or both be None" + ) + return self + + @model_validator(mode="after") + def _validate_anthropic_thinking_budget(self) -> "SynthesisModelSettings": + """Ensure Anthropic thinking budget is >= 1024 when enabled.""" + if ( + self.PROVIDER == "anthropic" + and self.THINKING_BUDGET_TOKENS > 0 + and self.THINKING_BUDGET_TOKENS < 1024 + ): + raise ValueError( + f"THINKING_BUDGET_TOKENS must be >= 1024 for Anthropic provider when enabled (got {self.THINKING_BUDGET_TOKENS})" + ) + return self + + class DialecticLevelSettings(BaseModel): - """Settings for a specific reasoning level in the dialectic.""" + """Settings for a specific reasoning level in the dialectic. + + Supports two modes: + 1. Single-model mode (default): One model handles both tool calling and synthesis + 2. Two-model mode: Search model handles tool calling, synthesis model generates final response + + To enable two-model mode, set the SYNTHESIS field with SynthesisModelSettings. + """ model_config = SettingsConfigDict(populate_by_name=True) # pyright: ignore @@ -323,6 +376,11 @@ class DialecticLevelSettings(BaseModel): None # None/auto lets model decide, "any"/"required" forces tool use ) + # Optional synthesis model for two-phase dialectic + SYNTHESIS: Annotated[ + SynthesisModelSettings | None, Field(validation_alias="synthesis") + ] = None + @model_validator(mode="after") def _validate_backup_configuration(self) -> "DialecticLevelSettings": """Ensure both backup fields are set together or both are None.""" @@ -364,29 +422,38 @@ class DialecticSettings(HonchoSettings): TOOL_CHOICE="any", ), "low": DialecticLevelSettings( - PROVIDER="google", - MODEL="gemini-2.5-flash-lite", + PROVIDER="custom", + MODEL="z-ai/glm-4.7-flash", THINKING_BUDGET_TOKENS=0, - MAX_TOOL_ITERATIONS=5, - TOOL_CHOICE="any", + MAX_TOOL_ITERATIONS=4, ), "medium": DialecticLevelSettings( - PROVIDER="anthropic", - MODEL="claude-haiku-4-5", - THINKING_BUDGET_TOKENS=1024, - MAX_TOOL_ITERATIONS=2, + PROVIDER="custom", + MODEL="z-ai/glm-4.7-flash", + THINKING_BUDGET_TOKENS=0, + MAX_TOOL_ITERATIONS=4, + SYNTHESIS=SynthesisModelSettings( + PROVIDER="anthropic", + MODEL="claude-haiku-4-5", + THINKING_BUDGET_TOKENS=1024, + ), ), "high": DialecticLevelSettings( PROVIDER="anthropic", MODEL="claude-haiku-4-5", - THINKING_BUDGET_TOKENS=1024, - MAX_TOOL_ITERATIONS=4, + THINKING_BUDGET_TOKENS=0, + MAX_TOOL_ITERATIONS=5, ), "max": DialecticLevelSettings( PROVIDER="anthropic", MODEL="claude-haiku-4-5", - THINKING_BUDGET_TOKENS=2048, - MAX_TOOL_ITERATIONS=10, + THINKING_BUDGET_TOKENS=0, + MAX_TOOL_ITERATIONS=8, + SYNTHESIS=SynthesisModelSettings( + PROVIDER="anthropic", + MODEL="claude-opus-4-5", + THINKING_BUDGET_TOKENS=1024, + ), ), } ) diff --git a/src/dialectic/core.py b/src/dialectic/core.py index ae547273..21e9cbd6 100644 --- a/src/dialectic/core.py +++ b/src/dialectic/core.py @@ -5,6 +5,7 @@ This agent uses tools to gather context from the memory system and synthesize responses to queries about a peer. """ +import json import logging import time import uuid @@ -17,7 +18,7 @@ from src import crud from src.config import ReasoningLevel, settings from src.dialectic import prompts from src.telemetry import otel_metrics -from src.telemetry.events import DialecticCompletedEvent, emit +from src.telemetry.events import DialecticCompletedEvent, DialecticPhaseMetrics, emit from src.telemetry.logging import ( accumulate_metric, log_performance_metrics, @@ -371,14 +372,269 @@ class DialecticAgent: ) ) + def _build_synthesis_messages( + self, + search_messages: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + """ + Build messages for synthesis model in two-phase dialectic. + + Takes the full search conversation (including tool calls and results) + and serializes them into a text-based context that any provider can handle. + + Args: + search_messages: Full conversation history from search phase + synthesis_provider: Provider for synthesis model (kept for API compatibility) + + Returns: + Messages list for synthesis model + """ + # Extract system message and serialize the rest into text + system_content = "" + search_context_parts: list[str] = [] + + for msg in search_messages: + role = msg.get("role", "") + + if role == "system": + # Preserve system message content + content = msg.get("content", "") + if isinstance(content, str): + system_content: str = content + elif isinstance(content, list): + # Anthropic-style content blocks + for block in content: # pyright: ignore[reportUnknownVariableType] + if isinstance(block, dict) and block.get("type") == "text": # pyright: ignore[reportUnknownMemberType] + system_content += cast(str, block.get("text", "")) # pyright: ignore[reportUnknownMemberType] + + elif role == "user": + content = msg.get("content", "") + if isinstance(content, str) and content: + search_context_parts.append(f"[USER]: {content}") + elif isinstance(content, list): + # Could be Anthropic tool results + for block in content: # pyright: ignore[reportUnknownVariableType] + if ( + isinstance(block, dict) + and block.get("type") == "tool_result" # pyright: ignore[reportUnknownMemberType] + ): + tool_id: str = cast( + str, + block.get("tool_use_id", "unknown"), # pyright: ignore[reportUnknownMemberType] + ) + result: str = cast(str, block.get("content", "")) # pyright: ignore[reportUnknownMemberType] + search_context_parts.append( + f"[TOOL RESULT ({tool_id})]: {result}" + ) + + elif role == "assistant": + content = msg.get("content", "") + tool_calls = msg.get("tool_calls", []) + + # Handle text content + if isinstance(content, str) and content: + search_context_parts.append(f"[ASSISTANT]: {content}") + elif isinstance(content, list): + # Anthropic-style content blocks + for block in content: # pyright: ignore[reportUnknownVariableType] + if isinstance(block, dict): + if block.get("type") == "text": # pyright: ignore[reportUnknownMemberType] + text: str = cast(str, block.get("text", "")) # pyright: ignore[reportUnknownMemberType] + if text: + search_context_parts.append(f"[ASSISTANT]: {text}") + elif block.get("type") == "tool_use": # pyright: ignore[reportUnknownMemberType] + name: str = cast(str, block.get("name", "unknown")) # pyright: ignore[reportUnknownMemberType] + tool_input: Any = block.get("input", {}) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] + search_context_parts.append( + f"[TOOL CALL: {name}]: {json.dumps(tool_input)}" + ) + + # Handle OpenAI-style tool_calls + for tc in tool_calls: + func = tc.get("function", {}) + name = func.get("name", "unknown") + args = func.get("arguments", "{}") + search_context_parts.append(f"[TOOL CALL: {name}]: {args}") + + elif role == "tool": + # OpenAI-style tool result + tool_id = msg.get("tool_call_id", "unknown") + content = msg.get("content", "") + search_context_parts.append(f"[TOOL RESULT ({tool_id})]: {content}") + + # Build the synthesis messages + messages_out: list[dict[str, Any]] = [] + + # Include system prompt if present + if system_content: + messages_out.append({"role": "system", "content": system_content}) + + # Add the search context as a user message + search_context = "\n\n".join(search_context_parts) + synthesis_prompt = ( + f"The following is the search process used to gather information:\n\n" + f"---\n{search_context}\n---\n\n" + f"Based on the information gathered above, provide your " + f"final response to the original query. Be direct and helpful." + ) + + messages_out.append({"role": "user", "content": synthesis_prompt}) + + return messages_out + + def _log_two_phase_metrics( + self, + task_name: str, + run_id: str | None, + start_time: float, + response_content: str, + search_input_tokens: int, + search_output_tokens: int, + search_cache_read_tokens: int, + search_cache_creation_tokens: int, + search_tool_calls_count: int, + search_iterations: int, + synthesis_input_tokens: int, + synthesis_output_tokens: int, + synthesis_cache_read_tokens: int, + synthesis_cache_creation_tokens: int, + synthesis_thinking_content: str | None, + ) -> None: + """ + Log metrics for two-phase dialectic (search + synthesis). + + Args: + task_name: Metrics task identifier + run_id: Run identifier (None if using caller-provided metric_key) + start_time: Start time from time.perf_counter() + response_content: The full response text + search_*: Metrics from search phase + synthesis_*: Metrics from synthesis phase + """ + # Total metrics + total_input_tokens = search_input_tokens + synthesis_input_tokens + total_output_tokens = search_output_tokens + synthesis_output_tokens + total_cache_read_tokens = search_cache_read_tokens + synthesis_cache_read_tokens + total_cache_creation_tokens = ( + search_cache_creation_tokens + synthesis_cache_creation_tokens + ) + + accumulate_metric(task_name, "tool_calls", search_tool_calls_count, "count") + accumulate_metric(task_name, "search_iterations", search_iterations, "count") + + if synthesis_thinking_content: + accumulate_metric( + task_name, "synthesis_thinking", synthesis_thinking_content, "blob" + ) + + log_token_usage_metrics( + task_name, + total_input_tokens, + total_output_tokens, + total_cache_read_tokens, + total_cache_creation_tokens, + ) + + # Log phase-specific token metrics for cost analysis + accumulate_metric( + task_name, "search_input_tokens", search_input_tokens, "tokens" + ) + accumulate_metric( + task_name, "search_output_tokens", search_output_tokens, "tokens" + ) + accumulate_metric( + task_name, "synthesis_input_tokens", synthesis_input_tokens, "tokens" + ) + accumulate_metric( + task_name, "synthesis_output_tokens", synthesis_output_tokens, "tokens" + ) + + accumulate_metric(task_name, "response", response_content, "blob") + + elapsed_ms = (time.perf_counter() - start_time) * 1000 + accumulate_metric(task_name, "total_duration", elapsed_ms, "ms") + + if not self.metric_key and run_id is not None: + log_performance_metrics("dialectic_chat", run_id) + + # OTel metrics (push-based) + if settings.OTEL.ENABLED: + otel_metrics.record_dialectic_tokens( + count=total_input_tokens, + token_type=TokenTypes.INPUT.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=self.reasoning_level, + ) + otel_metrics.record_dialectic_tokens( + count=total_output_tokens, + token_type=TokenTypes.OUTPUT.value, + component=DialecticComponents.TOTAL.value, + reasoning_level=self.reasoning_level, + ) + + # Get model/provider info for phase metrics + level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level] + synthesis_settings = level_settings.SYNTHESIS + + # Build phase metrics + phases = [ + DialecticPhaseMetrics( + phase_name="search", + provider=level_settings.PROVIDER, + model=level_settings.MODEL, + input_tokens=search_input_tokens, + output_tokens=search_output_tokens, + cache_read_tokens=search_cache_read_tokens, + cache_creation_tokens=search_cache_creation_tokens, + iterations=search_iterations, + tool_calls_count=search_tool_calls_count, + ), + DialecticPhaseMetrics( + phase_name="synthesis", + provider=synthesis_settings.PROVIDER if synthesis_settings else None, + model=synthesis_settings.MODEL if synthesis_settings else None, + input_tokens=synthesis_input_tokens, + output_tokens=synthesis_output_tokens, + cache_read_tokens=synthesis_cache_read_tokens, + cache_creation_tokens=synthesis_cache_creation_tokens, + iterations=1, # Synthesis is always 1 iteration + tool_calls_count=0, # No tools in synthesis + ), + ] + + # Emit telemetry event with total iterations (search + 1 for synthesis) + emit( + DialecticCompletedEvent( + run_id=self._run_id, + workspace_name=self.workspace_name, + peer_name=self.observed, + session_name=self.session_name, + reasoning_level=self.reasoning_level, + two_phase_mode=True, + total_iterations=search_iterations + 1, + prefetched_conclusion_count=self._prefetched_conclusion_count, + tool_calls_count=search_tool_calls_count, + total_duration_ms=elapsed_ms, + input_tokens=total_input_tokens, + output_tokens=total_output_tokens, + cache_read_tokens=total_cache_read_tokens, + cache_creation_tokens=total_cache_creation_tokens, + phases=phases, + ) + ) + async def answer(self, query: str) -> str: """ Answer a query about the peer using agentic tool calling. + Supports two modes: + 1. Single-model mode: One model handles both tool calling and synthesis + 2. Two-model mode: Search model handles tool calling, synthesis model generates response + The agent will: 1. Receive the query - 2. Use tools to gather relevant context - 3. Synthesize a response grounded in the gathered context + 2. Use tools to gather relevant context (search phase) + 3. Synthesize a response grounded in the gathered context (synthesis phase) Args: query: The question to answer about the peer @@ -390,6 +646,7 @@ class DialecticAgent: # Get level-specific settings level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level] + synthesis_settings = level_settings.SYNTHESIS # Use minimal tools for minimal reasoning to reduce cost tools = ( @@ -397,48 +654,164 @@ class DialecticAgent: if self.reasoning_level == "minimal" else DIALECTIC_TOOLS ) - # Use level-specific max_output_tokens if set, otherwise global default - max_tokens = ( + + # Check if two-phase mode is enabled (synthesis config exists and not minimal) + use_two_phase = ( + synthesis_settings is not None and self.reasoning_level != "minimal" + ) + + if not use_two_phase: + # Single-model path (original behavior) + max_tokens = ( + level_settings.MAX_OUTPUT_TOKENS + if level_settings.MAX_OUTPUT_TOKENS is not None + else settings.DIALECTIC.MAX_OUTPUT_TOKENS + ) + + response: HonchoLLMCallResponse[str] = await honcho_llm_call( + llm_settings=level_settings, + prompt="", # Ignored since we pass messages + max_tokens=max_tokens, + tools=tools, + tool_choice=level_settings.TOOL_CHOICE, + tool_executor=tool_executor, + max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, + messages=self.messages, + track_name="Dialectic Agent", + thinking_budget_tokens=level_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_chat", + ) + + self._log_response_metrics( + task_name=task_name, + run_id=run_id, + start_time=start_time, + response_content=response.content, + input_tokens=response.input_tokens, + output_tokens=response.output_tokens, + cache_read_input_tokens=response.cache_read_input_tokens, + cache_creation_input_tokens=response.cache_creation_input_tokens, + tool_calls_count=len(response.tool_calls_made), + thinking_content=response.thinking_content, + iterations=response.iterations, + ) + + return response.content + + # Two-phase mode: Search then Synthesis + # Type narrowing: synthesis_settings is guaranteed non-None in two-phase mode + assert synthesis_settings is not None # nosec B101 + + # Phase 1: Search (non-streaming, with tools) + search_max_tokens = ( level_settings.MAX_OUTPUT_TOKENS if level_settings.MAX_OUTPUT_TOKENS is not None + else 1024 # Lower default for search - just tool call JSON + ) + + try: + search_response: HonchoLLMCallResponse[str] = await honcho_llm_call( + llm_settings=level_settings, + prompt="", + max_tokens=search_max_tokens, + tools=tools, + tool_choice=level_settings.TOOL_CHOICE, + tool_executor=tool_executor, + max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, + messages=self.messages, + track_name="Dialectic Search", + thinking_budget_tokens=level_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_search", + ) + except Exception as e: + # Fallback to single-model on search failure + logger.warning(f"Search phase failed: {e}, falling back to single-model") + max_tokens = ( + synthesis_settings.MAX_OUTPUT_TOKENS + if synthesis_settings.MAX_OUTPUT_TOKENS is not None + else settings.DIALECTIC.MAX_OUTPUT_TOKENS + ) + response = await honcho_llm_call( + llm_settings=synthesis_settings, + prompt="", + max_tokens=max_tokens, + tools=None, # No tools for fallback + tool_executor=None, + max_tool_iterations=1, + messages=self.messages, + track_name="Dialectic Fallback", + thinking_budget_tokens=synthesis_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_fallback", + ) + self._log_response_metrics( + task_name=task_name, + run_id=run_id, + start_time=start_time, + response_content=response.content, + input_tokens=response.input_tokens, + output_tokens=response.output_tokens, + cache_read_input_tokens=response.cache_read_input_tokens, + cache_creation_input_tokens=response.cache_creation_input_tokens, + tool_calls_count=0, + thinking_content=response.thinking_content, + iterations=1, + ) + return response.content + + # Phase 2: Synthesis (non-streaming, no tools) + synthesis_messages = self._build_synthesis_messages(search_response.messages) + synthesis_max_tokens = ( + synthesis_settings.MAX_OUTPUT_TOKENS + if synthesis_settings.MAX_OUTPUT_TOKENS is not None else settings.DIALECTIC.MAX_OUTPUT_TOKENS ) - response: HonchoLLMCallResponse[str] = await honcho_llm_call( - llm_settings=level_settings, - prompt="", # Ignored since we pass messages - max_tokens=max_tokens, - tools=tools, - tool_choice=level_settings.TOOL_CHOICE, - tool_executor=tool_executor, - max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, - messages=self.messages, - track_name="Dialectic Agent", - thinking_budget_tokens=level_settings.THINKING_BUDGET_TOKENS, + synthesis_response: HonchoLLMCallResponse[str] = await honcho_llm_call( + llm_settings=synthesis_settings, + prompt="", + max_tokens=synthesis_max_tokens, + tools=None, # No tools for synthesis + tool_executor=None, + max_tool_iterations=1, + messages=synthesis_messages, + track_name="Dialectic Synthesis", + thinking_budget_tokens=synthesis_settings.THINKING_BUDGET_TOKENS, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, - trace_name="dialectic_chat", + trace_name="dialectic_synthesis", ) - self._log_response_metrics( + # Log combined metrics for two-phase + self._log_two_phase_metrics( task_name=task_name, run_id=run_id, start_time=start_time, - response_content=response.content, - input_tokens=response.input_tokens, - output_tokens=response.output_tokens, - cache_read_input_tokens=response.cache_read_input_tokens, - cache_creation_input_tokens=response.cache_creation_input_tokens, - tool_calls_count=len(response.tool_calls_made), - thinking_content=response.thinking_content, - iterations=response.iterations, + response_content=synthesis_response.content, + search_input_tokens=search_response.input_tokens, + search_output_tokens=search_response.output_tokens, + search_cache_read_tokens=search_response.cache_read_input_tokens, + search_cache_creation_tokens=search_response.cache_creation_input_tokens, + search_tool_calls_count=len(search_response.tool_calls_made), + search_iterations=search_response.iterations, + synthesis_input_tokens=synthesis_response.input_tokens, + synthesis_output_tokens=synthesis_response.output_tokens, + synthesis_cache_read_tokens=synthesis_response.cache_read_input_tokens, + synthesis_cache_creation_tokens=synthesis_response.cache_creation_input_tokens, + synthesis_thinking_content=synthesis_response.thinking_content, ) - return response.content + return synthesis_response.content async def answer_stream(self, query: str) -> AsyncIterator[str]: """ Answer a query about the peer using agentic tool calling, streaming the response. + Supports two modes: + 1. Single-model mode: One model handles both tool calling and synthesis (streams final) + 2. Two-model mode: Search model handles tool calling, synthesis model streams response + The agent will: 1. Receive the query 2. Use tools to gather relevant context (non-streaming) @@ -454,6 +827,7 @@ class DialecticAgent: # Get level-specific settings level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level] + synthesis_settings = level_settings.SYNTHESIS # Use minimal tools for minimal reasoning to reduce cost tools = ( @@ -461,49 +835,181 @@ class DialecticAgent: if self.reasoning_level == "minimal" else DIALECTIC_TOOLS ) - # Use level-specific max_output_tokens if set, otherwise global default - max_tokens = ( - level_settings.MAX_OUTPUT_TOKENS - if level_settings.MAX_OUTPUT_TOKENS is not None - else settings.DIALECTIC.MAX_OUTPUT_TOKENS + + # Check if two-phase mode is enabled (synthesis config exists and not minimal) + use_two_phase = ( + synthesis_settings is not None and self.reasoning_level != "minimal" ) - response = cast( - StreamingResponseWithMetadata, - await honcho_llm_call( + if not use_two_phase: + # Single-model path (original behavior - stream final response) + max_tokens = ( + level_settings.MAX_OUTPUT_TOKENS + if level_settings.MAX_OUTPUT_TOKENS is not None + else settings.DIALECTIC.MAX_OUTPUT_TOKENS + ) + + response = cast( + StreamingResponseWithMetadata, + await honcho_llm_call( + llm_settings=level_settings, + prompt="", # Ignored since we pass messages + max_tokens=max_tokens, + stream=True, + stream_final_only=True, + tools=tools, + tool_choice=level_settings.TOOL_CHOICE, + tool_executor=tool_executor, + max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, + messages=self.messages, + track_name="Dialectic Agent Stream", + thinking_budget_tokens=level_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_chat", + ), + ) + + accumulated_content: list[str] = [] + async for chunk in response: + if chunk.content: + accumulated_content.append(chunk.content) + yield chunk.content + + self._log_response_metrics( + task_name=task_name, + run_id=run_id, + start_time=start_time, + response_content="".join(accumulated_content), + input_tokens=response.input_tokens, + output_tokens=response.output_tokens, + cache_read_input_tokens=response.cache_read_input_tokens, + cache_creation_input_tokens=response.cache_creation_input_tokens, + tool_calls_count=len(response.tool_calls_made), + thinking_content=response.thinking_content, + iterations=response.iterations, + ) + return + + # Two-phase mode: Search (non-streaming) then Synthesis (streaming) + # Type narrowing: synthesis_settings is guaranteed non-None in two-phase mode + assert synthesis_settings is not None # nosec B101 + + # Phase 1: Search (non-streaming, with tools) + search_max_tokens = ( + level_settings.MAX_OUTPUT_TOKENS + if level_settings.MAX_OUTPUT_TOKENS is not None + else 1024 # Lower default for search + ) + + try: + search_response: HonchoLLMCallResponse[str] = await honcho_llm_call( llm_settings=level_settings, - prompt="", # Ignored since we pass messages - max_tokens=max_tokens, - stream=True, - stream_final_only=True, + prompt="", + max_tokens=search_max_tokens, tools=tools, tool_choice=level_settings.TOOL_CHOICE, tool_executor=tool_executor, max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, messages=self.messages, - track_name="Dialectic Agent Stream", + track_name="Dialectic Search Stream", thinking_budget_tokens=level_settings.THINKING_BUDGET_TOKENS, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, - trace_name="dialectic_chat", + trace_name="dialectic_search", + ) + except Exception as e: + # Fallback to single-model streaming on search failure + logger.warning( + f"Search phase failed: {e}, falling back to single-model streaming" + ) + max_tokens = ( + synthesis_settings.MAX_OUTPUT_TOKENS + if synthesis_settings.MAX_OUTPUT_TOKENS is not None + else settings.DIALECTIC.MAX_OUTPUT_TOKENS + ) + fallback_response = cast( + StreamingResponseWithMetadata, + await honcho_llm_call( + llm_settings=synthesis_settings, + prompt="", + max_tokens=max_tokens, + stream=True, + tools=None, + tool_executor=None, + max_tool_iterations=1, + messages=self.messages, + track_name="Dialectic Fallback Stream", + thinking_budget_tokens=synthesis_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_fallback", + ), + ) + accumulated_content = [] + async for chunk in fallback_response: + if chunk.content: + accumulated_content.append(chunk.content) + yield chunk.content + self._log_response_metrics( + task_name=task_name, + run_id=run_id, + start_time=start_time, + response_content="".join(accumulated_content), + input_tokens=fallback_response.input_tokens, + output_tokens=fallback_response.output_tokens, + cache_read_input_tokens=fallback_response.cache_read_input_tokens, + cache_creation_input_tokens=fallback_response.cache_creation_input_tokens, + tool_calls_count=0, + thinking_content=fallback_response.thinking_content, + iterations=1, + ) + return + + # Phase 2: Synthesis (streaming, no tools) + synthesis_messages = self._build_synthesis_messages(search_response.messages) + synthesis_max_tokens = ( + synthesis_settings.MAX_OUTPUT_TOKENS + if synthesis_settings.MAX_OUTPUT_TOKENS is not None + else settings.DIALECTIC.MAX_OUTPUT_TOKENS + ) + + synthesis_stream = cast( + StreamingResponseWithMetadata, + await honcho_llm_call( + llm_settings=synthesis_settings, + prompt="", + max_tokens=synthesis_max_tokens, + stream=True, + tools=None, # No tools for synthesis + tool_executor=None, + max_tool_iterations=1, + messages=synthesis_messages, + track_name="Dialectic Synthesis Stream", + thinking_budget_tokens=synthesis_settings.THINKING_BUDGET_TOKENS, + max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, + trace_name="dialectic_synthesis", ), ) - accumulated_content: list[str] = [] - async for chunk in response: + accumulated_content = [] + async for chunk in synthesis_stream: if chunk.content: accumulated_content.append(chunk.content) yield chunk.content - self._log_response_metrics( + # Log combined metrics for two-phase + self._log_two_phase_metrics( task_name=task_name, run_id=run_id, start_time=start_time, response_content="".join(accumulated_content), - input_tokens=response.input_tokens, - output_tokens=response.output_tokens, - cache_read_input_tokens=response.cache_read_input_tokens, - cache_creation_input_tokens=response.cache_creation_input_tokens, - tool_calls_count=len(response.tool_calls_made), - thinking_content=response.thinking_content, - iterations=response.iterations, + search_input_tokens=search_response.input_tokens, + search_output_tokens=search_response.output_tokens, + search_cache_read_tokens=search_response.cache_read_input_tokens, + search_cache_creation_tokens=search_response.cache_creation_input_tokens, + search_tool_calls_count=len(search_response.tool_calls_made), + search_iterations=search_response.iterations, + synthesis_input_tokens=synthesis_stream.input_tokens, + synthesis_output_tokens=synthesis_stream.output_tokens, + synthesis_cache_read_tokens=synthesis_stream.cache_read_input_tokens, + synthesis_cache_creation_tokens=synthesis_stream.cache_creation_input_tokens, + synthesis_thinking_content=synthesis_stream.thinking_content, ) diff --git a/src/telemetry/events/__init__.py b/src/telemetry/events/__init__.py index 697ecec8..44083edc 100644 --- a/src/telemetry/events/__init__.py +++ b/src/telemetry/events/__init__.py @@ -62,7 +62,10 @@ from src.telemetry.events.agent import ( ) from src.telemetry.events.base import BaseEvent, generate_event_id from src.telemetry.events.deletion import DeletionCompletedEvent -from src.telemetry.events.dialectic import DialecticCompletedEvent +from src.telemetry.events.dialectic import ( + DialecticCompletedEvent, + DialecticPhaseMetrics, +) from src.telemetry.events.dream import ( DreamRunEvent, DreamSpecialistEvent, @@ -87,6 +90,7 @@ __all__ = [ "DreamSpecialistEvent", # Dialectic events "DialecticCompletedEvent", + "DialecticPhaseMetrics", # Agent events "AgentIterationEvent", "AgentToolConclusionsCreatedEvent", diff --git a/src/telemetry/events/dialectic.py b/src/telemetry/events/dialectic.py index 89ab9044..470f39a7 100644 --- a/src/telemetry/events/dialectic.py +++ b/src/telemetry/events/dialectic.py @@ -8,11 +8,33 @@ The run_id field enables correlation with agent.iteration events. from typing import ClassVar -from pydantic import Field +from pydantic import BaseModel, Field from src.telemetry.events.base import BaseEvent +class DialecticPhaseMetrics(BaseModel): + """Metrics for a single phase of dialectic execution. + + In two-phase mode, there are separate metrics for search and synthesis. + In single-model mode, there is one phase with combined metrics. + """ + + phase_name: str = Field( + ..., description="Phase identifier: 'single', 'search', or 'synthesis'" + ) + provider: str | None = Field(default=None, description="LLM provider used") + model: str | None = Field(default=None, description="Model name used") + input_tokens: int = Field(default=0, description="Input tokens for this phase") + output_tokens: int = Field(default=0, description="Output tokens for this phase") + cache_read_tokens: int = Field(default=0, description="Tokens read from cache") + cache_creation_tokens: int = Field(default=0, description="Tokens written to cache") + iterations: int = Field(default=1, description="LLM call iterations in this phase") + tool_calls_count: int = Field( + default=0, description="Tool calls made in this phase" + ) + + class DialecticCompletedEvent(BaseEvent): """Emitted when a dialectic (chat) query completes. @@ -20,12 +42,16 @@ class DialecticCompletedEvent(BaseEvent): from memory. This event captures the full context of the query and its execution metrics. + Supports both single-model and two-phase (search + synthesis) modes. + In two-phase mode, the `phases` list contains separate metrics for each phase. + In single-model mode, `phases` contains one entry with combined metrics. + The run_id correlates with AgentIterationEvent and AgentTool* events for detailed analytics. """ _event_type: ClassVar[str] = "dialectic.completed" - _schema_version: ClassVar[int] = 1 + _schema_version: ClassVar[int] = 2 # Bumped for phase metrics support _category: ClassVar[str] = "dialectic" # Run identification (for correlating with iteration/tool events) @@ -47,24 +73,40 @@ class DialecticCompletedEvent(BaseEvent): ..., description="Reasoning level: minimal, low, medium, high, max" ) - # Execution metrics - total_iterations: int = Field(default=1, description="Number of LLM iterations") + # Execution mode + two_phase_mode: bool = Field( + default=False, + description="Whether two-phase (search + synthesis) mode was used", + ) + + # Execution metrics (aggregated totals for backward compatibility) + total_iterations: int = Field( + default=1, description="Total LLM iterations across all phases" + ) prefetched_conclusion_count: int = Field( default=0, description="Number of conclusions prefetched" ) - tool_calls_count: int = Field(default=0, description="Number of tool calls made") + tool_calls_count: int = Field( + default=0, description="Total tool calls across all phases" + ) # Timing metrics (milliseconds) total_duration_ms: float = Field(..., description="Total processing time") - # Token usage with cache breakdown - input_tokens: int = Field(..., description="Total input tokens") - output_tokens: int = Field(..., description="Output tokens generated") + # Token usage with cache breakdown (aggregated totals) + input_tokens: int = Field(..., description="Total input tokens across all phases") + output_tokens: int = Field(..., description="Total output tokens across all phases") cache_read_tokens: int = Field( - default=0, description="Tokens read from prompt cache" + default=0, description="Total tokens read from prompt cache" ) cache_creation_tokens: int = Field( - default=0, description="Tokens written to prompt cache" + default=0, description="Total tokens written to prompt cache" + ) + + # Per-phase metrics (optional, for detailed cost analysis) + phases: list[DialecticPhaseMetrics] = Field( + default_factory=list, + description="Per-phase metrics. Empty for single-model mode, [search, synthesis] for two-phase", ) def get_resource_id(self) -> str: @@ -72,4 +114,4 @@ class DialecticCompletedEvent(BaseEvent): return self.run_id -__all__ = ["DialecticCompletedEvent"] +__all__ = ["DialecticCompletedEvent", "DialecticPhaseMetrics"] diff --git a/src/utils/clients.py b/src/utils/clients.py index 5ce820f7..4ebeab77 100644 --- a/src/utils/clients.py +++ b/src/utils/clients.py @@ -278,9 +278,14 @@ SELECTED_PROVIDERS = [ ("Deriver", settings.DERIVER.PROVIDER), ] -# Add all dialectic level providers +# Add all dialectic level providers (search and synthesis) for level, level_settings in settings.DIALECTIC.LEVELS.items(): SELECTED_PROVIDERS.append((f"Dialectic ({level})", level_settings.PROVIDER)) + # Also validate synthesis provider if configured + if level_settings.SYNTHESIS is not None: + SELECTED_PROVIDERS.append( + (f"Dialectic ({level}) Synthesis", level_settings.SYNTHESIS.PROVIDER) + ) for provider_name, provider_value in SELECTED_PROVIDERS: if provider_value not in CLIENTS: @@ -293,9 +298,14 @@ BACKUP_PROVIDERS: list[tuple[str, SupportedProviders | None]] = [ ("Dream", settings.DREAM.BACKUP_PROVIDER), ] -# Add all dialectic level backup providers +# Add all dialectic level backup providers (search and synthesis) for level, level_settings in settings.DIALECTIC.LEVELS.items(): BACKUP_PROVIDERS.append((f"Dialectic ({level})", level_settings.BACKUP_PROVIDER)) + # Also validate synthesis backup provider if configured + if level_settings.SYNTHESIS is not None: + BACKUP_PROVIDERS.append( + (f"Dialectic ({level}) Synthesis", level_settings.SYNTHESIS.BACKUP_PROVIDER) + ) for component_name, backup_provider in BACKUP_PROVIDERS: if backup_provider is not None and backup_provider not in CLIENTS: @@ -477,6 +487,7 @@ class HonchoLLMCallResponse(BaseModel, Generic[T]): cache_read_input_tokens: Number of tokens read from cache. finish_reasons: List of finish reasons for the response. tool_calls_made: Optional list of all tool calls executed during the request. + messages: Full conversation history including tool calls and results (for two-phase dialectic). Note: Uncached input tokens = input_tokens - cache_read_input_tokens + cache_creation_input_tokens @@ -497,6 +508,8 @@ class HonchoLLMCallResponse(BaseModel, Generic[T]): thinking_blocks: list[dict[str, Any]] = Field(default_factory=list) # OpenRouter reasoning_details for Gemini models - must be preserved across turns reasoning_details: list[dict[str, Any]] = Field(default_factory=list) + # Full conversation history for two-phase dialectic (search -> synthesis) + messages: list[dict[str, Any]] = Field(default_factory=list) class HonchoLLMCallStreamChunk(BaseModel): @@ -532,6 +545,7 @@ class StreamingResponseWithMetadata: cache_read_input_tokens: int thinking_content: str | None iterations: int + messages: list[dict[str, Any]] def __init__( self, @@ -543,6 +557,7 @@ class StreamingResponseWithMetadata: cache_read_input_tokens: int, thinking_content: str | None = None, iterations: int = 0, + messages: list[dict[str, Any]] | None = None, ): self._stream = stream self.tool_calls_made = tool_calls_made @@ -552,6 +567,7 @@ class StreamingResponseWithMetadata: self.cache_read_input_tokens = cache_read_input_tokens self.thinking_content = thinking_content self.iterations = iterations + self.messages = messages or [] def __aiter__(self) -> AsyncIterator[HonchoLLMCallStreamChunk]: return self._stream.__aiter__() @@ -802,6 +818,7 @@ async def _execute_tool_loop( cache_read_input_tokens=total_cache_read_tokens, thinking_content=response.thinking_content, iterations=iteration + 1, + messages=conversation_messages, ) response.tool_calls_made = all_tool_calls @@ -810,6 +827,7 @@ async def _execute_tool_loop( response.cache_creation_input_tokens = total_cache_creation_tokens response.cache_read_input_tokens = total_cache_read_tokens response.iterations = iteration + 1 + response.messages = conversation_messages return response # Determine which provider we're using (reuse the helper) @@ -935,6 +953,7 @@ async def _execute_tool_loop( cache_read_input_tokens=total_cache_read_tokens, thinking_content=None, # No thinking content at max iterations iterations=iteration + 1, # +1 for the synthesis call + messages=conversation_messages, ) # Make one final call to get a text response @@ -988,6 +1007,7 @@ async def _execute_tool_loop( final_response.cache_read_input_tokens = ( total_cache_read_tokens + final_response.cache_read_input_tokens ) + final_response.messages = conversation_messages return final_response @@ -1409,6 +1429,7 @@ async def honcho_llm_call( True, # type: ignore[arg-type] converted_tools, tool_choice, + messages, ) else: return await honcho_llm_call_inner( @@ -1426,6 +1447,7 @@ async def honcho_llm_call( False, # type: ignore[arg-type] converted_tools, tool_choice, + messages, ) decorated = _call_with_provider_selection diff --git a/tests/telemetry/test_events.py b/tests/telemetry/test_events.py index 94f34aaa..54189abe 100644 --- a/tests/telemetry/test_events.py +++ b/tests/telemetry/test_events.py @@ -264,7 +264,7 @@ class TestDialecticCompletedEvent: def test_schema_version(self): """schema_version() returns correct value.""" - assert DialecticCompletedEvent.schema_version() == 1 + assert DialecticCompletedEvent.schema_version() == 2 def test_category(self): """category() returns correct value."""