Private
Public Access
107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
"""OpenAI-compatible dataclasses for the Manual Slop ai_client layer.
|
|
|
|
Promotes `NormalizedResponse` and `OpenAICompatibleRequest` from
|
|
`src/openai_compatible.py` to typed dataclasses. The 4 dataclasses
|
|
here model the OpenAI Chat Completion API shape:
|
|
|
|
- ToolCall: a single tool call from the model
|
|
- ToolCallFunction: the function portion of a tool call (name + JSON args)
|
|
- ChatMessage: a single message in the conversation (system/user/assistant/tool)
|
|
- UsageStats: token usage accounting (input, output, cache hits/creation)
|
|
|
|
`NormalizedResponse` and `OpenAICompatibleRequest` keep their public
|
|
shapes but consume these typed shapes internally.
|
|
|
|
CONVENTION: 1-space indentation. NO COMMENTS.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Optional
|
|
|
|
from src.type_aliases import JsonValue
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ToolCallFunction:
|
|
name: str
|
|
arguments: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ToolCall:
|
|
id: str
|
|
function: ToolCallFunction
|
|
type: str = "function"
|
|
|
|
def to_dict(self) -> JsonValue:
|
|
return {
|
|
"id": self.id,
|
|
"type": self.type,
|
|
"function": {
|
|
"name": self.function.name,
|
|
"arguments": self.function.arguments,
|
|
},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ChatMessage:
|
|
role: str
|
|
content: str | list # str for text; list of content parts for multimodal (text + image_url, etc.)
|
|
tool_calls: Optional[tuple[ToolCall, ...]] = None
|
|
tool_call_id: Optional[str] = None
|
|
name: Optional[str] = None
|
|
|
|
def to_dict(self) -> JsonValue:
|
|
d: JsonValue = {"role": self.role, "content": self.content}
|
|
if self.tool_calls is not None:
|
|
d["tool_calls"] = [tc.to_dict() for tc in self.tool_calls]
|
|
if self.tool_call_id is not None:
|
|
d["tool_call_id"] = self.tool_call_id
|
|
if self.name is not None:
|
|
d["name"] = self.name
|
|
return d
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UsageStats:
|
|
input_tokens: int
|
|
output_tokens: int
|
|
cache_read_tokens: int = 0
|
|
cache_creation_tokens: int = 0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NormalizedResponse:
|
|
text: str
|
|
tool_calls: tuple[ToolCall, ...] = ()
|
|
usage: UsageStats = field(default_factory=lambda: UsageStats(input_tokens=0, output_tokens=0))
|
|
raw_response: Any = None
|
|
|
|
def to_legacy_dict(self) -> JsonValue:
|
|
return {
|
|
"text": self.text,
|
|
"tool_calls": [tc.to_dict() for tc in self.tool_calls],
|
|
"usage": {
|
|
"input_tokens": self.usage.input_tokens,
|
|
"output_tokens": self.usage.output_tokens,
|
|
"cache_read_tokens": self.usage.cache_read_tokens,
|
|
"cache_creation_tokens": self.usage.cache_creation_tokens,
|
|
},
|
|
"raw_response": self.raw_response,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class OpenAICompatibleRequest:
|
|
messages: list[ChatMessage]
|
|
model: str
|
|
temperature: float = 0.0
|
|
top_p: float = 1.0
|
|
max_tokens: int = 8192
|
|
tools: Optional[list[JsonValue]] = None
|
|
tool_choice: str = "auto"
|
|
stream: bool = False
|
|
stream_callback: Optional[Callable[[str], None]] = None
|
|
extra_body: Optional[JsonValue] = None |