LLM 게이트웨이 직접 구현: 스트리밍부터 사용량 기록까지

이 글에서 직접 구현할 LLM 서버는 모델 가중치를 GPU에 올리는 추론 서버가 아닙니다. 여러 외부 또는 내부 모델 엔드포인트 앞에서 인증, 모델 선택, 스트리밍, 장애 처리, 사용량 기록을 담당하는 애플리케이션 계층의 LLM Gateway입니다. 모델 추론 자체는 vLLM 같은 별도 엔진이나 상용 API가 담당한다고 가정합니다.

완성된 제품을 그대로 대체하는 범용 Gateway를 만드는 것이 목표도 아닙니다. 두 개의 모델 공급자와 세 개의 내부 서비스가 있고, 조직 고유의 권한 및 정산 규칙 때문에 작은 Gateway가 필요한 상황을 기준으로 설계합니다.

먼저 고정할 경계

직접 구현에서 가장 중요한 결정은 지원할 기능보다 지원하지 않을 기능입니다. 첫 버전의 범위를 다음처럼 제한할 수 있습니다.

• 입력은 대화형 텍스트 메시지만 받습니다.
• 논리 모델 별칭 세 개만 공개합니다.
• 공급자는 두 곳만 연결합니다.
• 일반 JSON 응답과 SSE 스트리밍을 지원합니다.
• 도구 호출, 이미지, 오디오, 배치는 지원하지 않습니다.
• 애플리케이션은 공급자 이름과 실제 모델 ID를 알 수 없습니다.

이 제한이 없으면 Gateway는 빠르게 모든 공급자의 전체 API를 재현하는 프로젝트가 됩니다. 직접 구현의 장점은 필요한 계약만 작게 유지할 수 있다는 데 있습니다.

예시 외부 계약은 다음과 같습니다.

POST /v1/generate
Authorization: Bearer <client-key>
X-Request-Id: <request-id>
Content-Type: application/json

{
  "model": "support-fast",
  "messages": [
    {"role": "user", "content": "배송 상태를 알려주세요"}
  ],
  "stream": true
}

support-fast는 실제 공급자 모델명이 아니라 Gateway가 소유한 논리 별칭입니다. 애플리케이션이 임의의 base_url이나 공급자 모델명을 보내지 못하게 해야 라우팅 정책과 SSRF 방어 경계가 유지됩니다.

최소 구성 요소

Gateway는 단순 HTTP 중계기보다 다음 책임을 가진 작은 플랫폼에 가깝습니다.

Client
  -> Authentication and Policy
  -> Model Catalog
  -> Router
  -> Provider Adapter
  -> Provider API or Inference Server

Every request
  -> Request Ledger
  -> Metrics and Audit Events

각 구성 요소의 역할은 분리하는 편이 좋습니다.

구성 요소	책임
Authentication	클라이언트 키 검증과 주체 식별
Policy	허용 모델, 예산, RPM, TPM, 동시 요청 수 검사
Model Catalog	논리 별칭을 실제 배포 후보 목록으로 변환
Router	건강 상태와 우선순위에 따라 한 배포 대상 선택
Provider Adapter	공급자별 요청, 응답, 오류, 사용량을 공통 형식으로 변환
Request Ledger	논리 요청과 실제 공급자 호출 시도를 각각 기록
Observability	지연 시간, 첫 토큰 시간, 오류율, 스트림 중단 측정

HTTP 엔드포인트가 이 책임을 모두 직접 가지면 공급자 하나를 추가할 때 인증과 정산 코드까지 함께 흔들립니다. 엔드포인트는 검증과 응답 전송만 맡고, 라우팅과 공급자 변환은 별도 객체로 두는 편이 안전합니다.

공급자 어댑터의 계약

공급자별 차이는 Gateway 내부에서도 사라지지 않습니다. 차이를 한곳에 가두기 위해 어댑터가 공통 이벤트를 반환하도록 정의할 수 있습니다.

