내부 생성 계약을
Swift HTTP API로 번역한다

import와 반환 type을 먼저 읽기

import asyncio
from typing import Literal, TypeVar

import httpx2
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator

from app.engines.base import (
    GenerationRequest, GenerationResult, InferenceError, InferenceErrorCode,
    ModelInfo, TokenUsage,
)

WireResponse = TypeVar("WireResponse", bound=_WireModel)

읽기 순서를 위해 import와 TypeVar 선언을 모아 표시했다. 실제 파일의 WireResponse 선언은 아래의 응답 model class 정의 뒤에 있다. asyncio는 비동기 시간 제한, httpx2는 HTTP 전송, Pydantic import는 JSON 구조 검증을 맡는다. base.py에서 가져온 type은 서비스 내부의 요청·결과·오류 계약이다.

TypeVar("WireResponse", bound=_WireModel)는 “_WireModel 계열 중 호출자가 선택한 type”을 나타낸다. schema: type[WireResponse]에 _Health class를 넣으면 _request의 반환 type도 _Health로 이어지도록 정적 검사기에 알려 준다. TypeVar가 JSON을 검증하거나 객체를 만드는 것은 아니며 실제 검증은 schema.model_validate_json(...)이 한다.

응답용 private Pydantic model

class _WireModel(BaseModel):
    """Validate the supported upstream subset; tolerate unrelated new fields."""

    model_config = ConfigDict(strict=True, extra="ignore")


class _Health(_WireModel):
    status: Literal["ok"]


class _Model(_WireModel):
    id: str = Field(min_length=1)
    object: Literal["model"]


class _Models(_WireModel):
    object: Literal["list"]
    data: list[_Model]


class _Message(_WireModel):
    role: Literal["assistant"]
    content: str
    tool_calls: list[dict] = Field(default_factory=list, max_length=0)


class _Choice(_WireModel):
    index: Literal[0]
    message: _Message
    finish_reason: Literal["stop", "length"]


class _PromptDetails(_WireModel):
    cached_tokens: int = Field(ge=0)


class _Usage(_WireModel):
    prompt_tokens: int = Field(ge=0)
    completion_tokens: int = Field(ge=0)
    total_tokens: int = Field(ge=0)
    prompt_tokens_details: _PromptDetails

    @model_validator(mode="after")
    def check_counts(self) -> "_Usage":
        if self.total_tokens != self.prompt_tokens + self.completion_tokens:
            raise ValueError("Inconsistent total token count")
        if self.prompt_tokens_details.cached_tokens > self.prompt_tokens:
            raise ValueError("Cached token count exceeds prompt token count")
        return self


class _Completion(_WireModel):
    id: str = Field(min_length=1)
    object: Literal["chat.completion"]
    model: str
    choices: list[_Choice] = Field(min_length=1, max_length=1)
    usage: _Usage

생성자: 두 client를 목적별로 보관한다

def __init__(
    self, generation_client: httpx2.AsyncClient, *,
    probe_client: httpx2.AsyncClient, model: str,
    probe_timeout: float, generation_timeout: float,
) -> None:
    self._generation_client = generation_client
    self._probe_client = probe_client
    self._model = model
    self._probe_timeout = probe_timeout
    self._generation_timeout = generation_timeout

generation_client는 답변 생성용, probe_client는 상태·모델 목록 확인용이다. * 뒤의 인수는 이름을 붙여 전달해야 하므로 조립 지점에서 probe_client=...가 드러난다. 이 함수는 client를 만들거나 연결을 열지 않고 참조와 설정값을 보관한다.

_request(..., client=...)는 사용할 client를 명시적으로 받는다. 따라서 공통 JSON 검증과 오류 처리 코드를 유지하면서도 두 연결 풀이 섞이지 않는다. 구체적인 생성·종료 책임은 main.py의 lifespan에 있다.

오류 JSON에서 안전한 종류만 뽑는다

class _ErrorDetail(_WireModel):
    # Deliberately exclude message/param: they can contain user content.
    code: str


class _ErrorEnvelope(_WireModel):
    error: _ErrorDetail


