AI API를 프로덕션 환경에서 운영할 때 가장 중요한 것 중 하나는 호출 로그를 체계적으로 기록하고 비용을 정확하게 추적하는 것입니다. 이 튜토리얼에서는 HolySheep AI를 기반으로 한 AI API 호출 로그 및 비용 추적 시스템을 구축하는 방법을 상세히 설명합니다.

AI API 서비스 비교 분석

비교 항목 HolySheep AI 공식 OpenAI API 기타 릴레이 서비스
결제 방식 로컬 결제 지원 (해외 신용카드 불필요) 해외 신용카드 필수 다양하지만 복잡한 과정 필요
지원 모델 GPT-4.1, Claude, Gemini, DeepSeek 통합 OpenAI 모델만 제한된 모델 선택
GPT-4.1 가격 $8/MTok $8/MTok $10~$15/MTok
Claude Sonnet 4.5 $15/MTok $15/MTok $18~$20/MTok
Gemini 2.5 Flash $2.50/MTok $2.50/MTok $3~$5/MTok
DeepSeek V3.2 $0.42/MTok 미지원 $0.50~$1/MTok
API 키 관리 단일 키로 모든 모델 접근 서비스별 개별 키 필요 서비스별 키 관리 복잡
무료 크레딧 가입 시 제공 $5 초기 크레딧 제한적 제공

왜 로그 기록과 비용 추적이 중요한가?

AI API를 활용하는 시스템에서 로그 기록과 비용 추적은 다음과 같은 이유로 필수적입니다:

시스템 아키텍처 개요

이번 튜토리얼에서 구축할 시스템의 아키텍처는 다음과 같습니다:

┌─────────────────────────────────────────────────────────────────┐
│                      사용자 애플리케이션                           │
└─────────────────────────────┬───────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    API Gateway (HolySheep AI)                    │
│              base_url: https://api.holysheep.ai/v1              │
└─────────────────────────────┬───────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
        ▼                     ▼                     ▼
┌───────────────┐    ┌───────────────┐    ┌───────────────┐
│  로그 저장소   │    │  비용 추적 DB │    │  모니터링 대시보드 │
│  (SQLite/PG)  │    │  (시계열 DB)  │    │   (실시간)     │
└───────────────┘    └───────────────┘    └───────────────┘

핵심 구현 코드

1. 기본 설정 및 의존성

# requirements.txt
openai==1.12.0
anthropic==0.18.0
python-dotenv==1.0.0
sqlalchemy==2.0.25
psycopg2-binary==2.9.9
httpx==0.26.0
loguru==0.7.2
pydantic==2.6.1
dataclasses-json==0.6.4
python-dateutil==2.8.2


설치 명령어

pip install -r requirements.txt

2. 로그 기록 및 비용 추적 시스템 구현

# logger.py
import os
import json
import sqlite3
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict, Any
from contextlib import contextmanager
from loguru import logger

import httpx
from openai import OpenAI


HolySheep AI 설정

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")


모델별 가격 정보 ($/MTok)

MODEL_PRICING = {
    "gpt-4.1": {"prompt": 8.00, "completion": 32.00},
    "gpt-4.1-mini": {"prompt": 0.80, "completion": 3.20},
    "gpt-4.1-turbo": {"prompt": 8.00, "completion": 32.00},
    "claude-sonnet-4-5": {"prompt": 15.00, "completion": 75.00},
    "claude-opus-4-5": {"prompt": 75.00, "completion": 375.00},
    "claude-3-5-haiku": {"prompt": 0.80, "completion": 4.00},
    "gemini-2.5-flash": {"prompt": 2.50, "completion": 10.00},
    "gemini-2.5-flash-lite": {"prompt": 0.70, "completion": 2.80},
    "deepseek-v3.2": {"prompt": 0.42, "completion": 2.70},
    "deepseek-chat": {"prompt": 0.27, "completion": 1.10},
}

@dataclass
class APICallLog:
    id: Optional[int] = None
    timestamp: str = ""
    model: str = ""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    prompt_cost: float = 0.0
    completion_cost: float = 0.0
    total_cost: float = 0.0
    latency_ms: int = 0
    status: str = "success"
    error_message: Optional[str] = None
    request_id: Optional[str] = None
    user_id: Optional[str] = None
    metadata: Optional[str] = None