from collections.abc import AsyncIterator
from dataclasses import dataclass
from typing import Literal, Protocol

from pydantic import BaseModel


class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str


class GenerateRequest(BaseModel):
    model: str
    messages: list[Message]
    stream: bool = False


@dataclass(frozen=True)
class StreamChunk:
    kind: Literal["delta", "usage"]
    text: str | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None


class ProviderAdapter(Protocol):
    async def stream(
        self,
        deployment: "Deployment",
        payload: GenerateRequest,
    ) -> AsyncIterator[StreamChunk]: ...

실제 어댑터는 공급자 요청 형식으로 메시지를 변환하고, 스트림 조각을 StreamChunk로 정규화합니다. 공급자 오류도 RateLimited, ProviderUnavailable, InvalidRequest, PolicyRejected 같은 내부 오류로 매핑합니다. 원본 오류 본문에는 프롬프트나 내부 식별자가 포함될 수 있으므로 그대로 클라이언트에 반환하지 않습니다.

공통 계약에 없는 공급자 기능을 무작정 버리는 것도 피해야 합니다. 필요한 기능이 생기면 내부 계약의 새 버전으로 명시적으로 추가합니다. 임의의 공급자 매개변수를 통과시키는 extra 객체는 편리하지만, Gateway의 안정된 경계를 다시 공급자 API에 종속시킬 수 있습니다.

라우팅은 모델 선택이 아니라 배포 선택

Model Catalog는 논리 모델 하나에 여러 배포 후보를 연결합니다.

support-fast
  1. provider-a / region-1 / model-small
  2. provider-a / region-2 / model-small
  3. provider-b / region-1 / model-fast

Router는 요청마다 다음 순서로 후보를 줄일 수 있습니다.

1. 클라이언트가 사용할 수 없는 모델과 리전을 제외합니다.
2. 회로 차단 상태이거나 최근 건강 검사에 실패한 배포를 제외합니다.
3. TPM, RPM, 동시 요청 한도를 넘은 배포를 제외합니다.
4. 우선순위와 가중치에 따라 하나를 선택합니다.

처음부터 지연 시간 기반의 복잡한 최적화를 도입할 필요는 없습니다. 정적 우선순위와 단순 가중치만으로 시작하고, 실제 측정 데이터가 쌓인 뒤 정책을 바꾸는 편이 낫습니다. 라우팅 결정에는 선택한 정책 버전과 배포 ID를 남겨야 나중에 응답 품질과 비용을 설명할 수 있습니다.

스트리밍 경로 구현

SSE는 HTTP 응답을 유지하면서 text/event-stream 형식의 이벤트를 차례로 전송합니다. 각 이벤트는 빈 줄로 구분합니다. FastAPI의 StreamingResponse는 비동기 반복자가 내보내는 값을 응답 본문으로 스트리밍할 수 있습니다.

아래 코드는 핵심 제어 흐름만 보여주는 뼈대입니다. 인증 저장소, 정책 엔진, 실제 어댑터와 영속화 구현은 의도적으로 생략했습니다.

import asyncio
import json
from collections.abc import AsyncIterator
from uuid import uuid4

from fastapi import Depends, FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()


def encode_sse(event: str, data: dict) -> bytes:
    body = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
    return f"event: {event}\ndata: {body}\n\n".encode("utf-8")


