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

feat: refactor LLM client layer and update provider identifiers 

- Refactored the LLM client layer from a monolithic structure in `clients.py` to an adapter-based architecture in the new `src/utils/llm/` package, enhancing modularity and maintainability.
- Renamed the provider identifier from `"custom"` to `"openrouter"` for improved clarity in configuration settings.
- Updated relevant files to reflect the new structure and naming conventions, ensuring consistency across the codebase.
- Enhanced the changelog to document these significant changes.
This commit is contained in:
Benjamin McCormick 2026-01-22 17:30:10 -05:00
parent 9cd21ebd72
commit 9d98bb9f4b
27 changed files with 3449 additions and 2548 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
- API route renaming for consistency
- Dreamer and dialectic now respect peer card configuration settings
- Observations renamed to Conclusions across API and SDKs
- LLM client layer refactored from monolithic `clients.py` to adapter-based `src/utils/llm/` package
- Provider identifier `"custom"` renamed to `"openrouter"` for clarity
### Fixed

4
src/config.py
View File

@ -422,13 +422,13 @@ class DialecticSettings(HonchoSettings):
TOOL_CHOICE="any",
),
"low": DialecticLevelSettings(
PROVIDER="custom",
PROVIDER="openrouter",
MODEL="z-ai/glm-4.7-flash",
THINKING_BUDGET_TOKENS=0,
MAX_TOOL_ITERATIONS=4,
),
"medium": DialecticLevelSettings(
PROVIDER="custom",
PROVIDER="openrouter",
MODEL="z-ai/glm-4.7-flash",
THINKING_BUDGET_TOKENS=0,
MAX_TOOL_ITERATIONS=4,

2
src/deriver/deriver.py
View File

@ -12,9 +12,9 @@ from src.telemetry.events import RepresentationCompletedEvent, emit
from src.telemetry.logging import accumulate_metric, log_performance_metrics
from src.telemetry.otel.metrics import DeriverComponents, DeriverTaskTypes, TokenTypes
from src.telemetry.sentry import with_sentry_transaction
from src.utils.clients import honcho_llm_call
from src.utils.config_helpers import get_configuration
from src.utils.formatting import format_new_turn_with_timestamp
from src.utils.llm import honcho_llm_call
from src.utils.representation import PromptRepresentation, Representation
from src.utils.tokens import estimate_tokens, track_deriver_input_tokens

4
src/dialectic/core.py
View File

@ -31,12 +31,12 @@ from src.utils.agent_tools import (
create_tool_executor,
search_memory,
)
from src.utils.clients import (
from src.utils.formatting import format_new_turn_with_timestamp
from src.utils.llm import (
HonchoLLMCallResponse,
StreamingResponseWithMetadata,
honcho_llm_call,
)
from src.utils.formatting import format_new_turn_with_timestamp
logger = logging.getLogger(__name__)

2
src/dreamer/specialists.py
View File

@ -31,7 +31,7 @@ from src.utils.agent_tools import (
INDUCTION_SPECIALIST_TOOLS,
create_tool_executor,
)
from src.utils.clients import HonchoLLMCallResponse, honcho_llm_call
from src.utils.llm import HonchoLLMCallResponse, honcho_llm_call
logger = logging.getLogger(__name__)

2529
src/utils/clients.py
View File

File diff suppressed because it is too large Load Diff

35
src/utils/llm/__init__.py Normal file
View File

@ -0,0 +1,35 @@
"""
Honcho LLM client layer.
This package provides a provider-agnostic API (`honcho_llm_call`) backed by
provider adapters that encapsulate the idiosyncrasies of each upstream SDK.
"""
from src.utils.llm.core import (
handle_streaming_response,
honcho_llm_call,
honcho_llm_call_inner,
)
from src.utils.llm.history import count_message_tokens, truncate_messages_to_fit
from src.utils.llm.models import (
HonchoLLMCallResponse,
HonchoLLMCallStreamChunk,
IterationCallback,
IterationData,
StreamingResponseWithMetadata,
)
from src.utils.llm.registry import CLIENTS
__all__ = [
"CLIENTS",
"IterationCallback",
"IterationData",
"HonchoLLMCallResponse",
"HonchoLLMCallStreamChunk",
"StreamingResponseWithMetadata",
"count_message_tokens",
"truncate_messages_to_fit",
"honcho_llm_call",
"honcho_llm_call_inner",
"handle_streaming_response",
]

54
src/utils/llm/adapters/__init__.py Normal file
View File

@ -0,0 +1,54 @@
"""
Provider adapter registry.
This module exposes a single `get_adapter()` function that returns the adapter
implementation for a given provider identifier. Adapters are cached since they
are stateless.
"""
from __future__ import annotations
from src.utils.types import SupportedProviders
from .base import ProviderAdapter
_ADAPTERS: dict[str, ProviderAdapter] = {}
def _create_adapter(provider: str) -> ProviderAdapter:
"""Create a new adapter instance for the given provider."""
if provider == "anthropic":
from src.utils.llm.adapters.anthropic import AnthropicAdapter
return AnthropicAdapter()
if provider == "openai":
from src.utils.llm.adapters.openai import OpenAIAdapter
return OpenAIAdapter()
if provider == "openrouter":
from src.utils.llm.adapters.openrouter import OpenRouterAdapter
return OpenRouterAdapter()
if provider == "vllm":
from src.utils.llm.adapters.vllm import VLLMAdapter
return VLLMAdapter()
if provider == "google":
from src.utils.llm.adapters.google import GoogleAdapter
return GoogleAdapter()
if provider == "groq":
from src.utils.llm.adapters.groq import GroqAdapter
return GroqAdapter()
raise ValueError(f"Unsupported provider: {provider}")
def get_adapter(provider: SupportedProviders | str) -> ProviderAdapter:
"""Return a cached ProviderAdapter implementation for `provider`."""
if provider not in _ADAPTERS:
_ADAPTERS[provider] = _create_adapter(provider)
return _ADAPTERS[provider]
__all__ = ["ProviderAdapter", "get_adapter"]

322
src/utils/llm/adapters/anthropic.py Normal file
View File

@ -0,0 +1,322 @@
"""
Anthropic provider adapter.
This adapter encapsulates all Anthropic-specific behaviors:
- system messages passed as a top-level `system` parameter with cache_control
- extended thinking blocks (including signatures) and tool-use blocks
- JSON mode / response_model via prompt instructions and a prefixed `{`
"""
from __future__ import annotations
import json
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from anthropic import AsyncAnthropic
from anthropic.types import TextBlock, ThinkingBlock, ToolUseBlock
from anthropic.types.message import Message as AnthropicMessage
from anthropic.types.usage import Usage
from pydantic import BaseModel, ValidationError
from src.utils.llm.adapters.base import ProviderAdapter
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class AnthropicAdapter(ProviderAdapter):
"""ProviderAdapter implementation for Anthropic SDK clients."""
provider: SupportedProviders = "anthropic"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return tools unchanged (Anthropic-native schema)."""
return tools
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant tool-use message for Anthropic."""
_ = reasoning_details
content_blocks: list[dict[str, Any]] = []
if thinking_blocks:
content_blocks.extend(thinking_blocks)
if isinstance(content, str) and content:
content_blocks.append({"type": "text", "text": content})
for tool_call in tool_calls:
content_blocks.append(
{
"type": "tool_use",
"id": tool_call["id"],
"name": tool_call["name"],
"input": tool_call["input"],
}
)
return {"role": "assistant", "content": content_blocks}
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results as Anthropic tool_result blocks."""
result_blocks: list[dict[str, Any]] = []
for tr in tool_results:
result_blocks.append(
{
"type": "tool_result",
"tool_use_id": tr["tool_id"],
"content": str(tr["result"]),
"is_error": tr.get("is_error", False),
}
)
conversation_messages.append({"role": "user", "content": result_blocks})
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming Anthropic call and normalize the response."""
_ = (provider, prompt, stop_seqs, reasoning_effort, verbosity)
system_messages: list[str] = []
non_system_messages: list[dict[str, Any]] = []
for msg in messages:
if msg.get("role") == "system":
system_messages.append(cast(str, msg["content"]))
else:
non_system_messages.append(msg)
anthropic_params: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": non_system_messages,
}
if temperature is not None:
anthropic_params["temperature"] = temperature
if system_messages:
anthropic_params["system"] = [
{
"type": "text",
"text": "\n\n".join(system_messages),
"cache_control": {"type": "ephemeral"},
}
]
if tools:
anthropic_params["tools"] = tools
if tool_choice:
if isinstance(tool_choice, str):
if tool_choice == "auto":
anthropic_params["tool_choice"] = {"type": "auto"}
elif tool_choice in ("any", "required"):
anthropic_params["tool_choice"] = {"type": "any"}
elif tool_choice == "none":
pass
else:
anthropic_params["tool_choice"] = {
"type": "tool",
"name": tool_choice,
}
else:
anthropic_params["tool_choice"] = tool_choice
if response_model or json_mode:
if response_model:
schema_json = json.dumps(response_model.model_json_schema(), indent=2)
anthropic_params["messages"][-1]["content"] += (
f"\n\nRespond with valid JSON matching this schema:\n{schema_json}"
)
anthropic_params["messages"].append({"role": "assistant", "content": "{"})
if thinking_budget_tokens:
anthropic_params["thinking"] = {
"type": "enabled",
"budget_tokens": thinking_budget_tokens,
}
anthropic_client = cast(AsyncAnthropic, client)
anthropic_response: AnthropicMessage = cast(
AnthropicMessage, await anthropic_client.messages.create(**anthropic_params)
)
text_blocks: list[str] = []
thinking_text_blocks: list[str] = []
thinking_full_blocks: list[dict[str, Any]] = []
tool_calls: list[dict[str, Any]] = []
for block in anthropic_response.content:
if isinstance(block, TextBlock):
text_blocks.append(block.text)
elif isinstance(block, ThinkingBlock):
thinking_text_blocks.append(block.thinking)
thinking_full_blocks.append(
{
"type": "thinking",
"thinking": block.thinking,
"signature": block.signature,
}
)
elif isinstance(block, ToolUseBlock):
tool_calls.append(
{
"id": block.id,
"name": block.name,
"input": block.input,
}
)
usage: Any | Usage = anthropic_response.usage
stop_reason = anthropic_response.stop_reason
text_content = "\n".join(text_blocks)
thinking_content = (
"\n".join(thinking_text_blocks) if thinking_text_blocks else None
)
cache_creation_tokens = (
getattr(usage, "cache_creation_input_tokens", 0) or 0 if usage else 0
)
cache_read_tokens = (
getattr(usage, "cache_read_input_tokens", 0) or 0 if usage else 0
)
uncached_tokens = usage.input_tokens if usage else 0
total_input_tokens = uncached_tokens + cache_read_tokens + cache_creation_tokens
if response_model:
try:
json_content = "{" + text_content
parsed_json = json.loads(json_content)
parsed_content = response_model.model_validate(parsed_json)
return HonchoLLMCallResponse(
content=parsed_content,
input_tokens=total_input_tokens,
output_tokens=usage.output_tokens if usage else 0,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
finish_reasons=[stop_reason] if stop_reason else [],
tool_calls_made=tool_calls,
thinking_content=thinking_content,
thinking_blocks=thinking_full_blocks,
)
except (json.JSONDecodeError, ValidationError, ValueError) as e:
raise ValueError(
f"Failed to parse Anthropic response as {response_model}: {e}. Raw content: {text_content}"
) from e
return HonchoLLMCallResponse(
content=text_content,
input_tokens=total_input_tokens,
output_tokens=usage.output_tokens if usage else 0,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
finish_reasons=[stop_reason] if stop_reason else [],
tool_calls_made=tool_calls,
thinking_content=thinking_content,
thinking_blocks=thinking_full_blocks,
)
async def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream an Anthropic response and normalize chunks."""
_ = (provider, prompt, temperature, reasoning_effort, verbosity)
system_content = "\n\n".join(
m["content"] for m in messages if m.get("role") == "system"
)
anthropic_params: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": [m for m in messages if m.get("role") != "system"],
}
if system_content:
anthropic_params["system"] = [
{
"type": "text",
"text": system_content,
"cache_control": {"type": "ephemeral"},
}
]
if response_model or json_mode:
if response_model:
schema_json = json.dumps(response_model.model_json_schema(), indent=2)
anthropic_params["messages"][-1]["content"] += (
f"\n\nRespond with valid JSON matching this schema:\n{schema_json}"
)
anthropic_params["messages"].append({"role": "assistant", "content": "{"})
if thinking_budget_tokens:
anthropic_params["thinking"] = {
"type": "enabled",
"budget_tokens": thinking_budget_tokens,
}
anthropic_client = cast(AsyncAnthropic, client)
async with anthropic_client.messages.stream(
**anthropic_params
) as anthropic_stream:
async for chunk in anthropic_stream:
if (
chunk.type == "content_block_delta"
and hasattr(chunk, "delta")
and hasattr(chunk.delta, "text")
):
text_content = getattr(chunk.delta, "text", "")
yield HonchoLLMCallStreamChunk(content=text_content)
final_message = await anthropic_stream.get_final_message()
usage = final_message.usage
output_tokens = usage.output_tokens if usage else None
yield HonchoLLMCallStreamChunk(
content="",
is_done=True,
finish_reasons=[final_message.stop_reason]
if final_message.stop_reason
else [],
output_tokens=output_tokens,
)

122
src/utils/llm/adapters/base.py Normal file
View File

@ -0,0 +1,122 @@
"""
Provider adapter interfaces for the Honcho LLM layer.
Adapters encapsulate provider-specific request shaping, response parsing, and
tool-calling message formats. The orchestrator (retry/failover/tool-loop) stays
provider-agnostic by delegating those details to an adapter.
"""
from __future__ import annotations
import json
from collections.abc import AsyncIterator
from typing import Any, Protocol, runtime_checkable
from pydantic import BaseModel
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
@runtime_checkable
class ProviderAdapter(Protocol):
"""Interface implemented by per-provider adapters."""
provider: SupportedProviders
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""
Perform a non-streaming LLM call and return a normalized response.
Implementations must preserve all provider idiosyncrasies currently relied
upon by callers (token accounting, tool call parsing, reasoning extraction,
structured output handling, etc.).
"""
...
def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Perform a streaming LLM call and yield normalized stream chunks."""
...
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""
Convert tool definitions to the provider-specific schema.
The input is in the Anthropic-style schema used by Honcho tool definitions.
"""
...
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant message containing tool calls for this provider."""
...
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results to `conversation_messages` in provider format."""
...
def openai_tool_calls_from_tool_calls(tool_calls: list[dict[str, Any]]) -> list[Any]:
"""
Convert normalized Honcho tool calls to OpenAI tool_calls entries.
This helper is shared by OpenAI-compatible adapters when formatting
multi-turn tool calling messages.
"""
openai_tool_calls: list[Any] = []
for tool_call in tool_calls:
openai_tool_calls.append(
{
"id": tool_call["id"],
"type": "function",
"function": {
"name": tool_call["name"],
"arguments": json.dumps(tool_call["input"]),
},
}
)
return openai_tool_calls

375
src/utils/llm/adapters/google.py Normal file
View File

@ -0,0 +1,375 @@
"""
Google Gemini provider adapter.
This adapter encapsulates Gemini-specific behaviors:
- `system_instruction` is used instead of system-role messages in `contents`
- tool calling uses `function_call`/`function_response` parts
- `thought_signature` must be preserved and replayed for multi-turn tool use
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from google import genai
from google.genai.types import (
ContentListUnionDict,
GenerateContentConfigDict,
GenerateContentResponse,
)
from pydantic import BaseModel
from src.utils.llm.adapters.base import ProviderAdapter
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class GoogleAdapter(ProviderAdapter):
"""ProviderAdapter implementation for Google Gemini SDK clients."""
provider: SupportedProviders = "google"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Convert Anthropic-style tool schemas to Gemini function_declarations."""
return [
{
"function_declarations": [
{
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
}
for tool in tools
]
}
]
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format a Gemini model message containing function calls."""
_ = (thinking_blocks, reasoning_details)
parts: list[dict[str, Any]] = []
if isinstance(content, str) and content:
parts.append({"text": content})
for tool_call in tool_calls:
part_data: dict[str, Any] = {
"function_call": {"name": tool_call["name"], "args": tool_call["input"]}
}
if "thought_signature" in tool_call:
part_data["thought_signature"] = tool_call["thought_signature"]
parts.append(part_data)
return {"role": "model", "parts": parts}
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results in Gemini `function_response` part format."""
response_parts: list[dict[str, Any]] = []
for tr in tool_results:
response_parts.append(
{
"function_response": {
"name": tr["tool_name"],
"response": {"result": str(tr["result"])},
}
}
)
conversation_messages.append({"role": "user", "parts": response_parts})
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming Gemini call and normalize the response."""
_ = (
provider,
max_tokens,
stop_seqs,
reasoning_effort,
verbosity,
thinking_budget_tokens,
)
gemini_client = cast(genai.Client, client)
system_messages: list[str] = []
non_system_messages: list[dict[str, Any]] = []
gemini_config: dict[str, Any] = {}
if temperature is not None:
gemini_config["temperature"] = temperature
if tools:
gemini_config["tools"] = tools
if tool_choice:
if tool_choice == "auto":
gemini_config["tool_config"] = {
"function_calling_config": {"mode": "AUTO"}
}
elif tool_choice == "any" or tool_choice == "required":
gemini_config["tool_config"] = {
"function_calling_config": {"mode": "ANY"}
}
elif tool_choice == "none":
gemini_config["tool_config"] = {
"function_calling_config": {"mode": "NONE"}
}
elif isinstance(tool_choice, dict) and "name" in tool_choice:
gemini_config["tool_config"] = {
"function_calling_config": {
"mode": "ANY",
"allowed_function_names": [tool_choice["name"]],
}
}
if response_model is None:
if json_mode and not tools:
gemini_config["response_mime_type"] = "application/json"
if messages:
for msg in messages:
if msg.get("role") == "system":
if isinstance(msg.get("content"), str):
system_messages.append(msg["content"])
else:
non_system_messages.append(msg)
if system_messages:
gemini_config["system_instruction"] = "\n\n".join(system_messages)
gemini_contents: list[dict[str, Any]] = []
for msg in non_system_messages:
role = msg.get("role", "user")
if role == "assistant":
role = "model"
if isinstance(msg.get("content"), str):
gemini_contents.append(
{"role": role, "parts": [{"text": msg["content"]}]}
)
elif isinstance(msg.get("parts"), list):
msg_copy = msg.copy()
msg_copy["role"] = role
gemini_contents.append(msg_copy)
elif isinstance(msg.get("content"), list):
continue
else:
continue
contents: ContentListUnionDict = cast(
ContentListUnionDict, gemini_contents
)
else:
contents = prompt
gemini_response: GenerateContentResponse = (
await gemini_client.aio.models.generate_content(
model=model,
contents=contents,
config=cast(GenerateContentConfigDict, cast(object, gemini_config))
if gemini_config
else None,
)
)
text_parts: list[str] = []
gemini_tool_calls: list[dict[str, Any]] = []
if gemini_response.candidates and gemini_response.candidates[0].content:
for part in gemini_response.candidates[0].content.parts or []:
if hasattr(part, "text") and part.text:
text_parts.append(part.text)
if hasattr(part, "function_call") and part.function_call:
fc = part.function_call
tool_call_data: dict[str, Any] = {
"id": f"call_{fc.name}_{len(gemini_tool_calls)}",
"name": fc.name,
"input": dict(fc.args) if fc.args else {},
}
if (
hasattr(part, "thought_signature")
and part.thought_signature
):
tool_call_data["thought_signature"] = part.thought_signature
gemini_tool_calls.append(tool_call_data)
text_content = "\n".join(text_parts) if text_parts else ""
input_token_count = (
gemini_response.usage_metadata.prompt_token_count or 0
if gemini_response.usage_metadata
else 0
)
output_token_count = (
gemini_response.usage_metadata.candidates_token_count or 0
if gemini_response.usage_metadata
else 0
)
finish_reason = (
gemini_response.candidates[0].finish_reason.name
if gemini_response.candidates
and gemini_response.candidates[0].finish_reason
else "stop"
)
return HonchoLLMCallResponse(
content=text_content,
input_tokens=input_token_count,
output_tokens=output_token_count,
finish_reasons=[finish_reason],
tool_calls_made=gemini_tool_calls,
)
gemini_config["response_mime_type"] = "application/json"
gemini_config["response_schema"] = response_model
gemini_response = await gemini_client.aio.models.generate_content(
model=model,
contents=prompt,
config=cast(GenerateContentConfigDict, cast(object, gemini_config)),
)
input_token_count = (
gemini_response.usage_metadata.prompt_token_count or 0
if gemini_response.usage_metadata
else 0
)
output_token_count = (
gemini_response.usage_metadata.candidates_token_count or 0
if gemini_response.usage_metadata
else 0
)
finish_reason = (
gemini_response.candidates[0].finish_reason.name
if gemini_response.candidates
and gemini_response.candidates[0].finish_reason
else "stop"
)
if not isinstance(gemini_response.parsed, response_model):
raise ValueError(
f"Parsed content does not match the response model: {gemini_response.parsed} != {response_model}"
)
return HonchoLLMCallResponse(
content=gemini_response.parsed,
input_tokens=input_token_count,
output_tokens=output_token_count,
finish_reasons=[finish_reason],
tool_calls_made=[],
)
async def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream a Gemini response and normalize chunks."""
_ = (
provider,
prompt,
max_tokens,
temperature,
reasoning_effort,
verbosity,
thinking_budget_tokens,
)
gemini_client = cast(genai.Client, client)
prompt_text = messages[0]["content"] if messages else ""
if response_model is not None:
response_stream = await gemini_client.aio.models.generate_content_stream(
model=model,
contents=prompt_text,
config={
"response_mime_type": "application/json",
"response_schema": response_model,
},
)
else:
response_stream = await gemini_client.aio.models.generate_content_stream(
model=model,
contents=prompt_text,
config={
"response_mime_type": "application/json" if json_mode else None
},
)
final_chunk = None
async for chunk in response_stream:
if chunk.text:
yield HonchoLLMCallStreamChunk(content=chunk.text)
final_chunk = chunk
finish_reason = "stop"
gemini_output_tokens: int | None = None
if (
final_chunk
and hasattr(final_chunk, "candidates")
and final_chunk.candidates
and hasattr(final_chunk.candidates[0], "finish_reason")
and final_chunk.candidates[0].finish_reason
):
finish_reason = final_chunk.candidates[0].finish_reason.name
if (
final_chunk
and hasattr(final_chunk, "usage_metadata")
and final_chunk.usage_metadata
and hasattr(final_chunk.usage_metadata, "candidates_token_count")
):
gemini_output_tokens = (
final_chunk.usage_metadata.candidates_token_count or None
)
yield HonchoLLMCallStreamChunk(
content="",
is_done=True,
finish_reasons=[finish_reason],
output_tokens=gemini_output_tokens,
)

204
src/utils/llm/adapters/groq.py Normal file
View File

@ -0,0 +1,204 @@
"""
Groq provider adapter.
Groq's API is OpenAI-chat compatible for basic completions and streaming. Honcho
currently uses it without tool calling.
"""
from __future__ import annotations
import json
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from groq import AsyncGroq
from openai.types.chat import ChatCompletion, ChatCompletionChunk
from pydantic import BaseModel, ValidationError
from src.utils.llm.adapters.base import (
ProviderAdapter,
openai_tool_calls_from_tool_calls,
)
from src.utils.llm.adapters.openai_common import extract_openai_cache_tokens
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class GroqAdapter(ProviderAdapter):
"""ProviderAdapter implementation for the Groq SDK client."""
provider: SupportedProviders = "groq"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return tools unchanged; tool calling is not implemented for Groq."""
logger.warning(
"Tool calling not implemented for provider groq, returning tools as-is"
)
return tools
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant message with tool calls in OpenAI format."""
_ = (thinking_blocks, reasoning_details)
return {
"role": "assistant",
"content": content if isinstance(content, str) else None,
"tool_calls": openai_tool_calls_from_tool_calls(tool_calls),
}
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results in OpenAI tool-message format."""
for tr in tool_results:
conversation_messages.append(
{
"role": "tool",
"tool_call_id": tr["tool_id"],
"content": str(tr["result"]),
}
)
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming Groq call and normalize the response."""
_ = (
provider,
prompt,
stop_seqs,
reasoning_effort,
verbosity,
thinking_budget_tokens,
tools,
tool_choice,
)
groq_client = cast(AsyncGroq, client)
groq_params: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": messages,
}
if temperature is not None:
groq_params["temperature"] = temperature
if response_model:
groq_params["response_format"] = response_model
elif json_mode:
groq_params["response_format"] = {"type": "json_object"}
response = cast(
ChatCompletion, await groq_client.chat.completions.create(**groq_params)
)
if response.choices[0].message.content is None:
raise ValueError("No content in response")
usage = response.usage
finish_reason = response.choices[0].finish_reason
cache_creation, cache_read = extract_openai_cache_tokens(usage)
if response_model:
try:
json_content = json.loads(response.choices[0].message.content)
parsed_content = response_model.model_validate(json_content)
return HonchoLLMCallResponse(
content=parsed_content,
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=[],
)
except (json.JSONDecodeError, ValidationError, ValueError) as e:
raise ValueError(
f"Failed to parse Groq response as {response_model}: {e}. Raw content: {response.choices[0].message.content}"
) from e
return HonchoLLMCallResponse(
content=response.choices[0].message.content,
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=[],
)
async def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream a Groq response and normalize chunks."""
_ = (provider, prompt, reasoning_effort, verbosity, thinking_budget_tokens)
groq_client = cast(AsyncGroq, client)
groq_params: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": messages,
"stream": True,
}
if response_model:
groq_params["response_format"] = response_model
elif json_mode:
groq_params["response_format"] = {"type": "json_object"}
groq_stream = cast(
AsyncIterator[ChatCompletionChunk],
await groq_client.chat.completions.create(**groq_params),
)
async for chunk in groq_stream:
if chunk.choices and chunk.choices[0].delta.content:
yield HonchoLLMCallStreamChunk(content=chunk.choices[0].delta.content)
if chunk.choices and chunk.choices[0].finish_reason:
yield HonchoLLMCallStreamChunk(
content="",
is_done=True,
finish_reasons=[chunk.choices[0].finish_reason],
)

219
src/utils/llm/adapters/openai.py Normal file
View File

@ -0,0 +1,219 @@
"""
OpenAI provider adapter.
This adapter implements the OpenAI-native provider (`openai`). OpenAI-compatible
providers with special quirks (OpenRouter, vLLM) have their own adapters.
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletion
from pydantic import BaseModel
from src.utils.llm.adapters.base import (
ProviderAdapter,
openai_tool_calls_from_tool_calls,
)
from src.utils.llm.adapters.openai_common import (
extract_openai_cache_tokens,
extract_openai_reasoning_content,
extract_openai_reasoning_details,
extract_openai_tool_calls,
stream_openai_compatible,
)
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class OpenAIAdapter(ProviderAdapter):
"""ProviderAdapter implementation for OpenAI's native API."""
provider: SupportedProviders = "openai"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Convert Anthropic-style tool schemas to OpenAI tool definitions."""
return [
{
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
},
}
for tool in tools
]
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant message with tool calls for OpenAI-compatible APIs."""
_ = thinking_blocks
msg: dict[str, Any] = {
"role": "assistant",
"content": content if isinstance(content, str) else None,
"tool_calls": openai_tool_calls_from_tool_calls(tool_calls),
}
if reasoning_details:
msg["reasoning_details"] = reasoning_details
return msg
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results in OpenAI tool-message format."""
for tr in tool_results:
conversation_messages.append(
{
"role": "tool",
"tool_call_id": tr["tool_id"],
"content": str(tr["result"]),
}
)
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming OpenAI call and normalize the response."""
_ = (prompt, thinking_budget_tokens)
openai_params: dict[str, Any] = {"model": model, "messages": messages}
if temperature is not None and "gpt-5" not in model:
openai_params["temperature"] = temperature
if "gpt-5" in model:
openai_params["max_completion_tokens"] = max_tokens
if reasoning_effort:
openai_params["reasoning_effort"] = reasoning_effort
if verbosity:
openai_params["verbosity"] = verbosity
else:
openai_params["max_tokens"] = max_tokens
if tools and not response_model:
openai_params["tools"] = tools
if tool_choice:
openai_params["tool_choice"] = tool_choice
if json_mode:
openai_params["response_format"] = {"type": "json_object"}
openai_client = cast(AsyncOpenAI, client)
if response_model:
openai_params["response_format"] = response_model
response = cast(
ChatCompletion,
await openai_client.chat.completions.parse(**openai_params),
)
message_any = cast(Any, response.choices[0].message)
parsed_content = getattr(message_any, "parsed", None)
if parsed_content is None:
raise ValueError("No parsed content in structured response")
usage = response.usage
finish_reason = response.choices[0].finish_reason
if not isinstance(parsed_content, response_model):
raise ValueError(
f"Parsed content does not match the response model: {parsed_content} != {response_model}"
)
parsed_tool_calls = extract_openai_tool_calls(response.choices[0].message)
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=parsed_content,
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=parsed_tool_calls,
thinking_content=extract_openai_reasoning_content(response),
reasoning_details=extract_openai_reasoning_details(response),
)
response = cast(
ChatCompletion, await openai_client.chat.completions.create(**openai_params)
)
usage = response.usage
finish_reason = response.choices[0].finish_reason
tool_calls_list = extract_openai_tool_calls(response.choices[0].message)
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=response.choices[0].message.content or "",
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=tool_calls_list,
thinking_content=extract_openai_reasoning_content(response),
reasoning_details=extract_openai_reasoning_details(response),
)
def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream an OpenAI response and normalize chunks."""
_ = (provider, prompt, temperature, thinking_budget_tokens)
openai_client = cast(AsyncOpenAI, client)
return stream_openai_compatible(
client=openai_client,
model=model,
max_tokens=max_tokens,
messages=messages,
response_model=response_model,
json_mode=json_mode,
reasoning_effort=cast(Any, reasoning_effort),
verbosity=cast(Any, verbosity),
)

226
src/utils/llm/adapters/openai_common.py Normal file
View File

@ -0,0 +1,226 @@
"""
Shared helpers for OpenAI-compatible providers.
OpenAI, OpenRouter, and vLLM all use the OpenAI-compatible chat API shape but have
provider-specific quirks. This module holds shared parsing and streaming helpers
to keep per-provider adapters small and explicit.
"""
from __future__ import annotations
import json
import logging
from collections.abc import AsyncIterator
from typing import Any, Literal, cast
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionChunk
from pydantic import BaseModel
from src.utils.llm.models import HonchoLLMCallStreamChunk
logger = logging.getLogger(__name__)
def extract_openai_tool_calls(message: Any) -> list[dict[str, Any]]:
"""
Extract tool calls from an OpenAI-compatible message object.
OpenAI-compatible SDKs may return different tool call shapes. This helper
treats tool call entries as dynamic objects and extracts:
- id
- function name
- function arguments (JSON string)
"""
tool_calls_obj = getattr(message, "tool_calls", None)
if not tool_calls_obj:
return []
tool_calls_any = cast(list[Any], tool_calls_obj)
parsed: list[dict[str, Any]] = []
for tc in tool_calls_any:
if isinstance(tc, dict):
tc_dict: dict[str, Any] = cast(dict[str, Any], tc)
tc_id = cast(str, tc_dict.get("id", ""))
func = tc_dict.get("function")
if isinstance(func, dict):
func_dict: dict[str, Any] = cast(dict[str, Any], func)
name = cast(str, func_dict.get("name", ""))
args_str = func_dict.get("arguments")
else:
name = cast(str, tc_dict.get("name", ""))
args_str = tc_dict.get("arguments")
else:
tc_id = cast(str, getattr(tc, "id", ""))
func_obj = getattr(tc, "function", None)
if func_obj is not None:
name = cast(str, getattr(func_obj, "name", ""))
args_str = getattr(func_obj, "arguments", None)
else:
name = cast(str, getattr(tc, "name", ""))
args_str = getattr(tc, "arguments", None)
args: dict[str, Any] = {}
if isinstance(args_str, str) and args_str:
try:
parsed_args_any = json.loads(args_str)
if isinstance(parsed_args_any, dict):
args = cast(dict[str, Any], parsed_args_any)
except Exception:
args = {}
parsed.append({"id": tc_id, "name": name, "input": args})
return parsed
def extract_openai_reasoning_content(response: Any) -> str | None:
"""
Extract reasoning/thinking content from an OpenAI ChatCompletion response.
GPT-5 and o1 models may include `reasoning_details` in the response message.
Some OpenAI-compatible proxies may include `reasoning_content`.
"""
try:
message = response.choices[0].message
if hasattr(message, "reasoning_details") and message.reasoning_details:
reasoning_parts: list[Any] = []
for detail in message.reasoning_details:
if hasattr(detail, "content") and detail.content:
reasoning_parts.append(detail.content)
elif isinstance(detail, dict):
detail_dict: dict[str, Any] = cast(dict[str, Any], detail)
content = detail_dict.get("content")
if isinstance(content, str) and content:
reasoning_parts.append(content)
if reasoning_parts:
return "\n".join(reasoning_parts)
if hasattr(message, "reasoning_content") and message.reasoning_content:
return message.reasoning_content
except (AttributeError, IndexError, TypeError):
pass
return None
def extract_openai_reasoning_details(response: Any) -> list[dict[str, Any]]:
"""
Extract `reasoning_details` array from an OpenAI-compatible ChatCompletion response.
OpenRouter may return reasoning blocks in `reasoning_details` that must be preserved
and passed back in subsequent requests for Gemini models with tool use.
"""
try:
message = response.choices[0].message
if hasattr(message, "reasoning_details") and message.reasoning_details:
return [
detail.model_dump() if hasattr(detail, "model_dump") else dict(detail)
for detail in message.reasoning_details
]
except (AttributeError, IndexError, TypeError):
pass
return []
def extract_openai_cache_tokens(usage: Any) -> tuple[int, int]:
"""
Extract cache token counts from OpenAI-style usage objects.
Returns:
Tuple of (cache_creation_tokens, cache_read_tokens).
"""
if not usage:
return 0, 0
cache_read = 0
if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details:
details = usage.prompt_tokens_details
if hasattr(details, "cached_tokens") and details.cached_tokens:
cache_read = details.cached_tokens
if cache_read == 0:
if hasattr(usage, "cache_read_input_tokens") and usage.cache_read_input_tokens:
cache_read = usage.cache_read_input_tokens
elif hasattr(usage, "cached_tokens") and usage.cached_tokens:
cache_read = usage.cached_tokens
cache_creation = 0
if (
hasattr(usage, "cache_creation_input_tokens")
and usage.cache_creation_input_tokens
):
cache_creation = usage.cache_creation_input_tokens
return cache_creation, cache_read
async def stream_openai_compatible(
*,
client: AsyncOpenAI,
model: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None,
verbosity: Literal["low", "medium", "high"] | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""
Stream an OpenAI-compatible response and normalize chunks.
The OpenAI python client returns an async iterator of ChatCompletionChunk items.
With `include_usage`, a final chunk can include usage details in `chunk.usage`.
"""
openai_params: dict[str, Any] = {
"model": model,
"messages": messages,
"stream": True,
"stream_options": {"include_usage": True},
}
if "gpt-5" in model:
openai_params["max_completion_tokens"] = max_tokens
if reasoning_effort:
openai_params["reasoning_effort"] = reasoning_effort
if verbosity:
openai_params["verbosity"] = verbosity
else:
openai_params["max_tokens"] = max_tokens
if response_model:
openai_params["response_format"] = response_model
elif json_mode:
openai_params["response_format"] = {"type": "json_object"}
openai_stream = cast(
AsyncIterator[ChatCompletionChunk],
await client.chat.completions.create(**openai_params),
)
finish_reason: str | None = None
usage_chunk_received = False
async for chunk in openai_stream:
if chunk.choices and chunk.choices[0].delta.content:
yield HonchoLLMCallStreamChunk(content=chunk.choices[0].delta.content)
if chunk.choices and chunk.choices[0].finish_reason:
finish_reason = chunk.choices[0].finish_reason
if hasattr(chunk, "usage") and chunk.usage:
yield HonchoLLMCallStreamChunk(
content="",
is_done=True,
finish_reasons=[finish_reason] if finish_reason else [],
output_tokens=chunk.usage.completion_tokens,
)
usage_chunk_received = True
if not usage_chunk_received and finish_reason:
logger.warning(
"OpenAI-compatible stream ended without usage chunk (interrupted)"
)
yield HonchoLLMCallStreamChunk(
content="",
is_done=True,
finish_reasons=[finish_reason],
output_tokens=None,
)

