191 lines
7.3 KiB
Python
191 lines
7.3 KiB
Python
"""
|
|
OpenAI API compatible data models
|
|
Data model definitions compatible with the OpenAI API specification.
|
|
"""
|
|
|
|
from typing import List, Optional, Dict, Any, Union, Literal
|
|
from pydantic import BaseModel, Field
|
|
from datetime import datetime
|
|
|
|
|
|
class ChatMessageContentText(BaseModel):
|
|
"""Text content model."""
|
|
|
|
type: Literal["text"] = Field("text", description="Content type")
|
|
text: str = Field(..., description="Text content")
|
|
|
|
|
|
class ChatMessageContentImageUrl(BaseModel):
|
|
"""Image URL model."""
|
|
|
|
url: str = Field(..., description="Image URL or base64 data")
|
|
local_path: Optional[str] = Field(None, description="Image path already saved within the workspace")
|
|
|
|
|
|
class ChatMessageContentImage(BaseModel):
|
|
"""Image content model."""
|
|
|
|
type: Literal["image_url"] = Field("image_url", description="Content type")
|
|
image_url: ChatMessageContentImageUrl = Field(..., description="Image URL")
|
|
|
|
|
|
# Multimodal content type.
|
|
ChatMessageContent = Union[
|
|
str, List[Union[ChatMessageContentText, ChatMessageContentImage]]
|
|
]
|
|
|
|
|
|
class ChatMessage(BaseModel):
|
|
"""Conversation message model."""
|
|
|
|
role: Literal["system", "user", "assistant", "tool"] = Field(
|
|
..., description="Message role"
|
|
)
|
|
content: Optional[ChatMessageContent] = Field(None, description="Message content")
|
|
name: Optional[str] = Field(None, description="Sender name")
|
|
tool_calls: Optional[List[Dict[str, Any]]] = Field(None, description="Tool calls")
|
|
tool_call_id: Optional[str] = Field(None, description="Tool call ID")
|
|
|
|
|
|
class ChatCompletionRequest(BaseModel):
|
|
"""Chat completion request model."""
|
|
|
|
model: str = Field(..., description="Model name")
|
|
messages: List[ChatMessage] = Field(..., description="Conversation message list")
|
|
temperature: Optional[float] = Field(1.0, ge=0, le=2, description="Generation temperature")
|
|
top_p: Optional[float] = Field(1.0, ge=0, le=1, description="Nucleus sampling parameter")
|
|
n: Optional[int] = Field(1, ge=1, le=128, description="Number of generations")
|
|
stream: Optional[bool] = Field(False, description="Whether to stream output")
|
|
stop: Optional[Union[str, List[str]]] = Field(None, description="Stop words")
|
|
max_tokens: Optional[int] = Field(None, ge=1, description="Maximum token count")
|
|
presence_penalty: Optional[float] = Field(0, ge=-2, le=2, description="Presence penalty")
|
|
frequency_penalty: Optional[float] = Field(0, ge=-2, le=2, description="Frequency penalty")
|
|
logit_bias: Optional[Dict[str, float]] = Field(None, description="Logit bias")
|
|
user: Optional[str] = Field(None, description="User identifier")
|
|
tools: Optional[List[Dict[str, Any]]] = Field(None, description="Available tools")
|
|
tool_choice: Optional[Union[str, Dict[str, Any]]] = Field(
|
|
None, description="Tool selection strategy"
|
|
)
|
|
|
|
|
|
class Usage(BaseModel):
|
|
"""Token usage statistics."""
|
|
|
|
prompt_tokens: int = Field(..., description="Prompt token count")
|
|
completion_tokens: int = Field(..., description="Completion token count")
|
|
total_tokens: int = Field(..., description="Total token count")
|
|
|
|
|
|
class ChatChoice(BaseModel):
|
|
"""Chat choice result."""
|
|
|
|
index: int = Field(..., description="Choice index")
|
|
message: ChatMessage = Field(..., description="Generated message")
|
|
finish_reason: Optional[
|
|
Literal["stop", "length", "tool_calls", "content_filter"]
|
|
] = Field(None, description="Finish reason")
|
|
|
|
|
|
class ChatCompletionResponse(BaseModel):
|
|
"""Chat completion response model."""
|
|
|
|
id: str = Field(..., description="Request ID")
|
|
object: Literal["chat.completion"] = Field(
|
|
"chat.completion", description="Object type"
|
|
)
|
|
created: int = Field(..., description="Creation timestamp")
|
|
model: str = Field(..., description="Model name")
|
|
choices: List[ChatChoice] = Field(..., description="Generated choice list")
|
|
usage: Usage = Field(..., description="Token usage statistics")
|
|
system_fingerprint: Optional[str] = Field(None, description="System fingerprint")
|
|
|
|
|
|
class DeltaMessage(BaseModel):
|
|
"""Incremental message for streaming output."""
|
|
|
|
role: Optional[Literal["system", "user", "assistant", "tool"]] = Field(
|
|
None, description="Message role"
|
|
)
|
|
content: Optional[str] = Field(None, description="Message content")
|
|
tool_calls: Optional[List[Dict[str, Any]]] = Field(None, description="Tool calls")
|
|
|
|
|
|
class ChatCompletionChunkChoice(BaseModel):
|
|
"""Choice chunk for streaming output."""
|
|
|
|
index: int = Field(..., description="Choice index")
|
|
delta: DeltaMessage = Field(..., description="Incremental message")
|
|
finish_reason: Optional[
|
|
Literal["stop", "length", "tool_calls", "content_filter"]
|
|
] = Field(None, description="Finish reason")
|
|
|
|
|
|
class ChatCompletionChunk(BaseModel):
|
|
"""Response chunk for streaming output."""
|
|
|
|
id: str = Field(..., description="Request ID")
|
|
object: Literal["chat.completion.chunk"] = Field(
|
|
"chat.completion.chunk", description="Object type"
|
|
)
|
|
created: int = Field(..., description="Creation timestamp")
|
|
model: str = Field(..., description="Model name")
|
|
choices: List[ChatCompletionChunkChoice] = Field(..., description="Choice chunk list")
|
|
usage: Optional[Usage] = Field(None, description="Token usage statistics")
|
|
system_fingerprint: Optional[str] = Field(None, description="System fingerprint")
|
|
|
|
|
|
class ModelInfo(BaseModel):
|
|
"""Model information."""
|
|
|
|
id: str = Field(..., description="Model ID")
|
|
object: Literal["model"] = Field("model", description="Object type")
|
|
created: int = Field(..., description="Creation timestamp")
|
|
owned_by: str = Field(..., description="Owner")
|
|
permission: Optional[List[Dict[str, Any]]] = Field(None, description="Permission information")
|
|
root: Optional[str] = Field(None, description="Root model")
|
|
parent: Optional[str] = Field(None, description="Parent model")
|
|
|
|
|
|
class ModelListResponse(BaseModel):
|
|
"""Model list response."""
|
|
|
|
object: Literal["list"] = Field("list", description="Object type")
|
|
data: List[ModelInfo] = Field(..., description="Model list")
|
|
|
|
|
|
class ErrorDetail(BaseModel):
|
|
"""Error details."""
|
|
|
|
message: str = Field(..., description="Error message")
|
|
type: str = Field(..., description="Error type")
|
|
param: Optional[str] = Field(None, description="Error parameter")
|
|
code: Optional[str] = Field(None, description="Error code")
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""Error response."""
|
|
|
|
error: ErrorDetail = Field(..., description="Error details")
|
|
|
|
|
|
# Health check response.
|
|
class HealthResponse(BaseModel):
|
|
"""Health check response."""
|
|
|
|
status: Literal["ok"] = Field("ok", description="Service status")
|
|
timestamp: str = Field(..., description="Timestamp")
|
|
version: str = Field(..., description="Version information")
|
|
agent_name: str = Field(..., description="Agent name")
|
|
|
|
|
|
# Server information response.
|
|
class ServerInfoResponse(BaseModel):
|
|
"""Server information response."""
|
|
|
|
name: str = Field(..., description="Service name")
|
|
version: str = Field(..., description="Version information")
|
|
description: str = Field(..., description="Service description")
|
|
api_version: str = Field(..., description="API version")
|
|
supported_models: List[str] = Field(..., description="Supported model list")
|
|
capabilities: List[str] = Field(..., description="Service capability list")
|