브라우저가 부를 수 있는
첫 번째 API

전체 구조에서의 위치

경로가 /chat으로 선언돼 있는데 실제로는 /api/v1/chat이다. 접두사는 router.py가 붙인다. 이 파일은 자기가 어느 버전 아래에 놓일지 모른다.

전체 코드

import logging
from typing import Annotated, Final

from fastapi import APIRouter, Depends, status
from fastapi.responses import JSONResponse

from app.api.dependencies import get_inference_engine
from app.engines.base import (
    GenerationRequest,
    InferenceEngine,
    InferenceError,
    InferenceErrorCode,
    Message,
)
from app.schemas.chat import ChatError, ChatRequest, ChatResponse, ChatUsage


logger = logging.getLogger(__name__)

# Readiness answers 503 for every failure because it asks one yes/no question.
# Chat separates them so the browser can tell "fix your input" from "retry later".
_ERROR_STATUS: Final[dict[InferenceErrorCode, int]] = {
    "invalid_message": status.HTTP_400_BAD_REQUEST,
    "request_rejected": status.HTTP_400_BAD_REQUEST,
    "context_too_long": status.HTTP_413_CONTENT_TOO_LARGE,
    "busy": status.HTTP_429_TOO_MANY_REQUESTS,
    "timeout": status.HTTP_504_GATEWAY_TIMEOUT,
    "unavailable": status.HTTP_503_SERVICE_UNAVAILABLE,
    "model_mismatch": status.HTTP_503_SERVICE_UNAVAILABLE,
    "upstream_error": status.HTTP_502_BAD_GATEWAY,
    "invalid_response": status.HTTP_502_BAD_GATEWAY,
}

# The upstream never reports a queue estimate, so this is a fixed conservative hint.
_RETRY_AFTER_SECONDS: Final = 2

router = APIRouter(tags=["chat"])


def _error_response(error: InferenceError) -> JSONResponse:
    """Map one engine failure to a status, keeping upstream detail server-side."""
    http_status = _ERROR_STATUS.get(error.code, status.HTTP_502_BAD_GATEWAY)
    logger.warning(
        "chat_inference_failed code=%s upstream_status=%s http_status=%s",
        error.code, error.upstream_status, http_status,
    )
    headers = (
        {"Retry-After": str(_RETRY_AFTER_SECONDS)}
        if http_status == status.HTTP_429_TOO_MANY_REQUESTS
        else None
    )
    return JSONResponse(
        status_code=http_status,
        content=ChatError(error_code=error.code).model_dump(),
        headers=headers,
    )


@router.post(
    "/chat",
    response_model=ChatResponse,
    status_code=status.HTTP_200_OK,
    responses={code: {"model": ChatError} for code in sorted(set(_ERROR_STATUS.values()))},
    summary="Generate one answer without storing the conversation",
)
async def chat(
    request: ChatRequest,
    engine: Annotated[InferenceEngine, Depends(get_inference_engine)],
) -> ChatResponse | JSONResponse:
    """Translate one browser request into the engine contract and back."""
    try:
        result = await engine.generate(
            GenerationRequest(
                messages=tuple(
                    Message(role=message.role, content=message.content)
                    for message in request.messages
                )
            )
        )
    except InferenceError as error:
        return _error_response(error)

    return ChatResponse(
        id=result.id,
        model=result.model,
        text=result.text,
        finish_reason=result.finish_reason,
        usage=ChatUsage(
            input_tokens=result.usage.input_tokens,
            output_tokens=result.usage.output_tokens,
            total_tokens=result.usage.total_tokens,
            cached_input_tokens=result.usage.cached_input_tokens,
        ),
    )

오류 매핑표 — 이 파일의 핵심

# Readiness answers 503 for every failure because it asks one yes/no question.
# Chat separates them so the browser can tell "fix your input" from "retry later".
_ERROR_STATUS: Final[dict[InferenceErrorCode, int]] = {
    "invalid_message": status.HTTP_400_BAD_REQUEST,
    "request_rejected": status.HTTP_400_BAD_REQUEST,
    "context_too_long": status.HTTP_413_CONTENT_TOO_LARGE,
    "busy": status.HTTP_429_TOO_MANY_REQUESTS,
    "timeout": status.HTTP_504_GATEWAY_TIMEOUT,
    "unavailable": status.HTTP_503_SERVICE_UNAVAILABLE,
    "model_mismatch": status.HTTP_503_SERVICE_UNAVAILABLE,
    "upstream_error": status.HTTP_502_BAD_GATEWAY,
    "invalid_response": status.HTTP_502_BAD_GATEWAY,
}

