LLMOps 완전 가이드: AI 모델을 프로덕션에서 운영하는 법 (2025)

flowchart LR
    A[요구사항 정의] --> B[모델 선택]
    B --> C[프롬프트 개발]
    C --> D[평가 & 테스트]
    D --> E{품질 기준 통과?}
    E -- No --> C
    E -- Yes --> F[배포]
    F --> G[모니터링]
    G --> H{이슈 감지?}
    H -- Yes --> C
    H -- No --> G

flowchart TD
    Start[태스크 분석] --> Q1{실시간 응답 필요?}
    Q1 -- Yes --> Q2{복잡도?}
    Q1 -- No --> Q3{문서 길이?}
    Q2 -- 단순 --> Haiku[Claude Haiku / GPT-4o mini]
    Q2 -- 복잡 --> Sonnet[Claude Sonnet / GPT-4o]
    Q3 -- 32K 이하 --> Sonnet
    Q3 -- 32K 이상 --> Long[Claude / Gemini 1.5 Pro]
    Haiku --> Cost[$0.80~2.50 /1M]
    Sonnet --> Cost2[$3~10 /1M]
    Long --> Cost3[$3.50~15 /1M]

prompts/
├── customer_support/
│   ├── v1.0.0.txt      # 최초 버전
│   ├── v1.1.0.txt      # 한국어 개선
│   └── v2.0.0.txt      # 멀티턴 지원 추가
├── code_review/
│   └── v1.0.0.txt
└── metadata.json        # 버전별 평가 점수 기록

# 프롬프트 로더 예시
class PromptRegistry:
    def __init__(self, base_path: str = "prompts/"):
        self.base_path = Path(base_path)
        self.metadata = json.loads((self.base_path / "metadata.json").read_text())

    def get(self, name: str, version: str = "latest") -> str:
        if version == "latest":
            versions = sorted(self.metadata[name].keys())
            version = versions[-1]
        path = self.base_path / name / f"v{version}.txt"
        return path.read_text()

    def get_best(self, name: str) -> str:
        '''평가 점수 기준 최고 버전 반환'''
        versions = self.metadata[name]
        best = max(versions.items(), key=lambda x: x[1].get("score", 0))
        return self.get(name, best[0])

flowchart LR
    A[테스트셋 준비] --> B[LLM 실행]
    B --> C[자동 평가]
    C --> D[LLM-as-Judge]
    D --> E[사람 검토]
    E --> F[점수 기록]
    F --> G{기준치 이상?}
    G -- Yes --> H[프로덕션 배포]
    G -- No --> I[프롬프트 수정]
    I --> B

JUDGE_PROMPT = '''다음 AI 응답을 평가하세요.

[질문]: {question}
[AI 응답]: {response}
[정답 예시]: {reference}

평가 기준 (각 1-5점):
1. 정확성: 사실적으로 맞는가?
2. 완전성: 질문에 충분히 답했는가?
3. 간결성: 불필요한 내용이 없는가?

JSON으로 응답: {{"accuracy": 점수, "completeness": 점수, "conciseness": 점수, "reasoning": "이유"}}'''

def evaluate_with_llm(question, response, reference):
    result = gpt4o.complete(
        JUDGE_PROMPT.format(question=question, response=response, reference=reference)
    )
    return json.loads(result)

flowchart TB
    User[사용자] --> LB[Load Balancer]
    LB --> API[FastAPI / Next.js API]
    API --> Cache{Redis 캐시<br/>히트?}
    Cache -- Yes --> Return[캐시 응답 반환]
    Cache -- No --> Queue[작업 큐<br/>Celery/BullMQ]
    Queue --> Worker1[Worker 1<br/>GPT-4o]
    Queue --> Worker2[Worker 2<br/>GPT-4o]
    Worker1 --> Monitor[LangSmith / Langfuse]
    Worker2 --> Monitor
    Monitor --> DB[(PostgreSQL<br/>로그 저장)]

import hashlib
import redis

r = redis.Redis()
CACHE_TTL = 3600  # 1시간

def cached_llm_call(prompt: str, model: str) -> str:
    # 캐시 키: 프롬프트 + 모델의 해시
    key = f"llm:{hashlib.md5(f'{model}:{prompt}'.encode()).hexdigest()}"

    cached = r.get(key)
    if cached:
        return cached.decode()

    response = call_llm(prompt, model)
    r.setex(key, CACHE_TTL, response)
    return response

flowchart LR
    App[앱 서버] --> |로그 스트림| Langfuse[Langfuse<br/>오픈소스 LLM 관찰]
    Langfuse --> Dashboard[대시보드]
    Dashboard --> Metrics[지표:<br/>레이턴시/비용/에러율]
    Dashboard --> Traces[트레이스:<br/>프롬프트/응답 기록]
    Dashboard --> Evals[평가:<br/>LLM-as-Judge 점수]
    Metrics --> Alert[임계치 초과 시<br/>알림]

# Langfuse 연동 예시
from langfuse import Langfuse

langfuse = Langfuse()

def track_llm_call(user_id: str, prompt: str, response: str, tokens: int, cost: float):
    langfuse.trace(
        name="llm-call",
        user_id=user_id,
        input=prompt,
        output=response,
        metadata={
            "tokens": tokens,
            "cost_usd": cost,
            "model": "gpt-4o",
        }
    )