Private
Public Access
ed
04d723e420
Phase 2 of any_type_componentization_20260621. Promotes NormalizedResponse
+ OpenAICompatibleRequest from src/openai_compatible.py to typed
dataclasses. The 17 Any sites become 5 dataclasses:
NEW src/openai_schemas.py (138 lines):
- ToolCallFunction dataclass (name, arguments)
- ToolCall dataclass (id, function: ToolCallFunction, type='function')
- ChatMessage dataclass (role, content, tool_calls, tool_call_id, name)
- UsageStats dataclass (input_tokens, output_tokens, cache_read_*, cache_creation_*)
- NormalizedResponse dataclass (text, tool_calls: tuple, usage, raw_response: Any)
- OpenAICompatibleRequest dataclass (messages: list[ChatMessage], model, ...)
NEW tests/test_openai_schemas.py (19 tests, all pass):
- ToolCallFunction, ToolCall, ChatMessage round-trips
- UsageStats field access + frozen=True semantics
- NormalizedResponse.to_legacy_dict preserves shape
- raw_response stays Any (Pattern 3 preserved)
- tools field stays list[dict[str, Any]] for Phase 1 ToolSpec follow-up
MODIFIED src/openai_compatible.py:
- Removed inline NormalizedResponse + OpenAICompatibleRequest definitions
- Re-imported from src.openai_schemas
- _send_blocking: tool_calls -> tuple[ToolCall, ...]; usage_*_tokens -> UsageStats
- _send_streaming: same migration
- send_openai_compatible: messages_dicts = [m.to_dict() for m in request.messages]
- Exception handler: empty NormalizedResponse uses UsageStats
- All NormalizedResponse consumers still work (legacy dict shape preserved)
Verified:
uv run pytest tests/test_openai_schemas.py tests/test_mcp_tool_specs.py tests/test_audit_dataclass_coverage.py tests/test_type_aliases.py tests/test_mcp_client_beads.py tests/test_mcp_client_paths.py tests/test_arch_boundary_phase2.py --timeout=60
64 passed in 6.28s
105 lines
2.7 KiB
Python
105 lines
2.7 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
|
|
from typing import Any, Callable, Optional
|
|
|
|
|
|
@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) -> dict[str, Any]:
|
|
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
|
|
tool_calls: Optional[tuple[ToolCall, ...]] = None
|
|
tool_call_id: Optional[str] = None
|
|
name: Optional[str] = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
d: dict[str, Any] = {"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
|
|
raw_response: Any
|
|
|
|
def to_legacy_dict(self) -> dict[str, Any]:
|
|
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[dict[str, Any]]] = None
|
|
tool_choice: str = "auto"
|
|
stream: bool = False
|
|
stream_callback: Optional[Callable[[str], None]] = None
|
|
extra_body: Optional[dict[str, Any]] = None |