OpenRouter 완전 가이드 4편 — 모니터링, 레이트 리밋 관리, OAuth PKCE, 팀 운영, ZDR

[4편 핵심 요약]
→ Generations API: 모든 요청의 상세 로그 — 모델·프로바이더·비용·레이턴시
→ 레이트 리밋: OpenRouter 자체 제한 없음 (유료 모델) — 프로바이더 한도가 실제 제한
→ 429 처리: exponential backoff + jitter + 폴백 모델 자동 전환
→ API 키 관리 API: 프로그래밍으로 키 생성·삭제·한도 설정
→ OAuth PKCE: 사용자가 자기 OpenRouter 계정으로 직접 인증 → 내 크레딧 소모 없음
→ 가드레일: 조직 멤버·키별 예산·모델 접근 제한 설정
→ ZDR: 프롬프트·응답 저장 안 하는 프로바이더만 라우팅
→ Broadcast: Langfuse·Datadog·Braintrust에 트레이스 동시 전송

 

---

실전 1 — Generations API로 사용량 모니터링

import httpx
import os
from datetime import datetime, timedelta
from collections import defaultdict

API_KEY = os.environ["OPENROUTER_API_KEY"]
BASE_URL = "https://openrouter.ai/api/v1"

def get_generations(
    limit: int = 100,
    offset: int = 0,
    date_start: str = None,
    date_end: str = None,
    model: str = None,
) -> list[dict]:
    """Generations API — 요청별 상세 로그 조회"""
    params = {"limit": limit, "offset": offset}
    if date_start:
        params["date_start"] = date_start
    if date_end:
        params["date_end"] = date_end
    if model:
        params["model"] = model

    response = httpx.get(
        f"{BASE_URL}/generations",
        headers={"Authorization": f"Bearer {API_KEY}"},
        params=params,
    )
    return response.json().get("data", [])

def build_cost_report(days: int = 7) -> dict:
    """
    최근 N일 비용 리포트
    - 모델별 비용
    - 시간대별 요청량
    - 에러율
    - 평균 레이턴시
    """
    end = datetime.now()
    start = end - timedelta(days=days)

    all_generations = []
    offset = 0
    while True:
        batch = get_generations(
            limit=100,
            offset=offset,
            date_start=start.isoformat(),
            date_end=end.isoformat(),
        )
        if not batch:
            break
        all_generations.extend(batch)
        offset += 100

    # 집계
    model_stats = defaultdict(lambda: {
        "requests": 0,
        "total_cost": 0.0,
        "input_tokens": 0,
        "output_tokens": 0,
        "errors": 0,
        "total_latency_ms": 0,
    })

    total_cost = 0.0
    error_count = 0

    for gen in all_generations:
        model = gen.get("model", "unknown")
        cost = gen.get("total_cost", 0) or 0
        status = gen.get("finish_reason", "stop")
        latency = gen.get("latency", 0) or 0

        model_stats[model]["requests"] += 1
        model_stats[model]["total_cost"] += cost
        model_stats[model]["input_tokens"] += gen.get("tokens_prompt", 0) or 0
        model_stats[model]["output_tokens"] += gen.get("tokens_completion", 0) or 0
        model_stats[model]["total_latency_ms"] += latency
        total_cost += cost

        if status in ("error", "cancelled"):
            model_stats[model]["errors"] += 1
            error_count += 1

    # 출력
    print(f"\n{'='*60}")
    print(f"📊 OpenRouter 사용량 리포트 (최근 {days}일)")
    print(f"{'='*60}")
    print(f"총 요청: {len(all_generations):,}")
    print(f"총 비용: ${total_cost:.4f}")
    print(f"에러율: {error_count/max(len(all_generations),1)*100:.1f}%")

    print(f"\n[모델별 비용 Top 5]")
    sorted_models = sorted(
        model_stats.items(),
        key=lambda x: x[1]["total_cost"],
        reverse=True
    )

    for model, stats in sorted_models[:5]:
        avg_latency = stats["total_latency_ms"] / max(stats["requests"], 1)
        cost_pct = stats["total_cost"] / max(total_cost, 0.0001) * 100
        print(f"\n  📌 {model}")
        print(f"     요청: {stats['requests']:,} | 비용: ${stats['total_cost']:.4f} ({cost_pct:.1f}%)")
        print(f"     토큰: {stats['input_tokens']:,} in / {stats['output_tokens']:,} out")
        print(f"     평균 레이턴시: {avg_latency:.0f}ms")

    return {
        "total_cost": total_cost,
        "total_requests": len(all_generations),
        "error_rate": error_count / max(len(all_generations), 1),
        "by_model": dict(sorted_models),
    }