241
src/utils/llm/adapters/openrouter.py Normal file
View File

@ -0,0 +1,241 @@
"""
OpenRouter provider adapter.
OpenRouter uses an OpenAI-compatible API surface but has its own quirks:
- Some models (including Gemini routed through OpenRouter) require `reasoning_details`
to be preserved across multi-turn tool use.
- Prompt caching for Anthropic models proxied through OpenRouter can be enabled by
converting system messages into content blocks with `cache_control`.
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletion
from pydantic import BaseModel
from src.utils.llm.adapters.base import (
ProviderAdapter,
openai_tool_calls_from_tool_calls,
)
from src.utils.llm.adapters.openai_common import (
extract_openai_cache_tokens,
extract_openai_reasoning_content,
extract_openai_reasoning_details,
extract_openai_tool_calls,
stream_openai_compatible,
)
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class OpenRouterAdapter(ProviderAdapter):
"""ProviderAdapter implementation for OpenRouter via OpenAI-compatible SDK client."""
provider: SupportedProviders = "openrouter"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Convert Anthropic-style tool schemas to OpenAI tool definitions."""
return [
{
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
},
}
for tool in tools
]
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant message with tool calls for OpenRouter."""
_ = thinking_blocks
msg: dict[str, Any] = {
"role": "assistant",
"content": content if isinstance(content, str) else None,
"tool_calls": openai_tool_calls_from_tool_calls(tool_calls),
}
if reasoning_details:
msg["reasoning_details"] = reasoning_details
return msg
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results in OpenAI tool-message format."""
for tr in tool_results:
conversation_messages.append(
{
"role": "tool",
"tool_call_id": tr["tool_id"],
"content": str(tr["result"]),
}
)
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming OpenRouter call and normalize the response."""
_ = (provider, prompt, thinking_budget_tokens, stop_seqs)
processed_messages: list[dict[str, Any]] = []
for msg in messages:
if msg.get("role") == "system" and isinstance(msg.get("content"), str):
processed_messages.append(
{
"role": "system",
"content": [
{
"type": "text",
"text": msg["content"],
"cache_control": {"type": "ephemeral"},
}
],
}
)
else:
processed_messages.append(msg)
openai_params: dict[str, Any] = {"model": model, "messages": processed_messages}
if temperature is not None and "gpt-5" not in model:
openai_params["temperature"] = temperature
if "gpt-5" in model:
openai_params["max_completion_tokens"] = max_tokens
if reasoning_effort:
openai_params["reasoning_effort"] = reasoning_effort
if verbosity:
openai_params["verbosity"] = verbosity
else:
openai_params["max_tokens"] = max_tokens
if tools and not response_model:
openai_params["tools"] = tools
if tool_choice:
openai_params["tool_choice"] = tool_choice
if json_mode:
openai_params["response_format"] = {"type": "json_object"}
openai_client = cast(AsyncOpenAI, client)
if response_model:
openai_params["response_format"] = response_model
response = cast(
ChatCompletion,
await openai_client.chat.completions.parse(**openai_params),
)
message_any = cast(Any, response.choices[0].message)
parsed_content = getattr(message_any, "parsed", None)
if parsed_content is None:
raise ValueError("No parsed content in structured response")
usage = response.usage
finish_reason = response.choices[0].finish_reason
if not isinstance(parsed_content, response_model):
raise ValueError(
f"Parsed content does not match the response model: {parsed_content} != {response_model}"
)
parsed_tool_calls = extract_openai_tool_calls(response.choices[0].message)
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=parsed_content,
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=parsed_tool_calls,
thinking_content=extract_openai_reasoning_content(response),
reasoning_details=extract_openai_reasoning_details(response),
)
response = cast(
ChatCompletion, await openai_client.chat.completions.create(**openai_params)
)
usage = response.usage
finish_reason = response.choices[0].finish_reason
tool_calls_list = extract_openai_tool_calls(response.choices[0].message)
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=response.choices[0].message.content or "",
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=tool_calls_list,
thinking_content=extract_openai_reasoning_content(response),
reasoning_details=extract_openai_reasoning_details(response),
)
def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream an OpenRouter response and normalize chunks."""
_ = (provider, prompt, temperature, thinking_budget_tokens)
openai_client = cast(AsyncOpenAI, client)
return stream_openai_compatible(
client=openai_client,
model=model,
max_tokens=max_tokens,
messages=messages,
response_model=response_model,
json_mode=json_mode,
reasoning_effort=cast(Any, reasoning_effort),
verbosity=cast(Any, verbosity),
)

