File size: 3,815 Bytes
6172a47 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """Pydantic models for Anthropic-compatible requests."""
from enum import StrEnum
from typing import Any, Literal
from loguru import logger
from pydantic import BaseModel, field_validator, model_validator
from config.settings import Settings, get_settings
# =============================================================================
# Content Block Types
# =============================================================================
class Role(StrEnum):
user = "user"
assistant = "assistant"
system = "system"
class ContentBlockText(BaseModel):
type: Literal["text"]
text: str
class ContentBlockImage(BaseModel):
type: Literal["image"]
source: dict[str, Any]
class ContentBlockToolUse(BaseModel):
type: Literal["tool_use"]
id: str
name: str
input: dict[str, Any]
class ContentBlockToolResult(BaseModel):
type: Literal["tool_result"]
tool_use_id: str
content: str | list[Any] | dict[str, Any]
class ContentBlockThinking(BaseModel):
type: Literal["thinking"]
thinking: str
class SystemContent(BaseModel):
type: Literal["text"]
text: str
# =============================================================================
# Message Types
# =============================================================================
class Message(BaseModel):
role: Literal["user", "assistant"]
content: (
str
| list[
ContentBlockText
| ContentBlockImage
| ContentBlockToolUse
| ContentBlockToolResult
| ContentBlockThinking
]
)
reasoning_content: str | None = None
class Tool(BaseModel):
name: str
description: str | None = None
input_schema: dict[str, Any]
class ThinkingConfig(BaseModel):
enabled: bool = True
# =============================================================================
# Request Models
# =============================================================================
class MessagesRequest(BaseModel):
model: str
max_tokens: int | None = None
messages: list[Message]
system: str | list[SystemContent] | None = None
stop_sequences: list[str] | None = None
stream: bool | None = True
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
metadata: dict[str, Any] | None = None
tools: list[Tool] | None = None
tool_choice: dict[str, Any] | None = None
thinking: ThinkingConfig | None = None
extra_body: dict[str, Any] | None = None
original_model: str | None = None
resolved_provider_model: str | None = None
@model_validator(mode="after")
def map_model(self) -> MessagesRequest:
"""Map any Claude model name to the configured model (model-aware)."""
settings = get_settings()
if self.original_model is None:
self.original_model = self.model
resolved_full = settings.resolve_model(self.original_model)
self.resolved_provider_model = resolved_full
self.model = Settings.parse_model_name(resolved_full)
if self.model != self.original_model:
logger.debug(f"MODEL MAPPING: '{self.original_model}' -> '{self.model}'")
return self
class TokenCountRequest(BaseModel):
model: str
messages: list[Message]
system: str | list[SystemContent] | None = None
tools: list[Tool] | None = None
thinking: ThinkingConfig | None = None
tool_choice: dict[str, Any] | None = None
@field_validator("model")
@classmethod
def validate_model_field(cls, v: str, info) -> str:
"""Map any Claude model name to the configured model (model-aware)."""
settings = get_settings()
resolved_full = settings.resolve_model(v)
return Settings.parse_model_name(resolved_full)
|