feat: enhance validation of output token limits in DialecticSettings

- Updated the `_validate_token_budgets` method to ensure effective output token limits exceed thinking budgets for both levels and synthesis settings.
- Improved error messages to provide clearer feedback on validation failures.
- Introduced a new method `_stringify_tool_result_content` in DialecticAgent for consistent string representation of tool result payloads, accommodating various content types.
- Enhanced handling of Anthropic-style content blocks in synthesis message construction to ensure proper formatting and representation.
This commit is contained in:
Benjamin McCormick 2026-01-22 17:43:10 -05:00
parent 9cd21ebd72
commit 142032babe
2 changed files with 95 additions and 12 deletions

View File

@ -472,11 +472,36 @@ class DialecticSettings(HonchoSettings):
@model_validator(mode="after")
def _validate_token_budgets(self) -> "DialecticSettings":
"""Ensure the output token limit exceeds all thinking budgets."""
"""Ensure effective output token limits exceed all thinking budgets."""
for level, level_settings in self.LEVELS.items():
if self.MAX_OUTPUT_TOKENS <= level_settings.THINKING_BUDGET_TOKENS:
effective_level_max_output_tokens = (
level_settings.MAX_OUTPUT_TOKENS
if level_settings.MAX_OUTPUT_TOKENS is not None
else self.MAX_OUTPUT_TOKENS
)
if (
effective_level_max_output_tokens
<= level_settings.THINKING_BUDGET_TOKENS
):
raise ValueError(
f"MAX_OUTPUT_TOKENS must be greater than THINKING_BUDGET_TOKENS for level '{level}'"
f"Effective MAX_OUTPUT_TOKENS ({effective_level_max_output_tokens}) must be greater than THINKING_BUDGET_TOKENS ({level_settings.THINKING_BUDGET_TOKENS}) for level '{level}'"
)
synthesis_settings = level_settings.SYNTHESIS
if synthesis_settings is None:
continue
effective_synthesis_max_output_tokens = (
synthesis_settings.MAX_OUTPUT_TOKENS
if synthesis_settings.MAX_OUTPUT_TOKENS is not None
else self.MAX_OUTPUT_TOKENS
)
if (
effective_synthesis_max_output_tokens
<= synthesis_settings.THINKING_BUDGET_TOKENS
):
raise ValueError(
f"Effective SYNTHESIS.MAX_OUTPUT_TOKENS ({effective_synthesis_max_output_tokens}) must be greater than SYNTHESIS.THINKING_BUDGET_TOKENS ({synthesis_settings.THINKING_BUDGET_TOKENS}) for level '{level}'"
)
return self

View File

@ -372,6 +372,54 @@ class DialecticAgent:
)
)
def _stringify_tool_result_content(self, content: Any) -> str:
"""
Convert a tool result payload into a stable, human-readable string.
Tool results can vary by provider and may include nested content blocks
(e.g., Anthropic-style lists with text/image/attachment blocks). The
synthesis prompt needs a consistent text representation that preserves
all information without assuming a single schema.
Args:
content: Tool result payload (string, dict, list, or arbitrary object).
Returns:
A best-effort string representation suitable for inclusion in the
synthesis context.
"""
if content is None:
return ""
if isinstance(content, str):
return content
try:
if isinstance(content, list):
rendered_parts: list[str] = []
for raw_item in cast(list[Any], content):
item: Any = raw_item
if isinstance(item, dict):
item_dict = cast(dict[str, Any], item)
if item_dict.get("type") == "text":
text = item_dict.get("text", "")
if isinstance(text, str) and text:
rendered_parts.append(text)
continue
rendered_parts.append(
json.dumps(item, ensure_ascii=False, default=str)
)
return "\n".join(part for part in rendered_parts if part)
if isinstance(content, dict):
return json.dumps(
cast(dict[str, Any], content), ensure_ascii=False, default=str
)
return str(cast(object, content))
except Exception:
return str(cast(object, content))
def _build_synthesis_messages(
self,
search_messages: list[dict[str, Any]],
@ -412,20 +460,30 @@ class DialecticAgent:
if isinstance(content, str) and content:
search_context_parts.append(f"[USER]: {content}")
elif isinstance(content, list):
# Could be Anthropic tool results
# Anthropic-style content blocks (text + tool results)
for block in content: # pyright: ignore[reportUnknownVariableType]
if (
isinstance(block, dict)
and block.get("type") == "tool_result" # pyright: ignore[reportUnknownMemberType]
):
if not isinstance(block, dict):
continue
block_dict = cast(dict[str, Any], block)
if block_dict.get("type") == "text":
text_any = block_dict.get("text")
if isinstance(text_any, str) and text_any:
search_context_parts.append(f"[USER]: {text_any}")
continue
if block_dict.get("type") == "tool_result":
tool_id: str = cast(
str,
block.get("tool_use_id", "unknown"), # pyright: ignore[reportUnknownMemberType]
block_dict.get("tool_use_id", "unknown"),
)
result: str = cast(str, block.get("content", "")) # pyright: ignore[reportUnknownMemberType]
search_context_parts.append(
f"[TOOL RESULT ({tool_id})]: {result}"
result = self._stringify_tool_result_content(
block_dict.get("content")
)
if result:
search_context_parts.append(
f"[TOOL RESULT ({tool_id})]: {result}"
)
elif role == "assistant":
content = msg.get("content", "")