# 실행
report = build_cost_report(days=7)

# 실시간 비용 알림
def check_credit_balance() -> dict:
    """현재 크레딧 잔액 확인"""
    response = httpx.get(
        f"{BASE_URL}/credits",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    data = response.json()
    return {
        "balance_usd": data.get("data", {}).get("credits", 0),
        "usage_usd": data.get("data", {}).get("usage", 0),
    }

balance = check_credit_balance()
print(f"잔액: ${balance['balance_usd']:.2f}")

# 잔액 낮으면 경고
if balance["balance_usd"] < 5.0:
    print("⚠️ 크레딧 부족! 충전 필요")
    # 슬랙 알림, 이메일 등 연동

---

실전 2 — 레이트 리밋 처리와 재시도 전략

import time
import random
import httpx
from openai import OpenAI, RateLimitError, APIStatusError

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

def exponential_backoff_request(
    messages: list[dict],
    model: str = "anthropic/claude-sonnet-4-6",
    fallback_models: list[str] = None,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
) -> str:
    """
    Exponential Backoff + Jitter + 폴백 모델 자동 전환
    """
    fallback_models = fallback_models or [
        "google/gemini-3-flash",
        "deepseek/deepseek-chat",
        "meta-llama/llama-3.3-70b-instruct:free",
    ]
    all_models = [model] + fallback_models
    current_model_idx = 0

    for attempt in range(max_retries):
        current_model = all_models[min(current_model_idx, len(all_models) - 1)]

        try:
            response = client.chat.completions.create(
                model=current_model,
                messages=messages,
                extra_body={
                    "models": all_models,  # OpenRouter 서버사이드 폴백도 동시 활성화
                }
            )
            if attempt > 0:
                print(f"✅ {attempt+1}번째 시도 성공 (모델: {response.model})")
            return response.choices[0].message.content

        except RateLimitError as e:
            # 429 — 레이트 리밋
            if attempt == max_retries - 1:
                raise

            # Exponential Backoff + Jitter
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            wait_time = delay + jitter

            print(f"⚠️ 429 레이트 리밋 (시도 {attempt+1}/{max_retries}). {wait_time:.1f}초 대기...")
            time.sleep(wait_time)

            # 다음 폴백 모델로 전환
            current_model_idx += 1

        except APIStatusError as e:
            if e.status_code == 500:
                # 서버 에러 — 짧게 대기 후 재시도
                delay = min(base_delay * (2 ** attempt), 10.0)
                print(f"⚠️ 500 서버 에러. {delay:.1f}초 대기...")
                time.sleep(delay)
                current_model_idx += 1
            elif e.status_code == 402:
                # 크레딧 부족
                raise Exception("❌ 크레딧 부족. 충전 필요") from e
            else:
                raise

    raise Exception(f"최대 재시도 횟수 초과 ({max_retries}번)")

# 사용
result = exponential_backoff_request(
    messages=[{"role": "user", "content": "안녕"}],
    model="anthropic/claude-opus-4-7",
    fallback_models=["anthropic/claude-sonnet-4-6", "openai/gpt-5.4"],
)
print(result)

# 비동기 버전 — 동시 다중 요청
import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

# 세마포어로 동시 요청 수 제한
semaphore = asyncio.Semaphore(10)  # 최대 10개 동시 요청

async def safe_request(prompt: str, model: str = "google/gemini-3-flash") -> str:
    async with semaphore:
        for attempt in range(3):
            try:
                response = await async_client.chat.completions.create(
                    model=model,
                    messages=[{"role": "user", "content": prompt}]
                )
                return response.choices[0].message.content
            except Exception as e:
                if attempt == 2:
                    return f"에러: {e}"
                await asyncio.sleep(2 ** attempt)
        return ""

async def batch_process(prompts: list[str]) -> list[str]:
    """여러 프롬프트 동시 처리 (레이트 리밋 준수)"""
    tasks = [safe_request(p) for p in prompts]
    return await asyncio.gather(*tasks)

# 실행
prompts = [f"문장 {i} 번역해줘" for i in range(50)]
results = asyncio.run(batch_process(prompts))

---

실전 3 — API 키 프로그래밍 관리

# Management API로 API 키 생성·삭제·한도 설정
# API 키 설정 → Settings → API Keys → Management API Key 발급 필요

MANAGEMENT_KEY = os.environ["OPENROUTER_MANAGEMENT_KEY"]

def create_api_key(
    name: str,
    credit_limit: float = None,
    daily_reset: bool = False,
) -> dict:
    """
    새 API 키 생성
    credit_limit: 최대 크레딧 한도 (달러)
    daily_reset: 매일 예산 초기화 여부
    """
    payload = {"name": name}
    if credit_limit:
        payload["limit"] = credit_limit
    if daily_reset:
        payload["limit_reset_period"] = "daily"  # or "weekly", "monthly"

    response = httpx.post(
        f"{BASE_URL}/keys",
        headers={
            "Authorization": f"Bearer {MANAGEMENT_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
    )
    data = response.json()
    return {
        "key": data.get("key"),          # 실제 API 키 (한 번만 노출)
        "hash": data.get("hash"),         # 이후 식별자
        "name": data.get("name"),
        "limit": data.get("limit"),
    }

def list_api_keys() -> list[dict]:
    """모든 API 키 목록 조회"""
    response = httpx.get(
        f"{BASE_URL}/keys",
        headers={"Authorization": f"Bearer {MANAGEMENT_KEY}"},
    )
    return response.json().get("data", [])

def delete_api_key(key_hash: str) -> bool:
    """API 키 삭제"""
    response = httpx.delete(
        f"{BASE_URL}/keys/{key_hash}",
        headers={"Authorization": f"Bearer {MANAGEMENT_KEY}"},
    )
    return response.status_code == 200

def update_key_limit(key_hash: str, new_limit: float) -> dict:
    """API 키 크레딧 한도 변경"""
    response = httpx.patch(
        f"{BASE_URL}/keys/{key_hash}",
        headers={
            "Authorization": f"Bearer {MANAGEMENT_KEY}",
            "Content-Type": "application/json",
        },
        json={"limit": new_limit},
    )
    return response.json()

# 사용 예시 — 환경별 키 자동 관리
def setup_environment_keys():
    """개발/스테이징/프로덕션 키 분리 세팅"""
    keys = {}

    # 개발 환경 — 낮은 한도
    keys["dev"] = create_api_key(
        name="dev-environment",
        credit_limit=5.0,       # $5 한도
        daily_reset=True,       # 매일 초기화
    )

    # 스테이징 — 중간 한도
    keys["staging"] = create_api_key(
        name="staging-environment",
        credit_limit=20.0,
    )

    # 프로덕션 — 높은 한도
    keys["prod"] = create_api_key(
        name="production-environment",
        credit_limit=500.0,
    )

    return keys

---

실전 4 — OAuth PKCE: 사용자가 자기 키로 직접 인증

내 크레딧을 소모하지 않고 사용자가 자기 OpenRouter 계정으로 직접 인증합니다. 사용자 비용은 자기 계정에서 차감됩니다.

// Next.js App Router 기준 OAuth PKCE 구현

// lib/oauth.ts — 유틸 함수
import { createHash, randomBytes } from "crypto";

// Code Verifier 생성 (랜덤 문자열)
export function generateCodeVerifier(): string {
  return randomBytes(32).toString("base64url");
}

// Code Challenge 생성 (SHA-256 해시)
export async function generateCodeChallenge(verifier: string): Promise {
  const data = new TextEncoder().encode(verifier);
  const hash = await crypto.subtle.digest("SHA-256", data);
  return Buffer.from(hash).toString("base64url");
}

// app/api/auth/openrouter/route.ts — 인증 시작
export async function GET(req: Request) {
  const codeVerifier = generateCodeVerifier();
  const codeChallenge = await generateCodeChallenge(codeVerifier);

  // code_verifier를 세션에 저장 (Step 2에서 사용)
  const response = new Response(null, { status: 302 });
  response.headers.set("Set-Cookie",
    `code_verifier=${codeVerifier}; HttpOnly; Secure; SameSite=Lax; Path=/; Max-Age=600`
  );

  // OpenRouter 인증 페이지로 리다이렉트
  const authUrl = new URL("https://openrouter.ai/auth");
  authUrl.searchParams.set("callback_url", `${process.env.NEXT_PUBLIC_URL}/api/auth/callback`);
  authUrl.searchParams.set("code_challenge", codeChallenge);
  authUrl.searchParams.set("code_challenge_method", "S256");

  response.headers.set("Location", authUrl.toString());
  return response;
}

// app/api/auth/callback/route.ts — 콜백 처리
export async function GET(req: Request) {
  const url = new URL(req.url);
  const code = url.searchParams.get("code");

  if (!code) {
    return Response.json({ error: "code 없음" }, { status: 400 });
  }

  // 쿠키에서 code_verifier 복원
  const cookies = Object.fromEntries(
    req.headers.get("cookie")?.split("; ")
      .map(c => c.split("=")) ?? []
  );
  const codeVerifier = cookies["code_verifier"];

  // code → API 키 교환
  const tokenResponse = await fetch("https://openrouter.ai/api/v1/auth/keys", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      code,
      code_verifier: codeVerifier,
    }),
  });

  const data = await tokenResponse.json();
  const userApiKey = data.key;  // 사용자의 개인 OpenRouter API 키

  // 이 키를 DB에 암호화 저장하거나 세션에 저장
  // 이후 이 키로 API 호출 시 사용자 크레딧에서 차감
  console.log("사용자 API 키 획득:", userApiKey.substring(0, 10) + "...");

  // 성공 페이지로 리다이렉트
  return Response.redirect(`${process.env.NEXT_PUBLIC_URL}/dashboard`);
}

