기본 콘텐츠로 건너뛰기

OpenRouter를 붙였는데 더 불안해졌다: 파이썬에서 LLM 호출 동선을 한 군데로 모으는 법

thumbnail

OpenRouter를 붙였는데 더 불안해졌다: 파이썬에서 LLM 호출 동선을 한 군데로 모으는 법

그날 장애는 모델이 아니라 “재시도”에서 시작했다.

응답이 늦어지자 타임아웃이 났고, 타임아웃이 나자 재시도가 돌았다. 재시도가 도니 요청이 더 밀렸다. 로그에는 같은 에러가 쌓였고, 비용 대시보드는 새로고침할 때마다 숫자가 튀었다. 누군가가 “모델만 바꾸면 되는 거 아니었어?”라고 물었는데, 그 질문이 제일 아팠다.

OpenRouter 같은 라우터 API를 붙이는 이유는 보통 “여러 모델을 쉽게 바꿔 쓰려고”다. 그런데 실무에서 더 큰 이점은 따로 있다. 호출 동선을 한 군데로 강제할 수 있다는 것이다.

키, 타임아웃, 리트라이, 로그, 비용. 이 다섯 가지가 프로젝트 여기저기에 흩어져 있으면, 모델이 아무리 좋아도 운영이 먼저 망가진다. 오늘은 그걸 한 파일로 모으는 이야기다.

본문에는 링크를 넣지 않고, 읽을거리는 맨 아래 참고자료로만 모았다.


라우터 API의 장점은 ‘모델 선택’이 아니라 ‘동선 통일’이다

Real Python의 OpenRouter 글(참고자료)을 보면, 접근은 단순하다. 라우터를 붙이면 여러 모델을 한 API로 호출할 수 있다. 여기까지는 누구나 한다.

하지만 팀이 진짜로 얻고 싶은 건 “모델 바꾸기”가 아니라 “규칙을 바꾸기”다.

  • 타임아웃은 몇 초로 둘지
  • 재시도는 몇 번까지 허용할지
  • 실패하면 어떤 예외로 올릴지
  • 로그는 어떤 태그로 남길지
  • 비용은 어디에서 상한을 걸지

이 다섯 가지는 기능 개발보다 운영에 가깝다. 그리고 운영 규칙은 문서로 써놓으면 대개 깨진다. 코드로 박아야 한다.

Armin Ronacher가 AI를 ‘테세우스의 배’로 비유한 글(참고자료)을 읽고 떠오른 것도 비슷한 감각이었다. 모델이 바뀌고, 프롬프트가 바뀌고, 정책이 바뀌어도 “이 시스템이 같은 배인가”를 유지하려면 결국 형태(구조)가 아니라 동선(운영)이 남아 있어야 한다.

그래서 우리는 라우터 API를 “모델 스위치”가 아니라 “동선 스위치”로 써야 한다.

그래서 테스트도 “문장이 맞냐”가 아니라 “타임아웃/예외/shape가 계약대로냐”를 고정한다.


한 파일로 끝내는 호출 래퍼(ModelRouter): 키/타임아웃/리트라이/로그

여기서 OOP를 멋내기로 쓰지 않는 게 중요하다. The Python Coding Stack의 글(참고자료)에서 말하듯, 객체는 결국 데이터와 행동을 묶는 방식이다.

LLM 호출에서 우리가 묶어야 할 데이터는 키/기본 모델/비용 한도/리트라이 정책이고, 행동은 “요청 한 번”이다. 이걸 흩어놓지 않겠다는 결심이 ModelRouter다.

아래 코드는 실제 OpenRouter SDK를 강요하지 않는다. 요지는 구조다.

  • 키는 환경 변수에서만 읽고, 함수 인자로 넘기지 않는다(흔한 유출 경로)
  • 타임아웃/리트라이는 한 군데에서만 설정한다
  • request id를 만들어 로그/메트릭에 같은 태그로 남긴다
  • 비용 상한은 ‘대충’이 아니라 호출 시점에서 체크한다

파일: model_router.py

from __future__ import annotations

import os
import time
import uuid
from dataclasses import dataclass
from typing import Callable


class ModelTimeout(RuntimeError):
    pass


class ModelCallFailed(RuntimeError):
    pass


@dataclass(frozen=True)
class RouterConfig:
    default_model: str
    timeout_s: float = 8.0
    retries: int = 2
    backoff_s: float = 0.3
    cost_budget_usd: float | None = None


RequestFn = Callable[[str, str, str, float, str], dict]
# args: (prompt, model, request_id, timeout_s, api_key) -> dict