class AICallLogger:
    def __init__(self, db_path: str = "ai_api_logs.db"):
        self.db_path = db_path
        self._init_database()
        logger.add(
            "logs/ai_api_{time}.log",
            rotation="1 day",
            retention="30 days",
            level="INFO",
            format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}"
        )
    
    def _init_database(self):
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS api_call_logs (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    model TEXT NOT NULL,
                    prompt_tokens INTEGER DEFAULT 0,
                    completion_tokens INTEGER DEFAULT 0,
                    total_tokens INTEGER DEFAULT 0,
                    prompt_cost REAL DEFAULT 0.0,
                    completion_cost REAL DEFAULT 0.0,
                    total_cost REAL DEFAULT 0.0,
                    latency_ms INTEGER DEFAULT 0,
                    status TEXT DEFAULT 'success',
                    error_message TEXT,
                    request_id TEXT,
                    user_id TEXT,
                    metadata TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')
            
            cursor.execute('''
                CREATE INDEX IF NOT EXISTS idx_timestamp ON api_call_logs(timestamp)
            ''')
            cursor.execute('''
                CREATE INDEX IF NOT EXISTS idx_model ON api_call_logs(model)
            ''')
            cursor.execute('''
                CREATE INDEX IF NOT EXISTS idx_user_id ON api_call_logs(user_id)
            ''')
            conn.commit()
    
    @contextmanager
    def _get_connection(self):
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
        finally:
            conn.close()
    
    def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> Dict[str, float]:
        pricing = MODEL_PRICING.get(model, {"prompt": 0.0, "completion": 0.0})
        prompt_cost = (prompt_tokens / 1_000_000) * pricing["prompt"]
        completion_cost = (completion_tokens / 1_000_000) * pricing["completion"]
        return {
            "prompt_cost": round(prompt_cost, 6),
            "completion_cost": round(completion_cost, 6),
            "total_cost": round(prompt_cost + completion_cost, 6)
        }
    
    def log_call(self, log_entry: APICallLog) -> int:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO api_call_logs (
                    timestamp, model, prompt_tokens, completion_tokens, total_tokens,
                    prompt_cost, completion_cost, total_cost, latency_ms, status,
                    error_message, request_id, user_id, metadata
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                log_entry.timestamp,
                log_entry.model,
                log_entry.prompt_tokens,
                log_entry.completion_tokens,
                log_entry.total_tokens,
                log_entry.prompt_cost,
                log_entry.completion_cost,
                log_entry.total_cost,
                log_entry.latency_ms,
                log_entry.status,
                log_entry.error_message,
                log_entry.request_id,
                log_entry.user_id,
                log_entry.metadata
            ))
            conn.commit()
            logger.info(
                f"Logged API call: model={log_entry.model}, "
                f"tokens={log_entry.total_tokens}, cost=${log_entry.total_cost:.6f}, "
                f"latency={log_entry.latency_ms}ms"
            )
            return cursor.lastrowid
    
    def get_cost_summary(
        self,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        model: Optional[str] = None
    ) -> Dict[str, Any]:
        query = "SELECT model, COUNT(*), SUM(prompt_tokens), SUM(completion_tokens), SUM(total_cost) FROM api_call_logs WHERE 1=1"
        params = []
        
        if start_date:
            query += " AND timestamp >= ?"
            params.append(start_date)
        if end_date:
            query += " AND timestamp <= ?"
            params.append(end_date)
        if model:
            query += " AND model = ?"
            params.append(model)
        
        query += " GROUP BY model"
        
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            rows = cursor.fetchall()
        
        return {
            "summary": [
                {
                    "model": row[0],
                    "call_count": row[1],
                    "total_prompt_tokens": row[2],
                    "total_completion_tokens": row[3],
                    "total_cost_usd": row[4]
                }
                for row in rows
            ]
        }
    
    def get_recent_logs(self, limit: int = 100) -> List[APICallLog]:
        with self._get_connection() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM api_call_logs ORDER BY timestamp DESC LIMIT ?",
                (limit,)
            )
            rows = cursor.fetchall()
        
        return [APICallLog(**dict(row)) for row in rows]


