fahmiaziz98
[UPDATE] Refactoring code, dependencies, routers and exception
0231daa
raw
history blame
5.23 kB
"""
Common schemas shared across request and response models.
This module contains Pydantic models used by both requests and responses,
such as SparseEmbedding and ModelInfo.
"""
from typing import List, Optional, Literal
from pydantic import BaseModel, Field, ConfigDict
class SparseEmbedding(BaseModel):
"""
Sparse embedding representation.
Sparse embeddings are represented as two parallel arrays:
- indices: positions of non-zero values
- values: the actual values at those positions
Attributes:
indices: List of indices for non-zero elements
values: List of values corresponding to the indices
text: Optional original text that was embedded
"""
indices: List[int] = Field(
..., description="Indices of non-zero elements in the sparse vector"
)
values: List[float] = Field(..., description="Values corresponding to the indices")
text: Optional[str] = Field(None, description="Original text that was embedded")
class Config:
json_schema_extra = {
"example": {
"indices": [10, 25, 42, 100],
"values": [0.85, 0.62, 0.91, 0.73],
"text": "example query text",
}
}
class ModelInfo(BaseModel):
"""
Information about an available model.
Attributes:
id: Unique identifier for the model
name: Full model name (e.g., Hugging Face model path)
type: Model type ('embeddings' or 'sparse-embeddings')
loaded: Whether the model is currently loaded in memory
"""
id: str = Field(..., description="Unique model identifier")
name: str = Field(..., description="Full model name")
type: str = Field(..., description="Model type (embeddings or sparse-embeddings)")
loaded: bool = Field(..., description="Whether model is loaded in memory")
class Config:
json_schema_extra = {
"example": {
"id": "qwen3-0.6b",
"name": "Qwen/Qwen3-Embedding-0.6B",
"type": "embeddings",
"loaded": True,
}
}
class HealthStatus(BaseModel):
"""
Health check status information.
Attributes:
status: Overall status (ok or error)
total_models: Total number of configured models
loaded_models: Number of models currently loaded
startup_complete: Whether startup sequence is complete
"""
status: str = Field(..., description="Overall status")
total_models: int = Field(..., description="Total configured models")
loaded_models: int = Field(..., description="Currently loaded models")
startup_complete: bool = Field(..., description="Startup completion status")
class Config:
json_schema_extra = {
"example": {
"status": "ok",
"total_models": 2,
"loaded_models": 2,
"startup_complete": True,
}
}
class ErrorResponse(BaseModel):
"""
Standard error response format.
Attributes:
error: Error type/name
message: Detailed error message
detail: Additional error details (optional)
"""
error: str = Field(..., description="Error type")
message: str = Field(..., description="Error message")
detail: Optional[str] = Field(None, description="Additional details")
class Config:
json_schema_extra = {
"example": {
"error": "ModelNotFoundError",
"message": "Model 'unknown-model' not found in configuration",
"detail": "Available models: qwen3-0.6b, splade-pp-v2",
}
}
class EmbeddingOptions(BaseModel):
"""
Optional parameters for embedding generation.
These parameters are passed directly to the underlying sentence-transformers
model. Not all parameters work with all models - check model documentation.
Attributes:
normalize_embeddings: L2 normalize output embeddings
batch_size: Batch size for processing
convert_to_numpy: Return numpy arrays instead of lists
precision: Computation precision
"""
normalize_embeddings: Optional[bool] = Field(
None, description="L2 normalize the output embeddings"
)
batch_size: Optional[int] = Field(
None, ge=1, le=256, description="Batch size for processing texts"
)
convert_to_numpy: Optional[bool] = Field(
None, description="Return numpy arrays instead of Python lists"
)
precision: Optional[Literal["float32", "int8", "uint8", "binary", "ubinary"]] = (
Field(None, description="Computation precision for embeddings")
)
model_config = ConfigDict(
extra="forbid", # Prevent typos in field names
json_schema_extra={
"example": {
"normalize_embeddings": True,
"batch_size": 32,
}
},
)
def to_kwargs(self) -> dict:
"""
Convert options to kwargs dict, excluding None values.
Returns:
Dictionary of non-None parameters
"""
return {k: v for k, v in self.model_dump().items() if v is not None}