# Python 서버에서 사용자 키로 요청
def call_with_user_key(user_api_key: str, prompt: str) -> str:
    """
    사용자 자신의 OpenRouter 키로 요청
    → 비용이 내 계정이 아닌 사용자 계정에서 차감
    """
    user_client = OpenAI(
        base_url="https://openrouter.ai/api/v1",
        api_key=user_api_key,  # 사용자 키 사용
    )

    response = user_client.chat.completions.create(
        model="anthropic/claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

[OAuth PKCE 활용 시나리오]

내 크레딧 소모 vs 사용자 크레딧 소모:

내 키(서버 키) 사용:
→ 내 크레딧 차감
→ 모든 사용자 요청 내가 지불
→ SaaS 앱에서 구독료로 커버

사용자 OAuth 키 사용:
→ 각 사용자 자신의 크레딧 차감
→ 내가 지불하지 않음
→ "BYOK 앱" 모델 — 사용자가 자기 키 가져오는 앱

어디에 쓰나:
→ 개발자 도구, IDE 확장 — 사용자가 직접 비용 지불
→ AI 크레딧이 없는 초기 스타트업
→ 무제한 요청 앱 (내가 비용 제한 어려울 때)

---

실전 5 — 가드레일(Guardrails)로 팀 예산 관리

[가드레일 설정 — OpenRouter 콘솔]
Settings → Organization → Guardrails → Create Guardrail

설정 가능한 항목:
────────────────────────────────────────────────────────
항목                    설명
────────────────────────────────────────────────────────
Budget Limit            크레딧 한도 (일/주/월 단위 리셋)
Allowed Models          사용 가능한 모델 화이트리스트
Blocked Models          사용 금지 모델 블랙리스트
Max Price               최대 토큰 가격 필터
Zero Data Retention     ZDR 프로바이더만 라우팅 강제
────────────────────────────────────────────────────────

적용 단위:
→ 조직 멤버별 가드레일
→ API 키별 가드레일
→ 더 엄격한 규칙이 항상 우선 적용

# 프로그래밍으로 가드레일 설정 (API)
def create_guardrail(
    name: str,
    credit_limit: float,
    reset_period: str = "monthly",
    allowed_models: list[str] = None,
    require_zdr: bool = False,
) -> dict:
    """
    가드레일 생성
    reset_period: "daily" | "weekly" | "monthly"
    """
    payload = {
        "name": name,
        "budget": {
            "limit": credit_limit,
            "reset_period": reset_period,
        }
    }

    if allowed_models:
        payload["allowed_models"] = allowed_models

    if require_zdr:
        payload["require_zdr"] = True

    response = httpx.post(
        f"{BASE_URL}/guardrails",
        headers={
            "Authorization": f"Bearer {MANAGEMENT_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
    )
    return response.json()

# 팀 역할별 가드레일 예시
guardrails = {
    # 주니어 개발자 — 저렴한 모델만, 낮은 예산
    "junior": create_guardrail(
        name="Junior Dev",
        credit_limit=20.0,
        reset_period="monthly",
        allowed_models=[
            "google/gemini-3-flash",
            "deepseek/deepseek-chat",
            "meta-llama/llama-3.3-70b-instruct",
        ]
    ),
    # 시니어 개발자 — 모든 모델, 더 큰 예산
    "senior": create_guardrail(
        name="Senior Dev",
        credit_limit=100.0,
        reset_period="monthly",
    ),
    # 프로덕션 키 — ZDR 필수
    "prod": create_guardrail(
        name="Production",
        credit_limit=500.0,
        reset_period="monthly",
        require_zdr=True,  # 프로덕션은 데이터 보존 금지
    ),
}

---

실전 6 — ZDR(Zero Data Retention) 설정

# ZDR — 프롬프트·응답을 프로바이더가 저장하지 않도록 보장
response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[{"role": "user", "content": "기밀 계약서 분석해줘..."}],
    extra_body={
        "provider": {
            # ZDR 지원 프로바이더만 라우팅
            # ZDR 지원 프로바이더가 없으면 요청 실패
            "data_collection": "deny"  # ZDR 강제
        }
    }
)

# ZDR 지원 프로바이더 목록 확인
def get_zdr_providers() -> list[dict]:
    """ZDR 지원 모델·프로바이더 목록"""
    response = httpx.get(
        f"{BASE_URL}/endpoints",
        headers={"Authorization": f"Bearer {API_KEY}"},
        params={"zdr": True}  # ZDR 지원만 필터링
    )
    return response.json().get("data", [])

zdr_models = get_zdr_providers()
print(f"ZDR 지원 모델 수: {len(zdr_models)}")

[ZDR이 필요한 케이스]
→ 법률·의료·금융 민감 데이터 처리
→ GDPR 컴플라이언스 요구
→ 기업 기밀 정보 분석
→ 프롬프트가 모델 학습에 사용되는 것을 방지

[ZDR 한계]
→ ZDR 지원 프로바이더가 모든 모델에 있지 않음
→ 일부 프리미엄 모델은 ZDR 미지원
→ 가드레일에서 ZDR 강제 시 선택 가능한 모델 수 줄어듦

---

실전 7 — Broadcast: 외부 옵저버빌리티 연동

# Langfuse로 모든 OpenRouter 요청 트레이싱
# Langfuse 대시보드에서 요청·비용·품질 모니터링

response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[{"role": "user", "content": "분석 태스크"}],
    extra_body={
        # Broadcast: OpenRouter가 외부 서비스에 트레이스 전송
        "broadcast": [
            {
                "provider": "langfuse",
                "config": {
                    "public_key": os.environ["LANGFUSE_PUBLIC_KEY"],
                    "secret_key": os.environ["LANGFUSE_SECRET_KEY"],
                    "host": "https://cloud.langfuse.com",
                }
            }
        ]
    }
)