class HolySheepAIClient:
    def __init__(self, api_key: str = API_KEY, logger: Optional[AICallLogger] = None):
        self.api_key = api_key
        self.logger = logger or AICallLogger()
        self.base_url = BASE_URL
        
        if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
            raise ValueError(
                "유효한 HolySheep API 키를 설정해주세요. "
                "https://www.holysheep.ai/register 에서 가입하세요."
            )
        
        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url,
            timeout=httpx.Timeout(60.0, connect=10.0)
        )
        logger.info(f"HolySheep AI 클라이언트 초기화 완료: {self.base_url}")
    
    def chat_completion(
        self,
        model: str,
        messages: List[Dict[str, str]],
        user_id: Optional[str] = None,
        metadata: Optional[Dict] = None,
        **kwargs
    ) -> Dict[str, Any]:
        start_time = datetime.now()
        
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs
            )
            
            end_time = datetime.now()
            latency_ms = int((end_time - start_time).total_seconds() * 1000)
            
            usage = response.usage
            cost_info = self.logger.calculate_cost(
                model,
                usage.prompt_tokens,
                usage.completion_tokens
            )
            
            log_entry = APICallLog(
                timestamp=start_time.isoformat(),
                model=model,
                prompt_tokens=usage.prompt_tokens,
                completion_tokens=usage.completion_tokens,
                total_tokens=usage.total_tokens,
                prompt_cost=cost_info["prompt_cost"],
                completion_cost=cost_info["completion_cost"],
                total_cost=cost_info["total_cost"],
                latency_ms=latency_ms,
                status="success",
                request_id=response.id,
                user_id=user_id,
                metadata=json.dumps(metadata) if metadata else None
            )
            self.logger.log_call(log_entry)
            
            return {
                "success": True,
                "response": response.model_dump(),
                "usage": {
                    "prompt_tokens": usage.prompt_tokens,
                    "completion_tokens": usage.completion_tokens,
                    "total_tokens": usage.total_tokens
                },
                "cost": cost_info,
                "latency_ms": latency_ms
            }
            
        except Exception as e:
            end_time = datetime.now()
            latency_ms = int((end_time - start_time).total_seconds() * 1000)
            
            log_entry = APICallLog(
                timestamp=start_time.isoformat(),
                model=model,
                latency_ms=latency_ms,
                status="error",
                error_message=str(e),
                user_id=user_id,
                metadata=json.dumps(metadata) if metadata else None
            )
            self.logger.log_call(log_entry)
            
            logger.error(f"API 호출 오류: {str(e)}")
            return {
                "success": False,
                "error": str(e),
                "latency_ms": latency_ms
            }
    
    def batch_chat_completion(
        self,
        requests: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        results = []
        for req in requests:
            result = self.chat_completion(
                model=req.get("model", "gpt-4.1-mini"),
                messages=req.get("messages", []),
                user_id=req.get("user_id"),
                metadata=req.get("metadata")
            )
            results.append(result)
        return results



사용 예시

if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    
    load_dotenv()
    
    logger = AICallLogger("ai_logs.db")
    client = HolySheepAIClient(logger=logger)
    
    # 단일 호출 예시
    result = client.chat_completion(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
            {"role": "user", "content": "안녕하세요! AI API 사용법을 알려주세요."}
        ],
        user_id="user_123",
        metadata={"project": "tutorial", "env": "development"}
    )
    
    if result["success"]:
        print(f"✅ 응답 성공")
        print(f"   토큰: {result['usage']['total_tokens']}")
        print(f"   비용: ${result['cost']['total_cost']:.6f}")
        print(f"   지연: {result['latency_ms']}ms")
        print(f"   응답: {result['response']['choices'][0]['message']['content'][:100]}...")
    else:
        print(f"❌ 오류 발생: {result['error']}")
    
    # 비용 요약 조회
    summary = logger.get_cost_summary()
    print("\n📊 비용 요약:")
    for item in summary["summary"]:
        print(f"   {item['model']}: ${item['total_cost_usd']:.6f} ({item['call_count']}회 호출)")

3. 웹 대시보드 및 실시간 모니터링

# dashboard.py
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
from logger import AICallLogger

st.set_page_config(
    page_title="AI API 모니터링 대시보드",
    page_icon="📊",
    layout="wide"
)

st.title("📊 AI API 호출 로그 및 비용 추적 대시보드")


사이드바 설정

st.sidebar.header("설정")
db_path = st.sidebar.text_input("데이터베이스 경로", value="ai_logs.db")
refresh_interval = st.sidebar.slider("자동 새로고침 간격 (초)", 5, 60, 10)


Logger 초기화

logger = AICallLogger(db_path)


메트릭 표시

col1, col2, col3, col4 = st.columns(4)