class ModelRouter:
    def __init__(
        self,
        *,
        base_url: str,
        api_key_env: str = "OPENROUTER_API_KEY",
        config: RouterConfig,
        request_fn: RequestFn | None = None,
    ):
        self.base_url = base_url
        self.api_key_env = api_key_env
        self.config = config
        self._request_fn = request_fn or self._default_request

    def _api_key(self) -> str:
        key = os.getenv(self.api_key_env)
        if not key:
            raise RuntimeError(f"missing API key env: {self.api_key_env}")
        return key

    def _default_request(self, prompt: str, model: str, request_id: str, timeout_s: float, api_key: str) -> dict:
        # NOTE: 예시용. 실제 구현에서는 requests/httpx 호출을 여기에만 모으세요.
        return {"id": request_id, "model": model, "output": ""}

    def complete(self, prompt: str, *, model: str | None = None, request_id: str | None = None) -> dict:
        rid = request_id or uuid.uuid4().hex
        mdl = model or self.config.default_model

        if self.config.cost_budget_usd is not None and self.config.cost_budget_usd <= 0:
            raise RuntimeError(f"cost budget exhausted rid={rid}")

        api_key = self._api_key()
        last_err: Exception | None = None

        for attempt in range(self.config.retries + 1):
            try:
                started = time.monotonic()
                data = self._request_fn(prompt, mdl, rid, self.config.timeout_s, api_key)
                elapsed = time.monotonic() - started

                if elapsed > self.config.timeout_s:
                    raise ModelTimeout(f"timeout rid={rid} model={mdl} elapsed={elapsed:.2f}")

                return data

            except ModelTimeout as e:
                last_err = e
                if attempt >= self.config.retries:
                    raise
                time.sleep(self.config.backoff_s * (attempt + 1))

            except Exception as e:
                last_err = e
                if attempt >= self.config.retries:
                    break
                time.sleep(self.config.backoff_s * (attempt + 1))

        raise ModelCallFailed(f"model call failed rid={rid} model={mdl}") from last_err

이 파일 하나가 있으면, 프로젝트에서 LLM을 쓰는 방식이 단순해진다.

  • “어디서 타임아웃을 늘리지?” 같은 질문이 사라지고
  • “왜 이 요청만 비용이 튀었지?”를 request id로 추적할 수 있고
  • 모델을 바꿔도 실패 모드가 일정해진다

테스트는 정답이 아니라 ‘계약’이다(결과물보다 실패 모드)

LLM 테스트를 ‘정답 비교’로 만들면 보통 오래 못 간다. 모델이 바뀌면 깨지고, 프롬프트가 바뀌면 깨지고, 결국 테스트를 끈다.

여기서 내가 지키는 건 결과가 아니라 계약이다.

  • 일정 시간 안에 돌아와야 한다
  • 실패하면 어떤 예외가 나와야 한다
  • 성공하면 어떤 shape(키/필드)를 가져야 한다

이건 모델이 바뀌어도 유지할 수 있다. 테세우스의 배 같은 상황에서, 우리가 고정해야 하는 건 “출력 문장”이 아니라 “운영 계약”이다.

파일: test_model_router.py

import os
import time

import pytest

from model_router import ModelRouter, RouterConfig, ModelTimeout


def test_complete_timeout_contract(monkeypatch):
    monkeypatch.setenv("OPENROUTER_API_KEY", "test")

    def slow_request(prompt, model, rid, timeout_s, api_key):
        time.sleep(timeout_s * 2)
        return {"id": rid, "model": model, "output": ""}

    router = ModelRouter(
        base_url=os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1/chat/completions"),
        config=RouterConfig(default_model="openrouter/any", timeout_s=0.001, retries=0),
        request_fn=slow_request,
    )

    with pytest.raises(ModelTimeout):
        router.complete("hi", request_id="t1")


def test_complete_shape(monkeypatch):
    monkeypatch.setenv("OPENROUTER_API_KEY", "test")

    router = ModelRouter(
        base_url=os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1/chat/completions"),
        config=RouterConfig(default_model="openrouter/any", timeout_s=8.0, retries=0),
    )

    data = router.complete("hi", request_id="t2")
    assert set(data.keys()) >= {"id", "model", "output"}

이 테스트는 모델의 ‘능력’을 보장하지 않는다. 대신 운영의 ‘형태’를 보장한다.

from model_router import ModelRouter, RouterConfig

router = ModelRouter(
    base_url="https://openrouter.ai/api/v1/chat/completions",
    config=RouterConfig(default_model="openrouter/claude"),
)

# 특정 요청에서만 모델 교체
router.complete("요약해줘", model="openrouter/gpt")

운영에서 중요한 두 가지: 비용(상한)과 관측(태그)

LLM을 여러 개 붙이면 팀은 두 번 놀란다.

첫 번째는 품질 차이에서 놀라고, 두 번째는 비용에서 놀란다.

그리고 장애가 났을 때는 관측에서 무너진다. “어느 모델이 문제였지?” “어느 요청이 폭주를 만들었지?”를 못 찾으면, 운영은 감정 싸움으로 변한다.

