butler/transport/types.py

Rama main · réplica de lectura del repositorio original.

"""Normalized types for LLM transport responses.

Zero external dependencies — pure dataclasses + stdlib.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

@dataclass
class ToolCall:
    id: Optional[str]
    name: str
    arguments: str
    provider_data: Optional[Dict[str, Any]] = None

    @property
    def type(self) -> str:
        return "function"

    @property
    def function(self) -> ToolCall:
        return self

    def args_dict(self) -> dict[str, Any]:
        try:
            parsed = json.loads(self.arguments)
            return parsed if isinstance(parsed, dict) else {}
        except (json.JSONDecodeError, TypeError):
            return {}


@dataclass
class Usage:
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    cached_tokens: int = 0


@dataclass
class NormalizedResponse:
    content: Optional[str] = None
    tool_calls: Optional[List[ToolCall]] = None
    finish_reason: str = "stop"
    reasoning: Optional[str] = None
    usage: Optional[Usage] = None
    provider_data: Optional[Dict[str, Any]] = None

    def __post_init__(self) -> None:
        self._extract_inline_thinking()

    def _extract_inline_thinking(self) -> None:
        """Extract thinking blocks into reasoning; strip leaked tags from content."""
        from butler.transport.content_sanitize import extract_thinking_to_reasoning

        if not self.content:
            return
        visible, reasoning = extract_thinking_to_reasoning(self.content, self.reasoning)
        self.content = visible
        self.reasoning = reasoning


def build_tool_call(
    id: Optional[str],
    name: str,
    arguments: Any,
    **provider_fields: Any,
) -> ToolCall:
    if isinstance(arguments, dict):
        args_str = json.dumps(arguments, ensure_ascii=False)
    else:
        args_str = str(arguments) if arguments is not None else "{}"
    return ToolCall(
        id=id,
        name=name,
        arguments=args_str,
        provider_data=provider_fields or None,
    )


def map_finish_reason(
    reason: Optional[str],
    mapping: Optional[Dict[str, str]] = None,
) -> str:
    if reason is None:
        return "stop"
    if mapping and reason in mapping:
        return mapping[reason]
    return reason