class GatewayService:
    async def stream(
        self,
        request_id: str,
        principal: "Principal",
        payload: GenerateRequest,
    ) -> AsyncIterator[StreamChunk]:
        candidates = await router.candidates(principal, payload.model)

        for attempt_number, deployment in enumerate(candidates, start=1):
            emitted = False
            attempt_status = "failed"
            error_code = None
            attempt = await ledger.start_attempt(
                request_id=request_id,
                attempt_number=attempt_number,
                deployment_id=deployment.id,
            )

            try:
                adapter = adapters[deployment.provider]
                async for chunk in adapter.stream(deployment, payload):
                    emitted = True
                    yield chunk

                attempt_status = "succeeded"
                return
            except asyncio.CancelledError:
                attempt_status = "cancelled"
                raise
            except RetryableProviderError as error:
                error_code = error.code
                last_candidate = attempt_number == len(candidates)
                if emitted or last_candidate:
                    raise GatewayUnavailable() from error
            except ProviderError as error:
                error_code = error.code
                raise GatewayUpstreamError() from error
            finally:
                await ledger.finish_attempt(
                    attempt.id,
                    status=attempt_status,
                    error_code=error_code,
                )


gateway = GatewayService()


@app.post("/v1/generate")
async def generate(
    payload: GenerateRequest,
    request: Request,
    principal: "Principal" = Depends(authenticate),
):
    await policies.authorize(principal, payload)
    request_id = request.headers.get("x-request-id") or str(uuid4())
    await ledger.start_request(request_id, principal.id, payload.model)

    async def events() -> AsyncIterator[bytes]:
        try:
            async for chunk in gateway.stream(request_id, principal, payload):
                if await request.is_disconnected():
                    await ledger.cancel_request(request_id)
                    return

                if chunk.kind == "delta":
                    yield encode_sse("delta", {"text": chunk.text})
                else:
                    yield encode_sse(
                        "usage",
                        {
                            "input_tokens": chunk.input_tokens,
                            "output_tokens": chunk.output_tokens,
                        },
                    )

            await ledger.finish_request(request_id, status="succeeded")
            yield encode_sse("done", {"request_id": request_id})
        except GatewayError as error:
            await ledger.finish_request(
                request_id,
                status="failed",
                error_code=error.code,
            )
            yield encode_sse(
                "error",
                {"request_id": request_id, "code": error.public_code},
            )

    return StreamingResponse(
        events(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Request-Id": request_id,
        },
    )

이 예제에서 중요한 부분은 프레임 형식보다 제어 흐름입니다. 클라이언트 연결이 끊기면 공급자 스트림도 취소할 수 있도록 비동기 호출 체인이 취소를 전달해야 합니다. 공유 HTTP 클라이언트를 애플리케이션 수명 주기에 맞춰 재사용하고, 요청마다 새 연결 풀을 만들지 않는 것도 중요합니다.

스트리밍 응답이 시작된 뒤에는 이미 HTTP 상태 코드와 헤더가 전송됐습니다. 이후 공급자 오류가 발생해도 상태 코드를 500으로 바꿀 수 없으므로, 위 예제처럼 명시적인 error 이벤트를 보내고 스트림을 종료해야 합니다. 스트리밍이 아닌 요청은 응답을 시작하기 전까지 오류를 확정할 수 있으므로 일반 HTTP 오류 상태를 사용할 수 있습니다.

리버스 프록시나 로드 밸런서가 응답을 버퍼링하면 애플리케이션이 조각을 보내도 클라이언트에는 한꺼번에 도착할 수 있습니다. 운영 환경에서는 응답 버퍼링, 유휴 연결 제한, 전체 요청 제한을 스트리밍 특성에 맞춰 별도로 확인해야 합니다.

재시도의 안전 경계

일반 HTTP 요청은 실패하면 다시 보내기 쉽지만, LLM 스트림은 일부 출력이 전달되는 순간 되돌릴 수 없습니다. 따라서 기본 규칙은 단순해야 합니다.

클라이언트에 첫 이벤트를 보낸 뒤에는 다른 모델로 자동 전환하지 않습니다.

첫 출력 전 연결 오류나 429, 일시적인 5xx는 다음 후보로 재시도할 수 있습니다. 첫 출력 후 오류는 error 이벤트를 보내고 스트림을 끝냅니다. 이미 전달된 문장 뒤에 다른 모델의 답변을 이어 붙이면 하나의 응답처럼 보여도 의미와 안전 정책이 섞일 수 있습니다.