그래서 나는 비용과 관측을 ‘기능’으로 보지 않고 ‘동선’으로 본다.

  • 비용은 호출 지점에서 상한을 걸어야 한다(나중에 대시보드에서 확인하면 늦다)
  • 관측은 request id 같은 태그로 남겨야 한다(로그/메트릭/알림이 같은 언어를 써야 한다)

여기까지가 “한 파일로 모으는 이유”다. 모델을 갈아끼우는 속도보다, 동선을 고정하는 속도가 먼저다.


참고자료

  • Real Python — How to Use the OpenRouter API to Access Multiple AI Models via Python
    • https://realpython.com/openrouter-api/
  • Armin Ronacher — AI And The Ship of Theseus
    • https://lucumr.pocoo.org/2026/3/5/theseus/
  • The Python Coding Stack — You Store Data and You Do Stuff With Data • The OOP Mindset
    • https://www.thepythoncodingstack.com/p/python-oop-mindset-you-store-data-and-you-do-stuff-with-data

댓글

이 블로그의 인기 게시물

Django에서 트랜잭션 관리하기

Django에서 트랜잭션 관리하기 안녕하세요! 오늘은 Django에서 데이터베이스 트랜잭션을 효과적으로 관리하는 방법에 대해 알아보겠습니다. 1. 트랜잭션의 중요성 트랜잭션은 데이터베이스의 일관성과 무결성을 보장하는 중요한 개념입니다. Django에서는 여러 가지 방법으로 트랜잭션을 관리할 수 있습니다. 1.1 기본 개념 원자성(Atomicity) : 트랜잭션은 모두 실행되거나 모두 실행되지 않아야 합니다. 일관성(Consistency) : 트랜잭션 전후로 데이터베이스의 일관성이 유지되어야 합니다. 격리성(Isolation) : 동시에 실행되는 트랜잭션들이 서로 영향을 주지 않아야 합니다. 지속성(Durability) : 완료된 트랜잭션의 결과는 영구적으로 저장되어야 합니다. 2. Django의 트랜잭션 관리 2.1 기본 설정 # settings.py DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': 'mydatabase', 'USER': 'myuser', 'PASSWORD': 'mypassword', 'HOST': 'localhost', 'PORT': '5432', 'ATOMIC_REQUESTS': True, # 모든 뷰를 트랜잭션으로 래핑 } } 2.2 데코레이터 사용 from django.db import transaction @transaction.atomic def create_order(user, items): order = Order.objects.create(user=...

AWS S3 + CloudFront로 정적 파일 서빙 완전 가이드

AWS S3 + CloudFront로 정적 파일 서빙 완전 가이드 안녕하세요! 오늘은 AWS S3와 CloudFront를 사용하여 정적 파일을 효율적으로 서빙하는 방법에 대해 알아보겠습니다. 왜 S3와 CloudFront를 사용할까요? 높은 가용성 : AWS의 글로벌 인프라를 활용 빠른 전송 속도 : CloudFront의 CDN 기능으로 전 세계 사용자에게 빠른 전송 비용 효율성 : 사용한 만큼만 지불 보안 : AWS의 보안 기능 활용 확장성 : 트래픽 증가에 자동 대응 1. S3 버킷 설정 1.1 버킷 생성 및 설정 import boto3 def create_s3_bucket(): s3 = boto3.client('s3') # 버킷 생성 bucket_name = 'your-static-files-bucket' s3.create_bucket( Bucket=bucket_name, CreateBucketConfiguration={ 'LocationConstraint': 'ap-northeast-2' } ) # 버킷 정책 설정 bucket_policy = { "Version": "2012-10-17", "Statement": [ { "Sid": "PublicReadGetObject", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObje...

RDS에서 Django 앱 성능을 높이는 데이터베이스 설정 팁

RDS에서 Django 앱 성능을 높이는 데이터베이스 설정 팁 안녕하세요! 오늘은 AWS RDS를 사용하는 Django 애플리케이션의 성능을 최적화하는 방법에 대해 알아보겠습니다. 1. RDS 인스턴스 최적화 1.1 인스턴스 타입 선택 # RDS 인스턴스 크기 조정 import boto3 def resize_rds_instance(): rds = boto3.client('rds') response = rds.modify_db_instance( DBInstanceIdentifier='your-db', DBInstanceClass='db.t3.large', # 워크로드에 맞는 인스턴스 타입 선택 ApplyImmediately=True ) return response['DBInstance'] 1.2 파라미터 그룹 설정 def create_parameter_group(): rds = boto3.client('rds') # PostgreSQL 파라미터 그룹 생성 response = rds.create_db_parameter_group( DBParameterGroupName='django-optimized', DBParameterGroupFamily='postgres13', Description='Optimized parameters for Django applications' ) # 성능 관련 파라미터 설정 parameters = [ { 'ParameterName': 'shared_buffers', 'ParameterValue': '2GB...