RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: enhance dialectic cost calculator with two-phase support 

- Introduced a new two-phase cost calculation method for dialectic reasoning, allowing separate search and synthesis phases.
- Updated the dialectic configuration to support synthesis model settings, enabling more flexible model usage.
- Enhanced the `DialecticAgent` to handle both single-model and two-phase modes, improving cost estimation accuracy.
- Added telemetry metrics for detailed analysis of search and synthesis phases, including input/output token tracking and tool call counts.
- Updated related documentation and usage notes to reflect the new capabilities and configurations.
This commit is contained in:
Benjamin McCormick 2026-01-22 15:43:27 -05:00
parent dce96889bc
commit 9cd21ebd72
7 changed files with 1095 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

408
scripts/dialectic_cost_calculator.py
View File

@ -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"
)

93
src/config.py
View File

@ -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,
),
),
}
)

608
src/dialectic/core.py
View File

@ -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,
)

6
src/telemetry/events/__init__.py
View File

@ -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",

64
src/telemetry/events/dialectic.py
View File

@ -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"]

26
src/utils/clients.py
View File

@ -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

2
tests/telemetry/test_events.py
View File

@ -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."""