시간 범위 선택

time_range = st.selectbox(
    "조회 기간",
    ["오늘", "최근 7일", "최근 30일", "전체"]
)

now = datetime.now()
if time_range == "오늘":
    start_date = now.replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
elif time_range == "최근 7일":
    start_date = (now - timedelta(days=7)).isoformat()
elif time_range == "최근 30일":
    start_date = (now - timedelta(days=30)).isoformat()
else:
    start_date = None

summary = logger.get_cost_summary(start_date=start_date)


총 비용 계산

total_cost = sum(item["total_cost_usd"] for item in summary["summary"]) if summary["summary"] else 0
total_calls = sum(item["call_count"] for item in summary["summary"]) if summary["summary"] else 0
total_tokens = sum(
    item["total_prompt_tokens"] + item["total_completion_tokens"] 
    for item in summary["summary"]
) if summary["summary"] else 0

with col1:
    st.metric("총 비용", f"${total_cost:.4f}", delta_color="inverse")
with col2:
    st.metric("총 호출 횟수", f"{total_calls:,}")
with col3:
    st.metric("총 토큰 사용", f"{total_tokens:,}")
with col4:
    avg_cost = total_cost / total_calls if total_calls > 0 else 0
    st.metric("평균 호출 비용", f"${avg_cost:.6f}")


모델별 비용 차트

st.subheader("📈 모델별 비용 분포")

if summary["summary"]:
    df = pd.DataFrame(summary["summary"])
    
    col_chart1, col_chart2 = st.columns(2)
    
    with col_chart1:
        fig_pie = px.pie(
            df,
            values="total_cost_usd",
            names="model",
            title="모델별 비용 비율",
            hole=0.4
        )
        st.plotly_chart(fig_pie, use_container_width=True)
    
    with col_chart2:
        fig_bar = px.bar(
            df,
            x="model",
            y="total_cost_usd",
            color="call_count",
            title="모델별 총 비용",
            labels={"total_cost_usd": "비용 (USD)", "model": "모델"}
        )
        st.plotly_chart(fig_bar, use_container_width=True)
else:
    st.info("표시할 데이터가 없습니다.")


최근 로그 테이블

st.subheader("📋 최근 API 호출 로그")

logs = logger.get_recent_logs(limit=50)
if logs:
    log_data = [
        {
            "시간": log.timestamp,
            "모델": log.model,
            "Prompt 토큰": log.prompt_tokens,
            "Completion 토큰": log.completion_tokens,
            "총 토큰": log.total_tokens,
            "비용 ($)": log.total_cost,
            "지연 (ms)": log.latency_ms,
            "상태": log.status,
            "사용자": log.user_id or "-"
        }
        for log in logs
    ]
    
    df_logs = pd.DataFrame(log_data)
    st.dataframe(
        df_logs,
        use_container_width=True,
        hide_index=True
    )
else:
    st.info("로그 데이터가 없습니다.")


자동 새로고침

if refresh_interval > 0:
    import time
    time.sleep(refresh_interval)
    st.rerun()


실행 명령어


streamlit run dashboard.py

실제 측정 데이터 및 최적화 팁

제가 실제로 HolySheep AI를 사용하면서 측정한 성능 데이터입니다:

모델 평균 지연 시간 비용 효율성 권장 사용 사례
DeepSeek V3.2 1,200~2,500ms ⭐⭐⭐⭐⭐ 대량 텍스트 처리, 번역
Gemini 2.5 Flash 800~1,500ms ⭐⭐⭐⭐⭐ 빠른 응답 필요, 실시간 챗봇
GPT-4.1-mini 600~1,200ms ⭐⭐⭐⭐ 일반적인 대화, 요약
Claude Sonnet 4.5 1,000~2,000ms ⭐⭐⭐⭐ 고품질 콘텐츠 생성, 분석

비용 최적화 전략

# cost_optimizer.py - 고급 비용 최적화 모듈

from typing import List, Dict, Optional, Callable
from dataclasses import dataclass
import time

@dataclass
class CostBudget:
    daily_limit: float
    monthly_limit: float
    alert_threshold: float = 0.8