# 또는 Datadog
response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[{"role": "user", "content": "분석 태스크"}],
    extra_body={
        "broadcast": [
            {
                "provider": "datadog",
                "config": {
                    "api_key": os.environ["DATADOG_API_KEY"],
                    "site": "datadoghq.com",
                }
            }
        ]
    }
)

---

실전 8 — Zero Completion Insurance

# Zero Completion Insurance — 응답 없을 시 비용 미청구
# 기본 활성화 (별도 설정 불필요)
# 단, 응답 일부만 왔다가 끊기면 부분 청구될 수 있음

# 응답 완전성 검증
def validated_request(prompt: str) -> str:
    response = client.chat.completions.create(
        model="anthropic/claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
    )

    choice = response.choices[0]
    finish_reason = choice.finish_reason

    if finish_reason == "stop":
        return choice.message.content
    elif finish_reason == "length":
        # 출력 토큰 한도 초과 — max_tokens 늘리거나 프롬프트 축소
        print("⚠️ max_tokens 초과로 잘렸음")
        return choice.message.content
    elif finish_reason == "content_filter":
        raise Exception("컨텐츠 필터에 걸림")
    else:
        raise Exception(f"비정상 종료: {finish_reason}")

---

4편 최종 — OpenRouter 프로덕션 체크리스트