첫 출력 전 재시도도 완전히 무료는 아닙니다. Gateway가 응답을 받지 못했더라도 공급자가 요청을 수락해 비용을 계산했을 수 있습니다. 클라이언트의 동일 요청 재전송을 막는 멱등 키와, 실제 공급자 호출 횟수를 기록하는 attempt ledger가 모두 필요한 이유입니다.

재시도는 개수보다 시간 예산으로도 제한해야 합니다. 예를 들어 전체 20초 제한 안에서 첫 후보에 8초, 다음 후보에 남은 시간을 배분할 수 있습니다. 각 시도가 독립적으로 20초를 사용하면 장애 때 지연 시간이 후보 수만큼 늘어납니다.

사용량은 요청과 시도를 분리해 기록

논리 요청 하나가 여러 공급자 시도를 만들 수 있으므로 단일 테이블에 결과를 덮어쓰면 비용과 장애 원인을 잃습니다. 최소한 두 수준으로 기록합니다.

gateway_requests
  request_id, client_id, model_alias, policy_version,
  status, started_at, first_token_at, finished_at

provider_attempts
  request_id, attempt_number, deployment_id, provider_request_id,
  status, input_tokens, output_tokens, estimated_cost, error_code

사용량 이벤트가 스트림 마지막에만 오는 공급자가 있고, 중간에 연결이 끊기면 정확한 토큰 수를 받지 못할 수도 있습니다. 이때 값을 0으로 확정하지 말고 unknown, estimated, confirmed 상태를 구분하는 편이 좋습니다. 공급자 청구 데이터와 나중에 대조할 수 있도록 공급자 요청 ID도 저장합니다.

프롬프트와 응답 전문은 사용량 장부의 필수 필드가 아닙니다. 기본 로그에서는 제외하고, 디버깅을 위해 보관해야 할 때만 별도 동의, 마스킹, 암호화, 짧은 보존 기간을 적용합니다. 요청 ID와 모델, 토큰, 지연 시간만으로도 대부분의 운영 지표를 만들 수 있습니다.

인증과 정책

클라이언트에는 공급자 API 키 대신 Gateway 전용 키를 발급합니다. 저장소에는 원문이 아니라 단방향 해시를 보관하고, 키에는 다음 속성을 연결합니다.

• 허용된 논리 모델 목록
• 만료 시각과 활성 상태
• 분당 요청 및 토큰 한도
• 최대 동시 스트림 수
• 일별 또는 월별 예산
• 데이터 처리 리전과 로깅 정책

요청의 모든 헤더를 공급자에게 전달해서는 안 됩니다. 어댑터가 필요한 헤더만 새로 만들고, 공급자 자격 증명은 서버의 비밀 저장소에서 읽습니다. 사용자가 보낸 URL, 헤더, 공급자 키 이름을 그대로 네트워크 요청에 반영하지 않는 것이 안전한 기본값입니다.

예산 검사는 요청 전에 예상치를 기준으로 한 번 수행하고, 완료 후 실제 사용량으로 정산할 수 있습니다. 스트리밍 중 토큰이 계속 늘어나므로 엄격한 상한이 필요하다면 최대 출력 토큰을 요청 정책으로 제한해야 합니다.

타임아웃과 관측 지표

LLM 요청에는 하나의 타임아웃보다 단계별 제한이 유용합니다.

• 연결 타임아웃: 공급자와 TCP 또는 TLS 연결을 맺는 시간
• 첫 토큰 타임아웃: 요청 후 첫 출력이 도착할 때까지의 시간
• 스트림 유휴 타임아웃: 다음 조각이 오지 않는 최대 시간
• 전체 타임아웃: 재시도를 포함한 논리 요청의 총시간

평균 응답 시간만 보면 스트리밍 품질을 설명하기 어렵습니다. 다음 지표를 배포 대상과 모델 별칭별로 수집하는 편이 좋습니다.

