knowledge-engine / api /schemas /archive.py
m97j's picture
feat: authenticated archive inference with bounded CPU execution
f16b727 verified
Raw History Blame Contribute Delete
1.76 kB
"""Small, versioned machine contracts; never accept arbitrary model loading."""
import math
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
InputText = Annotated[str, Field(min_length=1, max_length=8000)]
class TextBatch(BaseModel):
model_config = ConfigDict(extra="forbid")
texts: list[InputText] = Field(min_length=1, max_length=64)
@field_validator("texts")
@classmethod
def bounded_texts(cls, values):
if any(not text.strip() for text in values):
raise ValueError("Empty text is not supported")
if sum(len(text.encode("utf-8")) for text in values) > 64000:
raise ValueError("Combined text exceeds 64000 UTF-8 bytes")
return values
class EmbeddingsRequest(TextBatch):
model: str | None = Field(default=None, max_length=200)
class Embedding(BaseModel):
dense: list[float] = Field(min_length=1, max_length=4096)
sparse: dict[str, float] = Field(max_length=8192)
@model_validator(mode="after")
def finite_vectors(self):
if any(not math.isfinite(value) for value in self.dense):
raise ValueError("Invalid dense vector")
if any(not key.isdecimal() or int(key) < 0 or not math.isfinite(value)
or value < 0 for key, value in self.sparse.items()):
raise ValueError("Invalid sparse vector")
return self
class EmbeddingsResponse(BaseModel):
model: str
dimension: int = Field(ge=1, le=4096)
embeddings: list[Embedding]
class TranslationRequest(TextBatch):
source_locale: Literal["ko", "en"]
target_locale: Literal["ko", "en"]
class TranslationResponse(BaseModel):
model: str
translations: list[str]