[프로덕션 배포 전 최종 체크리스트]

API 키 보안:
☐ 환경 변수로만 관리 (코드에 절대 하드코딩 금지)
☐ 환경별 키 분리 (dev/staging/prod)
☐ 키별 credit_limit 설정
☐ 사용하지 않는 키 삭제

모니터링:
☐ Generations API로 일별 비용 추적 자동화
☐ 크레딧 잔액 알림 설정 (잔액 $X 이하 시 알림)
☐ 에러율 모니터링 (5% 이상이면 조사)
☐ Broadcast로 Langfuse 등 외부 옵저버빌리티 연동

레이트 리밋 대응:
☐ Exponential Backoff + Jitter 재시도 구현
☐ 폴백 모델 체인 설정
☐ 동시 요청 수 세마포어로 제한

팀 운영:
☐ 역할별 가드레일 생성 (예산·모델 제한)
☐ 가드레일을 멤버·키에 할당
☐ 월간 비용 리포트 자동화

컴플라이언스:
☐ 민감 데이터 처리 시 ZDR 활성화
☐ 데이터 리전 필요 시 regional routing 설정
☐ GDPR 환경이면 EU 리전 락 설정

---

관련 글

• OpenRouter 완전 가이드 1편 — API 키 하나로 300개 모델 쓰는 법
• OpenRouter 완전 가이드 2편 — 폴백 라우팅, 프로바이더 제어, 비용 최적화
• OpenRouter 완전 가이드 3편 — 스트리밍, 툴 콜링, 멀티모달, LangGraph 통합