273
src/utils/llm/adapters/vllm.py Normal file
View File

@ -0,0 +1,273 @@
"""
vLLM provider adapter.
Honcho uses vLLM via an OpenAI-compatible API surface. However, structured output
support is currently implemented only for `PromptRepresentation` using JSON Schema,
and includes schema-aware repair to prevent downstream validation failures.
"""
from __future__ import annotations
import json
import logging
from collections.abc import AsyncIterator
from typing import Any, cast
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletion
from pydantic import BaseModel, ValidationError
from src.utils.json_parser import validate_and_repair_json
from src.utils.llm.adapters.base import (
ProviderAdapter,
openai_tool_calls_from_tool_calls,
)
from src.utils.llm.adapters.openai_common import (
extract_openai_cache_tokens,
extract_openai_reasoning_content,
extract_openai_reasoning_details,
extract_openai_tool_calls,
stream_openai_compatible,
)
from src.utils.llm.models import HonchoLLMCallResponse, HonchoLLMCallStreamChunk
from src.utils.representation import PromptRepresentation
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
class VLLMAdapter(ProviderAdapter):
"""ProviderAdapter implementation for vLLM via OpenAI-compatible SDK client."""
provider: SupportedProviders = "vllm"
def convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Convert Anthropic-style tool schemas to OpenAI tool definitions."""
return [
{
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
},
}
for tool in tools
]
def format_assistant_tool_message(
self,
*,
content: Any,
tool_calls: list[dict[str, Any]],
thinking_blocks: list[dict[str, Any]] | None = None,
reasoning_details: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Format an assistant message with tool calls for vLLM."""
_ = thinking_blocks
msg: dict[str, Any] = {
"role": "assistant",
"content": content if isinstance(content, str) else None,
"tool_calls": openai_tool_calls_from_tool_calls(tool_calls),
}
if reasoning_details:
msg["reasoning_details"] = reasoning_details
return msg
def append_tool_results(
self,
*,
tool_results: list[dict[str, Any]],
conversation_messages: list[dict[str, Any]],
) -> None:
"""Append tool results in OpenAI tool-message format."""
for tr in tool_results:
conversation_messages.append(
{
"role": "tool",
"tool_call_id": tr["tool_id"],
"content": str(tr["result"]),
}
)
async def call(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
tools: list[dict[str, Any]] | None,
tool_choice: str | dict[str, Any] | None,
) -> HonchoLLMCallResponse[Any]:
"""Perform a non-streaming vLLM call and normalize the response."""
_ = (
provider,
prompt,
json_mode,
reasoning_effort,
verbosity,
thinking_budget_tokens,
)
openai_client = cast(AsyncOpenAI, client)
openai_params: dict[str, Any] = {"model": model, "messages": messages}
if temperature is not None and "gpt-5" not in model:
openai_params["temperature"] = temperature
if "gpt-5" in model:
openai_params["max_completion_tokens"] = max_tokens
if reasoning_effort:
openai_params["reasoning_effort"] = reasoning_effort
if verbosity:
openai_params["verbosity"] = verbosity
else:
openai_params["max_tokens"] = max_tokens
if tools and not response_model:
openai_params["tools"] = tools
if tool_choice:
openai_params["tool_choice"] = tool_choice
if response_model:
if response_model is not PromptRepresentation:
raise NotImplementedError(
"vLLM structured output currently supports only PromptRepresentation"
)
openai_params["response_format"] = {
"type": "json_schema",
"json_schema": {
"name": response_model.__name__,
"schema": response_model.model_json_schema(),
},
}
if stop_seqs:
openai_params["stop"] = stop_seqs
vllm_response: ChatCompletion = cast(
ChatCompletion,
await openai_client.chat.completions.create(**openai_params),
)
usage = vllm_response.usage
finish_reason = vllm_response.choices[0].finish_reason
try:
test_rep = ""
if vllm_response.choices[0].message.content is not None:
test_rep = vllm_response.choices[0].message.content
final = validate_and_repair_json(test_rep)
repaired_data = json.loads(final)
if "deductive" in repaired_data and isinstance(
repaired_data["deductive"], list
):
for i, item in enumerate(repaired_data["deductive"]):
if isinstance(item, dict):
if "conclusion" not in item and "premises" in item:
logger.warning(
"Deductive observation %s missing conclusion, adding placeholder",
i,
)
if item["premises"]:
item["conclusion"] = (
f"[Incomplete reasoning from premises: {item['premises'][0][:100]}...]"
)
else:
item["conclusion"] = (
"[Incomplete reasoning - conclusion missing]"
)
if "premises" not in item:
item["premises"] = []
final = json.dumps(repaired_data)
except (json.JSONDecodeError, KeyError, TypeError) as e:
final = ""
logger.warning("Could not perform schema-aware repair: %s", e)
try:
response_obj = PromptRepresentation.model_validate_json(final)
except ValidationError as e:
logger.error("Validation error after repair: %s", e)
logger.debug("Problematic JSON: %s", final)
logger.warning(
"Using fallback empty Representation due to validation error"
)
response_obj = PromptRepresentation(explicit=[])
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=response_obj,
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=[],
thinking_content=extract_openai_reasoning_content(vllm_response),
)
response = cast(
ChatCompletion, await openai_client.chat.completions.create(**openai_params)
)
usage = response.usage
finish_reason = response.choices[0].finish_reason
tool_calls_list = extract_openai_tool_calls(response.choices[0].message)
cache_creation, cache_read = extract_openai_cache_tokens(usage)
return HonchoLLMCallResponse(
content=response.choices[0].message.content or "",
input_tokens=usage.prompt_tokens if usage else 0,
output_tokens=usage.completion_tokens if usage else 0,
cache_creation_input_tokens=cache_creation,
cache_read_input_tokens=cache_read,
finish_reasons=[finish_reason] if finish_reason else [],
tool_calls_made=tool_calls_list,
thinking_content=extract_openai_reasoning_content(response),
reasoning_details=extract_openai_reasoning_details(response),
)
def stream(
self,
*,
client: Any,
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""Stream a vLLM response and normalize chunks."""
_ = (provider, prompt, temperature, thinking_budget_tokens)
openai_client = cast(AsyncOpenAI, client)
return stream_openai_compatible(
client=openai_client,
model=model,
max_tokens=max_tokens,
messages=messages,
response_model=response_model,
json_mode=json_mode,
reasoning_effort=cast(Any, reasoning_effort),
verbosity=cast(Any, verbosity),
)

618
src/utils/llm/core.py Normal file
View File

@ -0,0 +1,618 @@
"""
Provider-agnostic LLM call orchestration.
This module provides the primary entry points (`honcho_llm_call`, `honcho_llm_call_inner`)
and encapsulates retry/failover behavior. Provider-specific request/response logic
is delegated to adapters.
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator, Callable
from contextvars import ContextVar
from typing import Any, Literal, TypeVar, overload
from anthropic import AsyncAnthropic
from google import genai
from groq import AsyncGroq
from openai import AsyncOpenAI
from pydantic import BaseModel
from sentry_sdk.ai.monitoring import ai_track
from tenacity import retry, stop_after_attempt, wait_exponential
from src.config import LLMComponentSettings
from src.telemetry.logging import conditional_observe
from src.telemetry.reasoning_traces import log_reasoning_trace
from src.utils.llm.adapters import get_adapter
from src.utils.llm.models import (
HonchoLLMCallResponse,
HonchoLLMCallStreamChunk,
IterationCallback,
StreamingResponseWithMetadata,
)
from src.utils.llm.registry import CLIENTS
from src.utils.llm.tool_loop import (
MAX_TOOL_ITERATIONS,
MIN_TOOL_ITERATIONS,
execute_tool_loop,
)
from src.utils.types import SupportedProviders
logger = logging.getLogger(__name__)
M = TypeVar("M", bound=BaseModel)
ReasoningEffortType = Literal["low", "medium", "high", "minimal"] | None
VerbosityType = Literal["low", "medium", "high"] | None
_current_attempt: ContextVar[int] = ContextVar("current_attempt", default=0)
def _get_effective_temperature(temperature: float | None) -> float | None:
"""Adjust temperature on retries - bump 0.0 to 0.2 to get different results."""
if temperature == 0.0 and _current_attempt.get() > 1:
logger.debug("Bumping temperature from 0.0 to 0.2 on retry")
return 0.2
return temperature
async def _stream_final_response(
*,
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
conversation_messages: list[dict[str, Any]],
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: ReasoningEffortType,
verbosity: VerbosityType,
thinking_budget_tokens: int | None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""
Stream the final response after tool execution is complete.
This performs a streaming call without tools using the accumulated conversation
messages (including tool call results).
"""
provider = llm_settings.PROVIDER
model = llm_settings.MODEL
client = CLIENTS.get(provider)
if not client:
raise ValueError(f"Missing client for {provider}")
stream_response = await honcho_llm_call_inner(
provider,
model,
prompt,
max_tokens,
response_model,
json_mode,
_get_effective_temperature(temperature),
stop_seqs,
reasoning_effort,
verbosity,
thinking_budget_tokens,
True,
None,
None,
conversation_messages,
)
async for chunk in stream_response:
yield chunk
async def handle_streaming_response(
client: AsyncAnthropic | AsyncOpenAI | genai.Client | AsyncGroq,
params: dict[str, Any],
json_mode: bool,
thinking_budget_tokens: int | None,
response_model: type[BaseModel] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]:
"""
Compatibility wrapper for streaming responses.
The legacy implementation selected behavior based on the concrete SDK client
type. This wrapper retains the same signature for tests and internal callers.
"""
if isinstance(client, AsyncAnthropic):
provider: SupportedProviders = "anthropic"
elif isinstance(client, genai.Client):
provider = "google"
elif isinstance(client, AsyncGroq):
provider = "groq"
else:
provider = "openai"
adapter = get_adapter(provider)
stream = adapter.stream(
client=client,
provider=provider,
model=params["model"],
prompt=params["messages"][0]["content"] if params.get("messages") else "",
max_tokens=params["max_tokens"],
messages=params["messages"],
response_model=response_model,
json_mode=json_mode,
temperature=params.get("temperature"),
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
)
async for chunk in stream:
yield chunk
@overload
async def honcho_llm_call_inner(
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
response_model: type[M],
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
stream: Literal[False] = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
messages: list[dict[str, Any]] | None = None,
) -> HonchoLLMCallResponse[M]: ...
@overload
async def honcho_llm_call_inner(
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
response_model: None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
stream: Literal[False] = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
messages: list[dict[str, Any]] | None = None,
) -> HonchoLLMCallResponse[str]: ...
@overload
async def honcho_llm_call_inner(
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
stream: Literal[True] = ...,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
messages: list[dict[str, Any]] | None = None,
) -> AsyncIterator[HonchoLLMCallStreamChunk]: ...
async def honcho_llm_call_inner(
provider: SupportedProviders,
model: str,
prompt: str,
max_tokens: int,
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
stream: bool = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
messages: list[dict[str, Any]] | None = None,
) -> HonchoLLMCallResponse[Any] | AsyncIterator[HonchoLLMCallStreamChunk]:
"""
Perform a single provider call (streaming or non-streaming).
Callers are responsible for converting tools to the provider-specific format.
"""
client = CLIENTS[provider]
if messages is None:
messages = [{"role": "user", "content": prompt}]
adapter = get_adapter(provider)
if stream:
return adapter.stream(
client=client,
provider=provider,
model=model,
prompt=prompt,
max_tokens=max_tokens,
messages=messages,
response_model=response_model,
json_mode=json_mode,
temperature=temperature,
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
)
return await adapter.call(
client=client,
provider=provider,
model=model,
prompt=prompt,
max_tokens=max_tokens,
messages=messages,
response_model=response_model,
json_mode=json_mode,
temperature=temperature,
stop_seqs=stop_seqs,
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
tools=tools,
tool_choice=tool_choice,
)
@overload
async def honcho_llm_call(
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
track_name: str | None = None,
*,
response_model: type[M],
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
stream: Literal[False] = False,
stream_final_only: bool = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
max_tool_iterations: int = 10,
messages: list[dict[str, Any]] | None = None,
max_input_tokens: int | None = None,
trace_name: str | None = None,
iteration_callback: IterationCallback | None = None,
) -> HonchoLLMCallResponse[M]: ...
@overload
async def honcho_llm_call(
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
track_name: str | None = None,
response_model: None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
stream: Literal[False] = False,
stream_final_only: bool = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
max_tool_iterations: int = 10,
messages: list[dict[str, Any]] | None = None,
max_input_tokens: int | None = None,
trace_name: str | None = None,
iteration_callback: IterationCallback | None = None,
) -> HonchoLLMCallResponse[str]: ...
@overload
async def honcho_llm_call(
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
track_name: str | None = None,
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
stream: Literal[True] = ...,
stream_final_only: bool = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
max_tool_iterations: int = 10,
messages: list[dict[str, Any]] | None = None,
max_input_tokens: int | None = None,
trace_name: str | None = None,
iteration_callback: IterationCallback | None = None,
) -> AsyncIterator[HonchoLLMCallStreamChunk] | StreamingResponseWithMetadata: ...
@conditional_observe(name="LLM Call")
async def honcho_llm_call(
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
track_name: str | None = None,
response_model: type[BaseModel] | None = None,
json_mode: bool = False,
temperature: float | None = None,
stop_seqs: list[str] | None = None,
reasoning_effort: Literal["low", "medium", "high", "minimal"] | None = None,
verbosity: Literal["low", "medium", "high"] | None = None,
thinking_budget_tokens: int | None = None,
enable_retry: bool = True,
retry_attempts: int = 3,
stream: bool = False,
stream_final_only: bool = False,
tools: list[dict[str, Any]] | None = None,
tool_choice: str | dict[str, Any] | None = None,
tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
max_tool_iterations: int = 10,
messages: list[dict[str, Any]] | None = None,
max_input_tokens: int | None = None,
trace_name: str | None = None,
iteration_callback: IterationCallback | None = None,
) -> (
HonchoLLMCallResponse[Any]
| AsyncIterator[HonchoLLMCallStreamChunk]
| StreamingResponseWithMetadata
):
"""
Make an LLM call with automatic backup provider failover.
Backup provider/model is used on the final retry attempt (3 by default).
"""
if stream and tools and not stream_final_only:
raise ValueError(
"Streaming is not supported with tool calling. Set stream=False when using tools, "
+ "or use stream_final_only=True to stream only the final response after tool calls."
)
_current_attempt.set(1)
def _get_provider_and_model() -> (
tuple[SupportedProviders, str, int | None, ReasoningEffortType, VerbosityType]
):
"""Get the provider and model to use based on current attempt."""
attempt = _current_attempt.get()
provider: SupportedProviders
model: str
thinking_budget: int | None
gpt5_reasoning_effort: ReasoningEffortType
gpt5_verbosity: VerbosityType
if (
attempt == retry_attempts
and llm_settings.BACKUP_PROVIDER is not None
and llm_settings.BACKUP_MODEL is not None
and llm_settings.BACKUP_PROVIDER in CLIENTS
):
provider = llm_settings.BACKUP_PROVIDER
model = llm_settings.BACKUP_MODEL
thinking_budget = thinking_budget_tokens
gpt5_reasoning_effort = reasoning_effort
gpt5_verbosity = verbosity
if provider != "anthropic" and thinking_budget:
logger.warning(
"thinking_budget_tokens not supported by %s, ignoring", provider
)
thinking_budget = None
if "gpt-5" not in model and (gpt5_reasoning_effort or gpt5_verbosity):
logger.warning(
"reasoning_effort/verbosity only supported by GPT-5 models, ignoring"
)
gpt5_reasoning_effort = None
gpt5_verbosity = None
logger.warning(
"Final retry attempt %s/%s: switching from %s/%s to backup %s/%s",
attempt,
retry_attempts,
llm_settings.PROVIDER,
llm_settings.MODEL,
provider,
model,
)
else:
provider = llm_settings.PROVIDER
model = llm_settings.MODEL
thinking_budget = thinking_budget_tokens
gpt5_reasoning_effort = reasoning_effort
gpt5_verbosity = verbosity
return provider, model, thinking_budget, gpt5_reasoning_effort, gpt5_verbosity
async def _call_with_provider_selection() -> (
HonchoLLMCallResponse[Any] | AsyncIterator[HonchoLLMCallStreamChunk]
):
"""Select provider/model per attempt and call once."""
provider, model, thinking_budget, gpt5_reasoning_effort, gpt5_verbosity = (
_get_provider_and_model()
)
client = CLIENTS.get(provider)
if not client:
raise ValueError(f"Missing client for {provider}")
converted_tools = None
if tools:
adapter = get_adapter(provider)
converted_tools = adapter.convert_tools(tools)
if stream:
return await honcho_llm_call_inner(
provider,
model,
prompt,
max_tokens,
response_model,
json_mode,
_get_effective_temperature(temperature),
stop_seqs,
gpt5_reasoning_effort,
gpt5_verbosity,
thinking_budget,
True,
converted_tools,
tool_choice,
messages,
)
return await honcho_llm_call_inner(
provider,
model,
prompt,
max_tokens,
response_model,
json_mode,
_get_effective_temperature(temperature),
stop_seqs,
gpt5_reasoning_effort,
gpt5_verbosity,
thinking_budget,
False,
converted_tools,
tool_choice,
messages,
)
decorated = _call_with_provider_selection
if track_name:
decorated = ai_track(track_name)(decorated)
def before_retry_callback(retry_state: Any) -> None:
"""Update attempt counter before each retry."""
next_attempt = retry_state.attempt_number + 1
_current_attempt.set(next_attempt)
exc = retry_state.outcome.exception() if retry_state.outcome else None
if exc:
logger.warning(
"Error on attempt %s/%s with %s/%s: %s",
retry_state.attempt_number,
retry_attempts,
llm_settings.PROVIDER,
llm_settings.MODEL,
exc,
)
logger.info("Will retry with attempt %s/%s", next_attempt, retry_attempts)
if enable_retry:
decorated = retry(
stop=stop_after_attempt(retry_attempts),
wait=wait_exponential(multiplier=1, min=4, max=10),
before_sleep=before_retry_callback,
)(decorated)
if not tools or not tool_executor:
result: (
HonchoLLMCallResponse[Any] | AsyncIterator[HonchoLLMCallStreamChunk]
) = await decorated()
if trace_name and isinstance(result, HonchoLLMCallResponse):
log_reasoning_trace(
task_type=trace_name,
llm_settings=llm_settings,
prompt=prompt,
response=result,
max_tokens=max_tokens,
thinking_budget_tokens=thinking_budget_tokens,
reasoning_effort=reasoning_effort,
json_mode=json_mode,
stop_seqs=stop_seqs,
messages=messages,
)
return result
clamped_iterations = max(
MIN_TOOL_ITERATIONS, min(max_tool_iterations, MAX_TOOL_ITERATIONS)
)
if clamped_iterations != max_tool_iterations:
logger.warning(
"max_tool_iterations %s clamped to %s (valid range: %s-%s)",
max_tool_iterations,
clamped_iterations,
MIN_TOOL_ITERATIONS,
MAX_TOOL_ITERATIONS,
)
def _set_attempt(attempt: int) -> None:
"""Set the current retry attempt counter."""
_current_attempt.set(attempt)
result = await execute_tool_loop(
llm_settings=llm_settings,
prompt=prompt,
max_tokens=max_tokens,
messages=messages,
tools=tools,
tool_choice=tool_choice,
tool_executor=tool_executor,
max_tool_iterations=clamped_iterations,
response_model=response_model,
json_mode=json_mode,
temperature=temperature,
stop_seqs=stop_seqs,
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
enable_retry=enable_retry,
retry_attempts=retry_attempts,
max_input_tokens=max_input_tokens,
get_provider_and_model=_get_provider_and_model,
before_retry_callback=before_retry_callback,
get_effective_temperature=_get_effective_temperature,
set_attempt=_set_attempt,
call_inner=honcho_llm_call_inner,
stream_final_response=_stream_final_response,
stream_final=stream_final_only,
iteration_callback=iteration_callback,
)
if trace_name and isinstance(result, HonchoLLMCallResponse):
log_reasoning_trace(
task_type=trace_name,
llm_settings=llm_settings,
prompt=prompt,
response=result,
max_tokens=max_tokens,
thinking_budget_tokens=thinking_budget_tokens,
reasoning_effort=reasoning_effort,
json_mode=json_mode,
stop_seqs=stop_seqs,
messages=messages,
)
return result

160
src/utils/llm/history.py Normal file
View File

@ -0,0 +1,160 @@
"""
Conversation history helpers for LLM calls.
This module is intentionally provider-agnostic. It focuses on token estimation and
safe truncation while preserving tool-call structure across providers.
"""
from __future__ import annotations
import json
import logging
from typing import Any, cast
from src.utils.tokens import estimate_tokens
logger = logging.getLogger(__name__)
def count_message_tokens(messages: list[dict[str, Any]]) -> int:
"""Count tokens in a list of messages using tiktoken-style estimation."""
total = 0
for msg in messages:
content = msg.get("content", "")
if isinstance(content, str):
total += estimate_tokens(content)
elif isinstance(content, list):
total += estimate_tokens(json.dumps(content))
if "parts" in msg:
try:
total += estimate_tokens(json.dumps(msg["parts"]))
except TypeError:
total += estimate_tokens(str(msg["parts"]))
return total
def _is_tool_use_message(msg: dict[str, Any]) -> bool:
"""Return True if a message contains tool calls in any supported format."""
content = msg.get("content")
if isinstance(content, list):
for block in cast(list[dict[str, Any]], content):
if block.get("type") == "tool_use":
return True
return bool(msg.get("tool_calls"))
def _is_tool_result_message(msg: dict[str, Any]) -> bool:
"""Return True if a message contains tool results in any supported format."""
content = msg.get("content")
if isinstance(content, list):
for block in cast(list[dict[str, Any]], content):
if block.get("type") == "tool_result":
return True
return msg.get("role") == "tool"
def _group_into_units(messages: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
"""
Group messages into logical conversation units.
A unit is either:
- A tool_use message + ALL consecutive tool_result messages that follow
- A single non-tool message
This ensures tool_use and tool_results stay together.
"""
units: list[list[dict[str, Any]]] = []
i = 0
while i < len(messages):
msg = messages[i]
if _is_tool_use_message(msg):
j = i + 1
while j < len(messages) and _is_tool_result_message(messages[j]):
j += 1
unit = messages[i:j]
if len(unit) > 1:
units.append(unit)
i = j
else:
logger.debug("Skipping orphaned tool_use at index %s", i)
i += 1
elif _is_tool_result_message(msg):
logger.debug("Skipping orphaned tool_result at index %s", i)
i += 1
else:
units.append([msg])
i += 1
return units
def truncate_messages_to_fit(
messages: list[dict[str, Any]],
max_tokens: int,
preserve_system: bool = True,
) -> list[dict[str, Any]]:
"""
Truncate messages to fit within a token limit while maintaining valid structure.
Strategy:
1. Group messages into units (tool_use + results together, or single messages)
2. Remove oldest units first to preserve recent context
3. Units stay intact so tool_use/tool_result pairs are never broken
"""
current_tokens = count_message_tokens(messages)
if current_tokens <= max_tokens:
return messages
logger.info("Truncating: %s tokens exceeds %s limit", current_tokens, max_tokens)
system_messages: list[dict[str, Any]] = []
conversation: list[dict[str, Any]] = []
for msg in messages:
if msg.get("role") == "system" and preserve_system:
system_messages.append(msg)
else:
conversation.append(msg)
system_tokens = count_message_tokens(system_messages)
available_tokens = max_tokens - system_tokens
if available_tokens <= 0:
logger.warning("System message exceeds max_input_tokens")
return messages
units = _group_into_units(conversation)
if not units:
logger.warning("No valid conversation units")
return system_messages
while len(units) > 1:
flat_messages = [msg for unit in units for msg in unit]
if count_message_tokens(flat_messages) <= available_tokens:
break
removed_unit = units.pop(0)
logger.debug(
"Removed unit with %s messages (~%s tokens)",
len(removed_unit),
count_message_tokens(removed_unit),
)
result_conversation = [msg for unit in units for msg in unit]
result = system_messages + result_conversation
result_tokens = count_message_tokens(result)
logger.info(
"Truncation complete: %s -> %s messages, %s -> %s tokens, %s units kept",
len(messages),
len(result),
current_tokens,
result_tokens,
len(units),
)
return result

150
src/utils/llm/models.py Normal file
View File

@ -0,0 +1,150 @@
"""
Shared models for Honcho LLM client orchestration.
This module contains the response/container types that are shared across providers
and orchestration layers (retry/failover, tool loop, streaming).
"""
from __future__ import annotations
from collections.abc import AsyncIterator, Callable
from dataclasses import dataclass
from typing import Any, Generic, TypeVar
from pydantic import BaseModel, Field
T = TypeVar("T")
@dataclass
class IterationData:
"""Data passed to iteration callbacks after each tool execution loop iteration."""
iteration: int
"""1-indexed iteration number."""
tool_calls: list[str]
"""List of tool names called in this iteration."""
input_tokens: int
"""Input tokens used in this iteration's LLM call."""
output_tokens: int
"""Output tokens generated in this iteration's LLM call."""
cache_read_tokens: int = 0
"""Tokens read from cache in this iteration."""
cache_creation_tokens: int = 0
"""Tokens written to cache in this iteration."""
# Callback invoked after each tool-loop iteration.
IterationCallback = Callable[[IterationData], None]
class HonchoLLMCallResponse(BaseModel, Generic[T]):
"""
Response object for LLM calls.
Args:
content: The response content. When a response_model is provided, this will be
the parsed object of that type. Otherwise, it will be a string.
input_tokens: Total number of input tokens (including cached).
output_tokens: Number of tokens generated in the response.
cache_creation_input_tokens: Number of tokens written to cache.
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
(cache_creation costs 25% more, cache_read costs 90% less)
"""
content: T
input_tokens: int = 0
output_tokens: int
cache_creation_input_tokens: int = 0
cache_read_input_tokens: int = 0
finish_reasons: list[str]
tool_calls_made: list[dict[str, Any]] = Field(default_factory=list)
iterations: int = 0
"""Number of LLM calls made in the tool execution loop (1 = single response, 2+ = tool iterations plus final synthesis)."""
thinking_content: str | None = None
"""Normalized reasoning/thinking content for telemetry/debugging."""
thinking_blocks: list[dict[str, Any]] = Field(default_factory=list)
"""Full thinking blocks with signatures for multi-turn conversation replay (Anthropic only)."""
reasoning_details: list[dict[str, Any]] = Field(default_factory=list)
"""OpenRouter reasoning_details for Gemini models - must be preserved across turns."""
messages: list[dict[str, Any]] = Field(default_factory=list)
"""Full conversation history for two-phase dialectic (search -> synthesis)."""
class HonchoLLMCallStreamChunk(BaseModel):
"""
A single chunk in a streaming LLM response.
Args:
content: The text content for this chunk. Empty for chunks that only contain metadata.
is_done: Whether this is the final chunk in the stream.
finish_reasons: List of finish reasons if the stream is complete.
output_tokens: Number of tokens generated in the response. Only set on the final chunk.
"""
content: str
is_done: bool = False
finish_reasons: list[str] = Field(default_factory=list)
output_tokens: int | None = None
class StreamingResponseWithMetadata:
"""
Wrapper for streaming responses that includes metadata from the tool execution phase.
This allows callers to access tool call counts, token usage, and thinking content
from the tool loop while still streaming the final response.
"""
_stream: AsyncIterator[HonchoLLMCallStreamChunk]
tool_calls_made: list[dict[str, Any]]
input_tokens: int
output_tokens: int
cache_creation_input_tokens: int
cache_read_input_tokens: int
thinking_content: str | None
iterations: int
messages: list[dict[str, Any]]
def __init__(
self,
stream: AsyncIterator[HonchoLLMCallStreamChunk],
tool_calls_made: list[dict[str, Any]],
input_tokens: int,
output_tokens: int,
cache_creation_input_tokens: int,
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
self.input_tokens = input_tokens
self.output_tokens = output_tokens
self.cache_creation_input_tokens = cache_creation_input_tokens
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__()
async def __anext__(self) -> HonchoLLMCallStreamChunk:
return await self._stream.__anext__()

96
src/utils/llm/registry.py Normal file
View File

@ -0,0 +1,96 @@
"""
Provider client initialization and validation.
This module owns the global `CLIENTS` registry that maps Honcho provider names to
initialized SDK clients. It also validates that all configured providers and
backup providers are available at import time.
"""
from __future__ import annotations
from typing import Any
from anthropic import AsyncAnthropic
from google import genai
from groq import AsyncGroq
from openai import AsyncOpenAI
from src.config import settings
from src.utils.types import SupportedProviders
# Global mapping of provider identifiers to initialized SDK clients.
CLIENTS: dict[
SupportedProviders,
AsyncAnthropic | AsyncOpenAI | genai.Client | AsyncGroq,
] = {}
if settings.LLM.ANTHROPIC_API_KEY:
anthropic = AsyncAnthropic(
api_key=settings.LLM.ANTHROPIC_API_KEY,
timeout=600.0,
)
CLIENTS["anthropic"] = anthropic
if settings.LLM.OPENAI_API_KEY:
openai_client = AsyncOpenAI(
api_key=settings.LLM.OPENAI_API_KEY,
)
CLIENTS["openai"] = openai_client
if settings.LLM.OPENAI_COMPATIBLE_API_KEY:
CLIENTS["openrouter"] = AsyncOpenAI(
api_key=settings.LLM.OPENAI_COMPATIBLE_API_KEY,
base_url=settings.LLM.OPENAI_COMPATIBLE_BASE_URL
or "https://openrouter.ai/api/v1",
)
if settings.LLM.VLLM_API_KEY and settings.LLM.VLLM_BASE_URL:
CLIENTS["vllm"] = AsyncOpenAI(
api_key=settings.LLM.VLLM_API_KEY,
base_url=settings.LLM.VLLM_BASE_URL,
)
if settings.LLM.GEMINI_API_KEY:
google = genai.client.Client(api_key=settings.LLM.GEMINI_API_KEY)
CLIENTS["google"] = google
if settings.LLM.GROQ_API_KEY:
groq = AsyncGroq(api_key=settings.LLM.GROQ_API_KEY)
CLIENTS["groq"] = groq
SELECTED_PROVIDERS: list[tuple[str, Any]] = [
("Summary", settings.SUMMARY.PROVIDER),
("Deriver", settings.DERIVER.PROVIDER),
]
for level, level_settings in settings.DIALECTIC.LEVELS.items():
SELECTED_PROVIDERS.append((f"Dialectic ({level})", level_settings.PROVIDER))
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:
raise ValueError(f"Missing client for {provider_name}: {provider_value}")
BACKUP_PROVIDERS: list[tuple[str, SupportedProviders | None]] = [
("Deriver", settings.DERIVER.BACKUP_PROVIDER),
("Summary", settings.SUMMARY.BACKUP_PROVIDER),
("Dream", settings.DREAM.BACKUP_PROVIDER),
]
for level, level_settings in settings.DIALECTIC.LEVELS.items():
BACKUP_PROVIDERS.append((f"Dialectic ({level})", level_settings.BACKUP_PROVIDER))
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:
raise ValueError(
f"Backup provider for {component_name} is set to {backup_provider}, "
+ "but this provider is not initialized. Please set the required API key/URL environment "
+ "variables or remove the backup configuration."
)

324
src/utils/llm/tool_loop.py Normal file
View File

@ -0,0 +1,324 @@
"""
Tool execution loop orchestration for agentic LLM interactions.
This module is intentionally provider-agnostic. Provider-specific formatting of
tool-call messages and tool-result messages is delegated to adapters.
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator, Awaitable, Callable
from typing import Any
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential
from src.config import LLMComponentSettings
from src.utils.llm.adapters import get_adapter
from src.utils.llm.history import truncate_messages_to_fit
from src.utils.llm.models import (
HonchoLLMCallResponse,
HonchoLLMCallStreamChunk,
IterationCallback,
IterationData,
StreamingResponseWithMetadata,
)
from src.utils.types import SupportedProviders, set_current_iteration
logger = logging.getLogger(__name__)
# Bounds for max_tool_iterations to prevent runaway loops
MIN_TOOL_ITERATIONS = 1
MAX_TOOL_ITERATIONS = 100
async def execute_tool_loop(
*,
llm_settings: LLMComponentSettings,
prompt: str,
max_tokens: int,
messages: list[dict[str, Any]] | None,
tools: list[dict[str, Any]],
tool_choice: str | dict[str, Any] | None,
tool_executor: Callable[[str, dict[str, Any]], Any],
max_tool_iterations: int,
response_model: type[BaseModel] | None,
json_mode: bool,
temperature: float | None,
stop_seqs: list[str] | None,
reasoning_effort: str | None,
verbosity: str | None,
thinking_budget_tokens: int | None,
enable_retry: bool,
retry_attempts: int,
max_input_tokens: int | None,
get_provider_and_model: Callable[
[],
tuple[
SupportedProviders,
str,
int | None,
str | None,
str | None,
],
],
before_retry_callback: Callable[[Any], None],
get_effective_temperature: Callable[[float | None], float | None],
set_attempt: Callable[[int], None],
call_inner: Callable[..., Awaitable[HonchoLLMCallResponse[Any]]],
stream_final_response: Callable[..., AsyncIterator[HonchoLLMCallStreamChunk]],
stream_final: bool = False,
iteration_callback: IterationCallback | None = None,
) -> HonchoLLMCallResponse[Any] | StreamingResponseWithMetadata:
"""
Execute the tool calling loop for agentic LLM interactions.
The loop repeatedly calls the LLM with tools available, executes any requested
tools, appends results back into the conversation, and continues until the
model stops calling tools or max iterations is reached.
"""
conversation_messages: list[dict[str, Any]] = (
messages.copy() if messages else [{"role": "user", "content": prompt}]
)
iteration = 0
all_tool_calls: list[dict[str, Any]] = []
total_input_tokens = 0
total_output_tokens = 0
total_cache_creation_tokens = 0
total_cache_read_tokens = 0
effective_tool_choice = tool_choice
while iteration < max_tool_iterations:
if max_input_tokens is not None:
conversation_messages = truncate_messages_to_fit(
conversation_messages, max_input_tokens
)
async def _call_with_messages(
effective_tool_choice: str | dict[str, Any] | None = effective_tool_choice,
conversation_messages: list[dict[str, Any]] = conversation_messages,
) -> HonchoLLMCallResponse[Any]:
(
provider,
model,
thinking_budget,
provider_reasoning_effort,
provider_verbosity,
) = get_provider_and_model()
adapter = get_adapter(provider)
converted_tools = adapter.convert_tools(tools) if tools else None
return await call_inner(
provider,
model,
prompt,
max_tokens,
response_model,
json_mode,
get_effective_temperature(temperature),
stop_seqs,
provider_reasoning_effort,
provider_verbosity,
thinking_budget,
False,
converted_tools,
effective_tool_choice,
conversation_messages,
)
if enable_retry:
call_func = retry(
stop=stop_after_attempt(retry_attempts),
wait=wait_exponential(multiplier=1, min=4, max=10),
before_sleep=before_retry_callback,
)(_call_with_messages)
else:
call_func = _call_with_messages
response = await call_func()
total_input_tokens += response.input_tokens
total_output_tokens += response.output_tokens
total_cache_creation_tokens += response.cache_creation_input_tokens
total_cache_read_tokens += response.cache_read_input_tokens
if not response.tool_calls_made:
if stream_final:
stream = stream_final_response(
llm_settings=llm_settings,
prompt=prompt,
max_tokens=max_tokens,
conversation_messages=conversation_messages,
response_model=response_model,
json_mode=json_mode,
temperature=temperature,
stop_seqs=stop_seqs,
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
)
return StreamingResponseWithMetadata(
stream=stream,
tool_calls_made=all_tool_calls,
input_tokens=total_input_tokens,
output_tokens=total_output_tokens,
cache_creation_input_tokens=total_cache_creation_tokens,
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
response.input_tokens = total_input_tokens
response.output_tokens = total_output_tokens
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
current_provider, _, _, _, _ = get_provider_and_model()
adapter = get_adapter(current_provider)
assistant_message = adapter.format_assistant_tool_message(
content=response.content,
tool_calls=response.tool_calls_made,
thinking_blocks=response.thinking_blocks,
reasoning_details=response.reasoning_details,
)
conversation_messages.append(assistant_message)
set_current_iteration(iteration + 1)
tool_results: list[dict[str, Any]] = []
for tool_call in response.tool_calls_made:
tool_name = tool_call["name"]
tool_input = tool_call["input"]
tool_id = tool_call.get("id", "")
try:
tool_result = await tool_executor(tool_name, tool_input)
tool_results.append(
{"tool_id": tool_id, "tool_name": tool_name, "result": tool_result}
)
all_tool_calls.append(
{
"tool_name": tool_name,
"tool_input": tool_input,
"tool_result": tool_result,
}
)
except Exception as e:
logger.error("Tool execution failed for %s: %s", tool_name, e)
tool_results.append(
{
"tool_id": tool_id,
"tool_name": tool_name,
"result": f"Error: {str(e)}",
"is_error": True,
}
)
adapter.append_tool_results(
tool_results=tool_results,
conversation_messages=conversation_messages,
)
if iteration_callback is not None:
try:
iteration_data = IterationData(
iteration=iteration + 1,
tool_calls=[tc["name"] for tc in response.tool_calls_made],
input_tokens=response.input_tokens,
output_tokens=response.output_tokens,
cache_read_tokens=response.cache_read_input_tokens or 0,
cache_creation_tokens=response.cache_creation_input_tokens or 0,
)
iteration_callback(iteration_data)
except Exception:
logger.warning("iteration_callback failed", exc_info=True)
if iteration == 0 and effective_tool_choice in ("required", "any"):
effective_tool_choice = "auto"
iteration += 1
synthesis_prompt = (
"You have reached the maximum number of tool calls. "
"Based on all the information you have gathered, provide your final response now. "
"Do not attempt to call any more tools."
)
conversation_messages.append({"role": "user", "content": synthesis_prompt})
if stream_final:
stream = stream_final_response(
llm_settings=llm_settings,
prompt=prompt,
max_tokens=max_tokens,
conversation_messages=conversation_messages,
response_model=response_model,
json_mode=json_mode,
temperature=temperature,
stop_seqs=stop_seqs,
reasoning_effort=reasoning_effort,
verbosity=verbosity,
thinking_budget_tokens=thinking_budget_tokens,
)
return StreamingResponseWithMetadata(
stream=stream,
tool_calls_made=all_tool_calls,
input_tokens=total_input_tokens,
output_tokens=total_output_tokens,
cache_creation_input_tokens=total_cache_creation_tokens,
cache_read_input_tokens=total_cache_read_tokens,
thinking_content=None,
iterations=iteration + 1,
messages=conversation_messages,
)
set_attempt(1)
async def _final_call() -> HonchoLLMCallResponse[Any]:
return await call_inner(
llm_settings.PROVIDER,
llm_settings.MODEL,
prompt,
max_tokens,
response_model,
json_mode,
get_effective_temperature(temperature),
stop_seqs,
reasoning_effort,
verbosity,
thinking_budget_tokens,
False,
None,
None,
conversation_messages,
)
if enable_retry:
final_call_func = retry(
stop=stop_after_attempt(retry_attempts),
wait=wait_exponential(multiplier=1, min=4, max=10),
before_sleep=before_retry_callback,
)(_final_call)
else:
final_call_func = _final_call
final_response = await final_call_func()
final_response.tool_calls_made = all_tool_calls
final_response.iterations = iteration + 1
final_response.input_tokens = total_input_tokens + final_response.input_tokens
final_response.output_tokens = total_output_tokens + final_response.output_tokens
final_response.cache_creation_input_tokens = (
total_cache_creation_tokens + final_response.cache_creation_input_tokens
)
final_response.cache_read_input_tokens = (
total_cache_read_tokens + final_response.cache_read_input_tokens
)
final_response.messages = conversation_messages
return final_response

2
src/utils/summarizer.py
View File

@ -20,8 +20,8 @@ from src.telemetry import otel_metrics
from src.telemetry.events import AgentToolSummaryCreatedEvent, emit
from src.telemetry.logging import accumulate_metric, conditional_observe
from src.telemetry.otel.metrics import DeriverComponents, DeriverTaskTypes, TokenTypes
from src.utils.clients import HonchoLLMCallResponse, honcho_llm_call
from src.utils.formatting import utc_now_iso
from src.utils.llm import HonchoLLMCallResponse, honcho_llm_call
from src.utils.tokens import estimate_tokens, track_deriver_input_tokens
from .. import crud, models

9
src/utils/types.py
View File

@ -25,7 +25,14 @@ class GetOrCreateResult(NamedTuple, Generic[T]):
created: bool
SupportedProviders = Literal["anthropic", "openai", "google", "groq", "custom", "vllm"]
SupportedProviders = Literal[
"anthropic",
"openai",
"openrouter",
"google",
"groq",
"vllm",
]
TaskType = Literal[
"webhook", "summary", "representation", "dream", "deletion", "reconciler"
]

2
tests/bench/obex.py
View File

@ -33,7 +33,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from src.config import settings
from src.deriver.prompts import minimal_deriver_prompt
from src.embedding_client import EmbeddingClient
from src.utils.clients import honcho_llm_call
from src.utils.llm import honcho_llm_call
from src.utils.representation import PromptRepresentation
CANDIDATES_DIR = Path(__file__).parent / "obexeval_data" / "candidates"

16
tests/conftest.py
View File

@ -602,13 +602,12 @@ def mock_honcho_llm_call():
# For string responses, return a simple string
return "Test response content"
# Patch the honcho_llm_call decorator to prevent actual LLM calls at module level
original_decorator = None
try:
import src.utils.clients
import src.utils.llm.core as llm_core
original_decorator = src.utils.clients.honcho_llm_call
src.utils.clients.honcho_llm_call = lambda *args, **kwargs: lambda func: func # pyright: ignore[reportUnknownLambdaType]
original_decorator = llm_core.honcho_llm_call
llm_core.honcho_llm_call = lambda *args, **kwargs: lambda func: func # pyright: ignore[reportUnknownLambdaType]
except ImportError:
pass
@ -642,15 +641,18 @@ def mock_honcho_llm_call():
return mock_llm_decorator
with patch("src.utils.clients.honcho_llm_call", side_effect=decorator_factory):
with (
patch("src.utils.llm.core.honcho_llm_call", side_effect=decorator_factory),
patch("src.utils.llm.honcho_llm_call", side_effect=decorator_factory),
):
yield decorator_factory
# Restore the original decorator
if original_decorator:
try:
import src.utils.clients
import src.utils.llm.core as llm_core
src.utils.clients.honcho_llm_call = original_decorator
llm_core.honcho_llm_call = original_decorator
except ImportError:
pass

2
tests/integration/test_token_metrics.py
View File

@ -30,7 +30,7 @@ from src.schemas import (
ResolvedSummaryConfiguration,
)
from src.telemetry.otel.metrics import otel_metrics
from src.utils.clients import HonchoLLMCallResponse
from src.utils.llm import HonchoLLMCallResponse
from src.utils.representation import ExplicitObservationBase, PromptRepresentation
from src.utils.summarizer import (
SummaryType,

4
tests/utils/test_clients.py
View File

@ -26,7 +26,7 @@ from openai.types.completion_usage import CompletionUsage
from pydantic import BaseModel, Field
from src.config import settings
from src.utils.clients import (
from src.utils.llm import (
CLIENTS,
HonchoLLMCallResponse,
HonchoLLMCallStreamChunk,
@ -209,7 +209,7 @@ class TestAnthropicClient:
# Instead of mocking the CLIENTS dict, we mock the entire AsyncAnthropic class
# to return our configured mock when instantiated
with patch("src.utils.clients.AsyncAnthropic") as mock_anthropic_class:
with patch("src.utils.llm.registry.AsyncAnthropic") as mock_anthropic_class:
mock_client_instance = Mock()
mock_client_instance.messages = mock_messages
mock_anthropic_class.return_value = mock_client_instance