모델과 HTTP를 모르는
내부 추론 계약

학습에 필요한 전체 코드

from typing import Literal, Protocol

from pydantic import BaseModel, ConfigDict, Field


class Contract(BaseModel):
    """Internal Python data, not the browser API or an upstream JSON envelope."""

    model_config = ConfigDict(frozen=True, extra="forbid")


class Message(Contract):
    role: Literal["system", "user", "assistant"]
    content: str = Field(min_length=1)


class GenerationRequest(Contract):
    messages: tuple[Message, ...] = Field(min_length=1)
    max_output_tokens: int = Field(default=128, gt=0)
    temperature: float = Field(default=0.2, ge=0, le=2)


class ModelInfo(Contract):
    id: str = Field(min_length=1)


class TokenUsage(Contract):
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)
    total_tokens: int = Field(ge=0)
    cached_input_tokens: int = Field(ge=0)


class GenerationResult(Contract):
    id: str
    model: str
    text: str
    finish_reason: Literal["stop", "length"]
    usage: TokenUsage


InferenceErrorCode = Literal[
    "unavailable", "timeout", "busy", "request_rejected",
    "upstream_error", "invalid_response", "model_mismatch",
    "context_too_long", "invalid_message",
]


class InferenceError(Exception):
    """Safe error category; never carry raw prompts or upstream error messages."""

    def __init__(
        self, code: InferenceErrorCode, *, upstream_status: int | None = None
    ) -> None:
        self.code = code
        self.upstream_status = upstream_status
        super().__init__(f"Inference failed: {code}")


class InferenceEngine(Protocol):
    """What application code needs, independent of TurboFieldfare endpoints."""

    async def check_ready(self) -> ModelInfo: ...

    async def list_models(self) -> tuple[ModelInfo, ...]: ...

    async def generate(self, request: GenerationRequest) -> GenerationResult: ...

Contract: 모든 내부 데이터의 공통 규칙

값을 넣어 객체를 만드는 실행 가능한 예제

아래는 입력·출력 객체의 모양을 익히는 합성 예제다. 모델에 요청하거나 실제 응답을 측정한 것이 아니다. GenerationResult의 필수 필드인 id와 model도 생략하지 않았다.

from app.engines.base import GenerationRequest, GenerationResult, Message, TokenUsage

request = GenerationRequest(
    messages=(Message(role="user", content="안녕"),),
    max_output_tokens=16,
    temperature=0,
)
result = GenerationResult(
    id="chatcmpl-example",
    model="gemma-4-26b-a4b-it",
    text="안녕하세요!",
    finish_reason="stop",
    usage=TokenUsage(
        input_tokens=10, output_tokens=4,
        total_tokens=14, cached_input_tokens=0,
    ),
)
assert request.max_output_tokens == 16
assert result.usage.total_tokens == 14

Protocol은 상속 강제가 아닌 “필요한 모양”이다

InferenceEngine은 세 async method가 있다는 약속이다. TurboFieldfareAdapter가 이 class를 명시적으로 상속하지 않아도 같은 method signature를 제공하면 type checker가 호환 구현으로 볼 수 있다. 아래 예시는 readiness route만 검사하기 위한 부분 대역이다. 세 method 중 하나만 구현하므로 완전한 InferenceEngine 구현은 아니며, 생성 route까지 검사하려면 나머지 method도 구현해야 한다.

class ReadyEngine:
    async def check_ready(self) -> ModelInfo:
        return ModelInfo(id="gemma-4-26b-a4b-it")

따라서 readiness route test는 15GB 모델이나 network 없이 route의 200/503 분기만 검사할 수 있다. 이것이 dependency injection이 주는 격리다.

오류 code를 작은 집합으로 고정한 이유

TurboFieldfare가 보내는 구체 메시지를 모든 route가 그대로 처리하면 문구 변경이 곧 application 변경이 된다. InferenceError는 원인을 아홉 범주로 정규화한다. upstream_status는 진단용 숫자만 보존하고, prompt나 생성 문장이 포함될 수 있는 원래 오류 본문은 exception에 담지 않는다. 문맥 길이 초과는 context_too_long, 잘못된 메시지는 invalid_message로 구분하고, 서버가 요청 모델을 찾지 못한 경우는 기존 model_mismatch로 통일한다. 구체적인 upstream status·code 해석은 이 파일이 아닌 Adapter가 맡는다.

또한 InferenceErrorCode = Literal[...]는 정적 type 검사에 쓰는 선언이다. 일반 Python class인 InferenceError 생성자가 Literal 값을 런타임에 자동 검증하는 것은 아니다. 실제 Adapter는 정해진 범주만 생성하고, Pydantic의 ReadinessResponse가 외부 응답 field를 검증한다.