같은 InferenceError를 두 route가 다르게 해석한다. 이것이 route 계층이 하는 일을 가장 잘 보여주는 지점이다.

/readyz는 “준비됐나?”라는 예/아니오 하나를 묻는다. 그래서 모든 실패가 503이다. 채팅은 브라우저가 4xx(내 잘못)와 5xx(서버 잘못)를 구분해 재시도할지 안내할지 정해야 하므로 갈랐다.

변환 함수 — 정보를 줄여서 내보낸다

def _error_response(error: InferenceError) -> JSONResponse:
    """Map one engine failure to a status, keeping upstream detail server-side."""
    http_status = _ERROR_STATUS.get(error.code, status.HTTP_502_BAD_GATEWAY)
    logger.warning(
        "chat_inference_failed code=%s upstream_status=%s http_status=%s",
        error.code, error.upstream_status, http_status,
    )
    headers = (
        {"Retry-After": str(_RETRY_AFTER_SECONDS)}
        if http_status == status.HTTP_429_TOO_MANY_REQUESTS
        else None
    )
    return JSONResponse(
        status_code=http_status,
        content=ChatError(error_code=error.code).model_dump(),
        headers=headers,
    )

WARNING app.api.routes.chat chat_inference_failed code=unavailable upstream_status=None http_status=503

route 함수 — 성공 경로

@router.post(
    "/chat",
    response_model=ChatResponse,
    status_code=status.HTTP_200_OK,
    responses={code: {"model": ChatError} for code in sorted(set(_ERROR_STATUS.values()))},
    summary="Generate one answer without storing the conversation",
)
async def chat(
    request: ChatRequest,
    engine: Annotated[InferenceEngine, Depends(get_inference_engine)],
) -> ChatResponse | JSONResponse:

try:
        result = await engine.generate(
            GenerationRequest(
                messages=tuple(
                    Message(role=message.role, content=message.content)
                    for message in request.messages
                )
            )
        )
    except InferenceError as error:
        return _error_response(error)

ChatMessage를 Message로 한 개씩 바꾼다. 필드가 같아서 불필요해 보이지만, 이 한 줄이 브라우저 계약과 엔진 계약을 분리해 둔 대가이자 이득이다. 7단계에서 브라우저 메시지에 id가 붙어도 엔진 계약은 그대로다.

생성 옵션을 넘기지 않는다. GenerationRequest의 기본값 max_output_tokens=128, temperature=0.2가 적용된다. 서버가 소유한다는 결정이며, ChatRequest의 extra="forbid"가 client의 지정을 막는다.

관찰된 한계 — 128 토큰은 짧다

실제로 써 보니 답변이 자주 잘린다. 두 번째와 세 번째 요청 모두 finish_reason이 length였다.

usage: {"output_tokens": 128, ...}
finish_reason: "length"
text 끝: "... ### 1. 핵심 개념: 왜 Transformer인가?\n\n"   ← 문장 도중에 멈춤

GenerationRequest의 기본값 128이 짧은 답변 한 문단 정도라서 그렇다. 실제 채팅으로 쓰려면 512~1024 정도가 필요하다. 다만 8GB 환경에서 토큰당 약 0.4초가 걸리므로, 1024 토큰이면 한 답변에 7분이 걸린다. 4단계의 streaming이 필요한 이유가 여기서 분명해진다 — 완성될 때까지 기다리는 방식으로는 쓸 수 없다.

이 값을 설정으로 빼서 조절할지는 다음 작업에서 정할 항목이다.

자주 발생하는 오류

설계 선택과 대안

이전 단계와 다음 파일

이 route가 검증에 쓰는 계약은 schemas/chat.py, 번역해서 넘기는 내부 계약은 engines/base.py, 실제로 Gemma를 부르는 구현은 turbofieldfare.py에 있다. 경로에 /api/v1을 붙이는 곳은 router.py다.

다음 단계는 streaming이다. 위에서 본 대로 완성을 기다리는 방식은 실사용이 어렵다. 4단계에서 base.py에 조각 타입이 추가되고, 이 파일에 StreamingResponse를 쓰는 route가 생긴다.