class CostBudgetManager:
    def __init__(self, logger: AICallLogger, budget: CostBudget):
        self.logger = logger
        self.budget = budget
        self.spent_today = 0.0
        self.spent_month = 0.0
        self.alerts: List[str] = []
    
    def check_budget(self) -> bool:
        """예산 범위 내인지 확인"""
        summary = self.logger.get_cost_summary()
        total_spent = sum(item["total_cost_usd"] for item in summary["summary"])
        
        daily_threshold = self.budget.daily_limit * self.budget.alert_threshold
        monthly_threshold = self.budget.monthly_limit * self.budget.alert_threshold
        
        if total_spent >= daily_threshold:
            self.alerts.append(f"⚠️ 일일 예산의 {self.budget.alert_threshold*100}% 이상 사용: ${total_spent:.4f}")
        
        if total_spent >= monthly_threshold:
            self.alerts.append(f"🚨 월간 예산의 {self.budget.alert_threshold*100}% 이상 사용: ${total_spent:.4f}")
        
        return total_spent < self.budget.daily_limit
    
    def select_cost_efficient_model(
        self,
        task_complexity: str,
        preferred_model: Optional[str] = None
    ) -> str:
        """작업 복잡도에 따른 비용 효율적 모델 선택"""
        
        model_mapping = {
            "simple": "deepseek-chat",
            "moderate": "gemini-2.5-flash",
            "complex": "claude-sonnet-4-5",
            "high_quality": "gpt-4.1"
        }
        
        if preferred_model:
            return preferred_model
        
        return model_mapping.get(task_complexity, "gemini-2.5-flash")


class RequestBatcher:
    """요청 일괄 처리로 비용 최적화"""
    
    def __init__(self, client: HolySheepAIClient, batch_size: int = 10):
        self.client = client
        self.batch_size = batch_size
        self.pending_requests: List[Dict] = []
    
    def add_request(
        self,
        model: str,
        messages: List[Dict],
        user_id: Optional[str] = None
    ):
        self.pending_requests.append({
            "model": model,
            "messages": messages,
            "user_id": user_id
        })
        
        if len(self.pending_requests) >= self.batch_size:
            return self.flush()
        return None
    
    def flush(self) -> List[Dict]:
        """대기 중인 요청 일괄 처리"""
        if not self.pending_requests:
            return []
        
        results = self.client.batch_chat_completion(self.pending_requests)
        self.pending_requests = []
        return results


def estimate_cost(model: str, avg_prompt_tokens: int, avg_completion_tokens: int) -> float:
    """비용 예상 함수"""
    pricing = {
        "deepseek-chat": {"prompt": 0.27, "completion": 1.10},
        "gemini-2.5-flash": {"prompt": 2.50, "completion": 10.00},
        "gpt-4.1-mini": {"prompt": 0.80, "completion": 3.20},
        "claude-sonnet-4-5": {"prompt": 15.00, "completion": 75.00}
    }
    
    if model not in pricing:
        return 0.0
    
    p = pricing[model]
    cost = (avg_prompt_tokens / 1_000_000) * p["prompt"]
    cost += (avg_completion_tokens / 1_000_000) * p["completion"]
    return cost



비용 예상 예시

if __name__ == "__main__":
    test_cases = [
        ("deepseek-chat", 500, 200),
        ("gemini-2.5-flash", 1000, 500),
        ("gpt-4.1-mini", 200, 100),
    ]
    
    print("📊 예상 비용 비교 (10,000회 호출 기준)")
    print("-" * 60)
    
    for model, prompt, completion in test_cases:
        single_cost = estimate_cost(model, prompt, completion)
        total_cost = single_cost * 10000
        print(f"{model:25} | 1회: ${single_cost:.6f} | 10K: ${total_cost:.2f}")

자주 발생하는 오류와 해결책

1. API 키 인증 오류

# ❌ 오류 메시지

AuthenticationError: Incorrect API key provided



✅ 해결 방법

import os


방법 1: 환경 변수로 설정

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_ACTUAL_API_KEY"


방법 2: 직접 전달

client = HolySheepAIClient(api_key="YOUR_ACTUAL_API_KEY")


⚠️ 중요: API 키 발급은 https://www.holysheep.ai/register 에서

2._RATE_LIMIT 오류 ( Rate LimitExceeded)

# ❌ 오류 메시지

RateLimitError: Rate limit exceeded for model gpt-4.1



✅ 해결 방법

import time
from tenacity import retry, wait_exponential, stop_after_attempt