• 첫 토큰까지 걸린 시간
• 초당 출력 토큰 수
• 정상 완료율과 중간 중단율
• 429 및 5xx 비율
• 재시도와 대체 경로 사용률
• 입력 및 출력 토큰과 추정 비용
• 클라이언트 취소율
• 정책 및 예산 거부율

Gateway 자체의 오류와 공급자 오류도 구분해야 합니다. 인증 실패, 정책 거부, 라우팅 후보 없음, 공급자 제한, 공급자 장애를 같은 500 오류로 합치면 운영자가 대응할 수 없습니다.

단계적으로 구현하는 순서

첫 버전부터 범용 AI Gateway를 만들 필요는 없습니다.

1. 공통 요청과 응답 계약, 공급자 어댑터 두 개를 구현합니다. 자동 대체는 아직 넣지 않습니다.
2. 논리 모델 별칭과 정적 우선순위 Router를 추가합니다. 요청과 공급자 시도 장부를 분리합니다.
3. 가상 클라이언트 키, 모델 권한, RPM, 동시 스트림 제한을 적용합니다.
4. 첫 출력 전 재시도, 회로 차단, 건강 상태 기반 제외를 추가합니다.
5. 실제 사용 데이터가 쌓인 뒤 예산, 비용 기반 라우팅, 관리 화면을 검토합니다.

각 단계에는 부하 테스트보다 먼저 실패 테스트가 필요합니다. 첫 조각 전 연결 종료, 중간 스트림 종료, 느린 클라이언트, 클라이언트 취소, 공급자 429, 사용량 이벤트 누락을 재현해야 합니다. 정상 응답만 확인한 Gateway는 실제 장애에서 가장 중요한 장부와 상태 전이를 잃기 쉽습니다.

직접 구현하지 않는 편이 나은 경우

공급자와 엔드포인트가 계속 늘고, 팀별 예산과 가상 키, 가드레일, 관측 도구 연동까지 빠르게 필요하다면 직접 구현의 범위가 제품 하나만큼 커집니다. LiteLLM 같은 기존 Gateway는 이미 공급자 변환, Router, 가상 키, 비용 추적, 속도 제한, 관리 기능을 제공합니다.

반대로 공급자가 두세 곳으로 고정돼 있고, 내부 인증과 정산 규칙이 독특하며, 지원할 API 계약을 작게 유지할 수 있다면 직접 구현이 합리적일 수 있습니다. 판단 기준은 코드 몇 줄을 줄이는가가 아니라 다음 질문입니다.

공급자 차이를 따라가는 비용과 Gateway를 운영하는 비용 중 어느 쪽을 조직이 더 잘 통제할 수 있습니까?

직접 구현을 선택한다면 범용 호환성을 목표로 삼기보다, 좁고 버전이 명확한 내부 계약을 만드는 편이 성공 가능성이 높습니다.

마무리

LLM Gateway의 핵심은 요청을 다른 URL로 전달하는 코드가 아닙니다. 공급자 자격 증명을 격리하고, 논리 모델을 실제 배포에 연결하고, 스트림의 취소와 실패를 처리하고, 한 논리 요청에서 발생한 모든 비용을 설명할 수 있어야 합니다.

작게 시작하더라도 인증, 라우팅, 공급자 어댑터, 스트리밍, 요청 장부를 처음부터 분리하면 기능이 늘어날 때 경계를 유지할 수 있습니다. 특히 첫 출력 이후에는 자동 재시도하지 않는 규칙과 요청 및 시도 기록의 분리는 운영 사고를 막는 가장 중요한 출발점입니다.

참고 자료

• LiteLLM 공식 시작 문서
• LiteLLM Router 공식 문서
• LiteLLM Fallback 공식 문서
• LiteLLM 가상 키 공식 문서
• FastAPI StreamingResponse 공식 문서
• Starlette Request 공식 문서
• WHATWG Server-Sent Events 표준
• HTTPX 비동기 클라이언트 공식 문서