기본 콘텐츠로 건너뛰기

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...

Python에서 asyncio 완전 정복 (await, async, gather 등)

어휴, 요즘 파이썬으로 비동기 프로그래밍 하는 재미에 푹 빠졌어요! 특히 asyncio 는 정말 마법 같더라고요. 처음엔 좀 낯설었는데, 익숙해지니까 속도 향상이 눈에 띄게 느껴져서 완전 반해버렸습니다. 이 글에선 제가 asyncio 를 배우면서 깨달은 점들을 풀어놓을게요. 혹시 비동기 프로그래밍이 뭔지 잘 모르시겠다면, 간단히 말해 여러 작업을 동시에 처리해서 프로그램 속도를 엄청나게 높이는 기술이라고 생각하시면 돼요. 마치 여러 요리사가 동시에 음식을 만들어서 손님에게 빨리 제공하는 것과 비슷하죠! 일단 async 와 await 라는 녀석들이 핵심인데요, async 는 함수 앞에 붙여서 "얘는 비동기 함수야!"라고 선언하는 거예요. 그리고 await 는 다른 비동기 함수가 끝날 때까지 기다리라고 지시하는 역할을 하죠. 예를 들어, 네트워크에서 데이터를 가져오는 함수가 있다면, await 를 사용해서 데이터가 다 가져올 때까지 기다렸다가 다음 작업을 진행할 수 있어요. 그 동안 다른 작업을 처리할 수 있으니, 마치 멀티태스킹을 하는 것처럼 느껴져요. 신기하지 않나요? 그리고 asyncio.gather 는 여러 비동기 함수를 동시에 실행하고 결과를 모아주는 아주 유용한 친구입니다. 제가 웹사이트 여러 개에서 데이터를 동시에 가져와야 할 때 정말 요긴하게 썼어요. 하나씩 순서대로 가져오는 것보다 훨씬 빠르더라고요! 마치 여러 개의 탭을 동시에 열어놓고 작업하는 것과 같다고 생각하시면 될 것 같아요. 실제로 제가 썼던 코드를 보여드릴게요. 세 개의 웹사이트에서 데이터를 가져오는 예제인데요. (아래 코드 삽입) 이 코드를 보시면, fetch_data 함수가 각 웹사이트에서 데이터를 가져오는 역할을 하고, asyncio.gather 가 이 함수들을 동시에 실행하도록 도와주는 것을 볼 수 있을 거예요. asyncio.sleep(2) 는 네트워크 지연을 시뮬레이션하기 위해 넣...