_KNOWN_REQUEST_ERRORS: dict[tuple[int, str], InferenceErrorCode] = {
    (400, "context_length_exceeded"): "context_too_long",
    (400, "invalid_message"): "invalid_message",
    (404, "model_not_found"): "model_mismatch",
}


def _error_code(response: httpx2.Response) -> InferenceErrorCode:
    """Keep only allowlisted categories; unknown/malformed bodies use status."""
    if response.status_code == 429:
        return "busy"
    if 400 <= response.status_code < 500:
        try:
            error = _ErrorEnvelope.model_validate_json(response.content)
        except ValidationError:
            return "request_rejected"
        return _KNOWN_REQUEST_ERRORS.get(
            (response.status_code, error.error.code), "request_rejected",
        )
    return "upstream_error"

공통 HTTP 경로 _request()

async def _request(
    self, method: str, path: str, schema: type[WireResponse], *,
    client: httpx2.AsyncClient, deadline: float, payload: dict | None = None,
) -> WireResponse:
    try:
        # A total deadline, including pool wait and upstream queue time.
        async with asyncio.timeout(deadline):
            response = await client.request(method, path, json=payload)
    except httpx2.PoolTimeout:
        raise InferenceError("busy") from None
    except (httpx2.TimeoutException, TimeoutError):
        raise InferenceError("timeout") from None
    except httpx2.RequestError:
        raise InferenceError("unavailable") from None

    if response.status_code != 200:
        raise InferenceError(
            _error_code(response), upstream_status=response.status_code,
        )
    try:
        return schema.model_validate_json(response.content)
    except ValidationError:
        # Do not leak validation errors: they may contain generated text.
        raise InferenceError("invalid_response") from None

Readiness: 생존과 model ID를 함께 확인한다

async def list_models(self) -> tuple[ModelInfo, ...]:
    result = await self._request(
        "GET", "/v1/models", _Models,
        client=self._probe_client, deadline=self._probe_timeout,
    )
    return tuple(ModelInfo(id=model.id) for model in result.data)

async def check_ready(self) -> ModelInfo:
    await self._request(
        "GET", "/health", _Health,
        client=self._probe_client, deadline=self._probe_timeout,
    )
    models = await self.list_models()
    for model in models:
        if model.id == self._model:
            return model
    raise InferenceError("model_mismatch")

/health만 보면 다른 설정의 TurboFieldfare가 같은 port에 떠 있어도 ready로 오판할 수 있다. 그래서 model 목록에 설정한 ID가 실제로 있는지까지 확인한다. 현재 설정의 기본값은 gemma-4-26b-a4b-it이며 이 method에 고정된 문자열은 아니다. 반대로 readiness에서 token 생성을 하지 않는 이유는 probe 한 번마다 수초의 GPU·SSD 작업과 queue 점유가 생기기 때문이다.

Generation: 내부 요청을 전송 JSON으로 바꾼다

async def generate(self, request: GenerationRequest) -> GenerationResult:
    payload = {
        "model": self._model,
        "messages": [message.model_dump() for message in request.messages],
        "max_completion_tokens": request.max_output_tokens,
        "temperature": request.temperature,
        "stream": False,
    }
    result = await self._request(
        "POST", "/v1/chat/completions", _Completion,
        client=self._generation_client,
        deadline=self._generation_timeout, payload=payload,
    )
    if result.model != self._model:
        raise InferenceError("model_mismatch")
    choice = result.choices[0]
    return GenerationResult(
        id=result.id, model=result.model, text=choice.message.content,
        finish_reason=choice.finish_reason,
        usage=TokenUsage(
            input_tokens=result.usage.prompt_tokens,
            output_tokens=result.usage.completion_tokens,
            total_tokens=result.usage.total_tokens,
            cached_input_tokens=result.usage.prompt_tokens_details.cached_tokens,
        ),
    )

실제 값의 변환

stream=false는 이번 단계의 의도적인 경계다. SSE는 chunk를 읽으면서 즉시 browser로 넘기고 client disconnect 때 upstream을 취소해야 하므로 별도의 lifecycle을 갖는다. 다음 streaming 단계에서 버퍼링 없이 추가한다.