class RateLimitHandler:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def execute_with_retry(self, func: Callable, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except RateLimitError as e:
                if attempt == self.max_retries - 1:
                    raise
                wait_time = self.base_delay * (2 ** attempt)
                print(f"Rate limit 초과. {wait_time}초 후 재시도... ({attempt + 1}/{self.max_retries})")
                time.sleep(wait_time)


사용 예시

handler = RateLimitHandler(max_retries=3)
result = handler.execute_with_retry(
    client.chat_completion,
    model="gpt-4.1",
    messages=[{"role": "user", "content": "안녕하세요"}]
)

3. 타임아웃 및 연결 오류

# ❌ 오류 메시지

httpx.ConnectTimeout: Connection timeout



✅ 해결 방법

from httpx import Timeout, HTTPTransport, ASGITransport


방법 1: 타임아웃 설정 증가

client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    timeout=Timeout(120.0, connect=30.0)  # 읽기 120초, 연결 30초
)


방법 2: 재시도 로직과 함께 사용

from openai import APIConnectionError, APITimeoutError

def robust_request(request_func, *args, **kwargs):
    max_attempts = 3
    for attempt in range(max_attempts):
        try:
            return request_func(*args, **kwargs)
        except (APIConnectionError, APITimeoutError) as e:
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt)
            logger.warning(f"연결 재시도: {attempt + 1}/{max_attempts}")


방법 3: 연결 풀링 설정

transport = HTTPTransport(
    pool_limits=100,  # 최대 연결 수
    retries=3
)
client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    http_client=httpx.Client(transport=transport)
)

4. 토큰 계산 오류 및 부정확한 비용

# ❌ 문제: 응답의 usage 정보가 None인 경우


✅ 해결 방법

def safe_extract_usage(response, fallback_prompt_tokens: int = 0, fallback_completion_tokens: int = 0):
    """안전하게 usage 정보 추출"""
    if response.usage is None:
        logger.warning("usage 정보 없음. 추정치 사용.")
        # Claude API의 경우 토큰 수가 없을 수 있음
        return {
            "prompt_tokens": fallback_prompt_tokens,
            "completion_tokens": fallback_completion_tokens,
            "total_tokens": fallback_prompt_tokens + fallback_completion_tokens,
            "estimated": True
        }
    
    return {
        "prompt_tokens": response.usage.prompt_tokens,
        "completion_tokens": response.usage.completion_tokens,
        "total_tokens": response.usage.total_tokens,
        "estimated": False
    }


모델별 토큰 추정 함수 (Claude 등에서 사용)

def estimate_tokens_from_text(text: str, encoding_name: str = "cl100k_base") -> int:
    """대략적인 토큰 수 추정 (글자 수 기반)"""
    # 영어: ~4자 = 1토큰
    # 한국어: ~2자 = 1토큰 (대략적)
    char_count = len(text)
    
    # 한국어 비율 추정
    korean_ratio = sum(1 for c in text if '\uAC00' <= c <= '\uD7AF') / max(char_count, 1)
    english_estimate = char_count * (1 - korean_ratio) / 4
    korean_estimate = char_count * korean_ratio / 2
    
    return int(english_estimate + korean_estimate)

5. 데이터베이스 잠금 및 동시성 문제

# ❌ 오류 메시지

sqlite3.OperationalError: database is locked



✅ 해결 방법

import sqlite3
import threading
from contextlib import contextmanager

class ThreadSafeLogger(AICallLogger):
    def __init__(self, db_path: str = "ai_logs.db"):
        self._lock = threading.Lock()
        super().__init__(db_path)
    
    @contextmanager
    def _get_connection(self):
        with self._lock:
            conn = sqlite3.connect(
                self.db_path,
                timeout=30.0,  # 잠금 대기 시간
                check_same_thread=False  # 스레드 간 공유 허용
            )
            conn.execute("PRAGMA journal_mode=WAL")  # WAL 모드로 전환
            try:
                yield conn
            finally:
                conn.close()
    
    def batch_log(self, log_entries: List[APICallLog]):
        """배치 삽입으로 성능 향상"""
        with self._lock:
            conn = sqlite3.connect(self.db_path, timeout=30.0)
            try:
                cursor = conn.cursor()
                data = [
                    (
                        entry.timestamp, entry.model, entry.prompt_tokens,
                        entry.completion_tokens, entry.total_tokens,
                        entry.prompt_cost, entry.completion_cost, entry.total_cost,
                        entry.latency_ms, entry.status, entry.error_message,
                        entry.request_id, entry.user_id, entry

관련 리소스
관련 문서