LiteLLM 완전 가이드 4편 — LangChain·LangGraph 통합, 가드레일, Prometheus 모니터링, 프로덕션 운영

3편에서 Proxy 서버를 띄웠습니다. 4편은 그 위에 쌓는 것들입니다. LangChain·LangGraph가 Proxy를 모르게 쓰고, 콘텐츠 필터로 민감한 출력을 막고, Prometheus로 모든 메트릭을 시각화합니다. 마지막으로 LiteLLM의 실제 한계와 대안도 솔직하게 정리합니다.

[4편 핵심 요약]
→ LangChain + Proxy: base_url=http://proxy:4000 — 모델 교체 없이 모든 LLM 사용
→ LangGraph: LiteLLM이 LangGraph 에이전트를 모델처럼 호출 가능 (A2A 프로토콜)
→ 가드레일: 콘텐츠 필터 / PII 마스킹 (Presidio) / 커스텀 훅
→ Prometheus: /metrics 엔드포인트 → Grafana 대시보드 자동 연동
→ 보안 주의: v1.82.7/1.82.8 — TeamPCP 공급망 공격 (2026년 3월), 즉시 업그레이드
→ 프로덕션 한계: Python GIL로 1000+ RPS 한계, 고트래픽엔 Redis 필수
→ 대안 고려: Bifrost (Go 기반, 9.4x 처리량), OpenRouter (관리형 SaaS)
→ LiteLLM이 맞는 경우: 셀프호스팅 필수, 데이터 주권, 중간 규모 팀

 

---

실전 1 — LangChain + LiteLLM Proxy 통합

# ── 방법 1: ChatOpenAI로 Proxy 연결 ──────────────────
# 가장 간단 — OpenAI 호환 인터페이스 그대로 사용
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate

# Proxy를 OpenAI처럼 사용
llm = ChatOpenAI(
    model="claude-sonnet",          # config.yaml의 model_name
    openai_api_key="sk-team-xxxx",  # 가상 키
    openai_api_base="http://localhost:4000",
    temperature=0.7,
)

# 일반 호출
response = llm.invoke([
    SystemMessage(content="당신은 친절한 AI입니다."),
    HumanMessage(content="파이썬 데코레이터를 설명해줘")
])
print(response.content)

# 프롬프트 템플릿 + 체인
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 {language} 전문가입니다."),
    ("human", "{question}")
])
chain = prompt | llm
result = chain.invoke({"language": "Python", "question": "asyncio란?"})
print(result.content)

# ── 방법 2: 여러 모델 동적 전환 ──────────────────────
# Proxy 덕에 코드 변경 없이 모델만 바꿔서 A/B 테스트

models = {
    "claude": ChatOpenAI(
        model="claude-sonnet",
        openai_api_key="sk-team-xxxx",
        openai_api_base="http://localhost:4000",
    ),
    "gpt": ChatOpenAI(
        model="gpt",
        openai_api_key="sk-team-xxxx",
        openai_api_base="http://localhost:4000",
    ),
    "cheap": ChatOpenAI(
        model="gemini-flash",   # 저렴한 모델
        openai_api_key="sk-team-xxxx",
        openai_api_base="http://localhost:4000",
    ),
}

# 태스크별 모델 선택
def get_llm(task_type: str):
    if task_type in ("code_review", "architecture"):
        return models["claude"]
    elif task_type in ("classification", "simple_qa"):
        return models["cheap"]
    else:
        return models["gpt"]

# 사용
llm = get_llm("code_review")
result = llm.invoke([HumanMessage(content="이 코드 리뷰해줘: def f(x): return x*2")])

# ── 방법 3: 구조화 출력 (with_structured_output) ─────
from pydantic import BaseModel
from typing import Literal

class CodeReview(BaseModel):
    score: int
    issues: list[str]
    recommendation: Literal["approve", "request_changes", "needs_work"]

llm = ChatOpenAI(
    model="claude-sonnet",
    openai_api_key="sk-team-xxxx",
    openai_api_base="http://localhost:4000",
)

structured_llm = llm.with_structured_output(CodeReview)
review = structured_llm.invoke("이 코드를 리뷰해줘:\n\ndef add(a, b):\n    return a+b")
print(f"점수: {review.score}, 권고: {review.recommendation}")

---

실전 2 — LangGraph + LiteLLM 통합

# ── 방법 1: LangGraph 에이전트에서 Proxy 모델 사용 ────
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
import operator

# Proxy를 통한 모델 초기화
model = ChatOpenAI(
    model="claude-sonnet",
    openai_api_key="sk-engineering-xxxx",
    openai_api_base="http://localhost:4000",
)

@tool
def calculator(expression: str) -> str:
    """수식 계산"""
    return str(eval(expression))

@tool
def get_time() -> str:
    """현재 시간 반환"""
    from datetime import datetime
    return datetime.now().isoformat()

tools = [calculator, get_time]
model_with_tools = model.bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

def call_model(state: AgentState) -> dict:
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state: AgentState) -> dict:
    from langchain_core.messages import ToolMessage
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        tool_fn = next(t for t in tools if t.name == tc["name"])
        result = tool_fn.invoke(tc["args"])
        results.append(ToolMessage(
            content=str(result),
            tool_call_id=tc["id"]
        ))
    return {"messages": results}

def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
app = workflow.compile()

result = app.invoke({"messages": [HumanMessage(content="지금 몇 시야? 그리고 123 * 456은?")]})
print(result["messages"][-1].content)

# ── 방법 2: LiteLLM에서 LangGraph 에이전트를 모델로 호출
# LiteLLM이 LangGraph 서버를 A2A 프로토콜로 호출

import litellm

# LangGraph 서버가 http://localhost:2024 에서 실행 중이어야 함
response = litellm.completion(
    model="langgraph/agent",   # provider/agent-name 형식
    messages=[{"role": "user", "content": "서울 날씨 알려줘"}],
    api_base="http://localhost:2024",
)
print(response.choices[0].message.content)

# Proxy config.yaml에 LangGraph 에이전트 등록
# model_list:
#   - model_name: my-agent
#     litellm_params:
#       model: langgraph/agent
#       api_base: http://langgraph-server:2024

---

실전 3 — 가드레일 설정

# config.yaml에 가드레일 추가

# ── Presidio PII 마스킹 (오픈소스, 셀프호스팅 가능) ───
guardrails:
  - prompt_injection_detection: true
  - pii_masking:
      provider: "presidio"
      # 마스킹할 PII 유형
      entities:
        - "PERSON"
        - "EMAIL_ADDRESS"
        - "PHONE_NUMBER"
        - "CREDIT_CARD"
        - "IBAN_CODE"
        - "US_SSN"
      # 마스킹 방식: replace(기본) / hash / mask / redact
      anonymize_action: "replace"

# ── 커스텀 가드레일 — Python 훅 ───────────────────────
# custom_guardrail.py

from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks

class MyContentGuardrail(CustomGuardrail):
    """커스텀 콘텐츠 필터"""

    def __init__(self):
        super().__init__(
            guardrail_name="my-content-filter",
            supported_event_hooks=[
                GuardrailEventHooks.pre_call,    # 요청 전
                GuardrailEventHooks.post_call,   # 응답 후
            ]
        )

    async def async_pre_call_hook(self, data, cache, call_type, request_data):
        """요청 전 검사 — 금지 키워드 차단"""
        messages = data.get("messages", [])
        forbidden = ["비밀번호", "개인정보", "주민번호"]

        for msg in messages:
            content = msg.get("content", "")
            for word in forbidden:
                if word in content:
                    raise ValueError(
                        f"금지된 내용 감지: '{word}' 포함 요청 차단"
                    )
        return data

    async def async_post_call_success_hook(
        self, data, cache, response, request_data
    ):
        """응답 후 검사 — 민감 정보 포함 여부"""
        output = response.choices[0].message.content or ""

        # 응답에서 카드 번호 패턴 감지
        import re
        card_pattern = r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'
        if re.search(card_pattern, output):
            # 응답에서 카드 번호 마스킹
            response.choices[0].message.content = re.sub(
                card_pattern, "****-****-****-****", output
            )
        return response

# config.yaml에 커스텀 가드레일 등록
guardrails:
  - guardrail_name: my-content-filter
    litellm_params:
      guardrail: custom
      guardrail_path: ./custom_guardrail.py
      guardrail_class: MyContentGuardrail
      mode: "pre_and_post_call"

# 특정 모델에만 가드레일 적용
model_list:
  - model_name: claude-sonnet
    litellm_params:
      model: anthropic/claude-sonnet-4-6
      api_key: os.environ/ANTHROPIC_API_KEY
    model_info:
      guardrails: ["my-content-filter"]  # 이 모델에만 적용

# 클라이언트에서 가드레일 동적 활성화/비활성화
from openai import OpenAI

client = OpenAI(
    api_key="sk-team-xxxx",
    base_url="http://localhost:4000"
)

# 특정 요청에 가드레일 적용
response = client.chat.completions.create(
    model="claude-sonnet",
    messages=[{"role": "user", "content": "이 문서에서 정보 추출해줘: ..."}],
    extra_body={
        "guardrails": {
            "enabled": True,
            "guardrail_name": "my-content-filter"
        }
    }
)

# 가드레일 비활성화 (신뢰할 수 있는 내부 요청 등)
response = client.chat.completions.create(
    model="claude-sonnet",
    messages=[{"role": "user", "content": "내부 디버그 요청..."}],
    extra_body={
        "guardrails": {"enabled": False}
    }
)

---

실전 4 — Prometheus + Grafana 모니터링

# config.yaml — Prometheus 메트릭 활성화
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL
  # Prometheus 활성화
  enable_prometheus: true
  prometheus_port: 4000  # /metrics 엔드포인트가 같은 포트에 노출됨

# prometheus.yml — Prometheus 스크레이핑 설정
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "litellm"
    static_configs:
      - targets: ["litellm:4000"]
    metrics_path: "/metrics"
    bearer_token: "sk-my-admin-key-2026"  # master key로 인증

# docker-compose.yml에 Prometheus + Grafana 추가
services:
  # ... 기존 litellm, db, redis ...

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus_data:/prometheus
    command:
      - "--config.file=/etc/prometheus/prometheus.yml"
      - "--storage.tsdb.retention.time=30d"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
      GF_DATASOURCES_DEFAULT_URL: http://prometheus:9090
    volumes:
      - grafana_data:/var/lib/grafana
    depends_on:
      - prometheus
    restart: unless-stopped

volumes:
  prometheus_data:
  grafana_data:

[LiteLLM /metrics 엔드포인트에서 제공하는 메트릭]

요청 메트릭:
→ litellm_requests_total — 총 요청 수 (모델·상태별)
→ litellm_request_duration_seconds — 응답 시간 히스토그램
→ litellm_failed_requests_total — 실패 요청 수 (에러 유형별)

토큰 메트릭:
→ litellm_input_tokens_total — 입력 토큰 합계 (모델별)
→ litellm_output_tokens_total — 출력 토큰 합계 (모델별)

비용 메트릭:
→ litellm_spend_total — 총 비용 (팀·키·모델별)
→ litellm_team_budget_remaining — 팀 잔여 예산

캐시 메트릭:
→ litellm_cache_hits_total — 캐시 히트 수
→ litellm_cache_misses_total — 캐시 미스 수

[Grafana 대시보드 패널 예시]
→ 시간별 요청 수 (Line Chart)
→ P95 레이턴시 (Gauge)
→ 모델별 비용 (Bar Chart)
→ 팀별 예산 사용률 (Pie Chart)
→ 에러율 (Alert Panel — 5% 초과 시 알림)
→ 캐시 히트율 (Line Chart)

---

실전 5 — 프로덕션 보안 체크리스트

[2026년 3월 LiteLLM 공급망 사고 — 알아야 할 것]

사건:
→ 2026년 3월 24일, 해커 그룹 TeamPCP가 LiteLLM PyPI 배포 파이프라인 침해
→ 악성 버전 v1.82.7, v1.82.8 배포 (약 3시간 동안 40,000회 다운로드)
→ 자격증명 탈취 악성코드 포함

조치:
→ v1.82.7, v1.82.8 즉시 제거
→ 깨끗한 v1.83.0 배포
→ 해당 버전 사용 환경: 전체 자격증명 교체 필요

지금 해야 할 것:
☐ 현재 버전 확인: pip show litellm 또는 docker image 태그 확인
☐ v1.82.7/1.82.8 사용 시 즉시 v1.83.0+ 업그레이드
☐ 해당 버전이 설치됐던 환경의 모든 API 키 교체
☐ 이후 배포: 항상 특정 버전 고정 (latest 금지)

# ── 보안 강화 설정 ────────────────────────────────────

# 1. Docker 이미지 버전 고정 (latest 절대 금지)
# ❌ image: ghcr.io/berriai/litellm:latest
# ✅ image: ghcr.io/berriai/litellm:v1.83.2-stable

# 2. cosign으로 이미지 서명 검증
cosign verify \
  --key https://raw.githubusercontent.com/BerriAI/litellm/0112e53046018d726492c814b3644b7d376029d0/cosign.pub \
  ghcr.io/berriai/litellm:v1.83.2-stable

# 3. 프록시를 인터넷에 직접 노출 금지 → Nginx 리버스 프록시 앞에 두기
# 4. HTTPS 필수 (인터넷 facing 배포)
# 5. Master key를 비밀번호 관리자 또는 Vault에서 로드

# Nginx 리버스 프록시 설정 (HTTPS)
# nginx.conf
server {
    listen 443 ssl;
    server_name ai-gateway.company.com;

    ssl_certificate /etc/letsencrypt/live/ai-gateway.company.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/ai-gateway.company.com/privkey.pem;

    location / {
        proxy_pass http://litellm:4000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # 스트리밍 지원
        proxy_buffering off;
        proxy_read_timeout 300s;
    }
}

server {
    listen 80;
    server_name ai-gateway.company.com;
    return 301 https://$server_name$request_uri;  # HTTP → HTTPS 리다이렉트
}

---

실전 6 — 고가용성 설정

# docker-compose.yml — 다중 인스턴스 (고가용성)
services:
  litellm-1:
    image: ghcr.io/berriai/litellm:v1.83.2-stable
    ports:
      - "4001:4000"  # 첫 번째 인스턴스
    env_file: .env
    volumes:
      - ./config.yaml:/app/config.yaml:ro

  litellm-2:
    image: ghcr.io/berriai/litellm:v1.83.2-stable
    ports:
      - "4002:4000"  # 두 번째 인스턴스
    env_file: .env
    volumes:
      - ./config.yaml:/app/config.yaml:ro

  # Nginx 로드밸런서
  nginx:
    image: nginx:alpine
    ports:
      - "4000:80"
    volumes:
      - ./nginx-lb.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - litellm-1
      - litellm-2

# nginx-lb.conf — 로드밸런서
upstream litellm_backend {
    least_conn;
    server litellm-1:4000;
    server litellm-2:4000;
    keepalive 32;
}

server {
    listen 80;
    location / {
        proxy_pass http://litellm_backend;
        proxy_buffering off;
        proxy_read_timeout 300s;
    }
}

# config.yaml — Redis 트랜잭션 버퍼 (고트래픽 DB 데드락 방지)
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL
  use_redis_transaction_buffer: true  # 1000+ RPS 환경에서 필수

litellm_settings:
  cache: true
  cache_params:
    type: redis
    host: redis
    port: 6379

---

LiteLLM 솔직한 평가 — 언제 쓰고 언제 대안을 찾나

[LiteLLM이 잘 맞는 경우]
→ 셀프호스팅 필수 (데이터 주권, 에어갭 환경)
→ 팀 5~50명 규모, 월 100만 요청 이하
→ 빠른 프로토타이핑 + 점진적 프로덕션 전환
→ 오픈소스 커스터마이징 필요 (직접 수정 가능)
→ 파이썬 생태계 깊이 통합 필요

[LiteLLM 한계 — 알고 쓸 것]
→ Python GIL: 1000+ RPS에서 처리량 한계
→ DB 병목: 고트래픽 시 PostgreSQL 연결 고갈 (Redis 필수)
→ 보안: 공급망 사고 이력 (버전 고정 + 이미지 서명 검증 필수)
→ 엔터프라이즈 기능: 가드레일·SSO·감사 로그는 Enterprise 플랜 ($250/월~)
→ K8s: 2025년 9월 OOM 버그 이력 — K8s 배포 시 꼼꼼한 검증 필요

[대안 선택 기준]
관리형 SaaS 원함 → OpenRouter
고트래픽 + 오픈소스 → Bifrost (Go 기반, 9.4x 처리량)
기존 Kong 스택 → Kong AI Gateway
Cloudflare 인프라 → Cloudflare AI Gateway

---

마무리 — 4편 완결 + 시리즈 전체 정리

[LiteLLM 4편 시리즈 전체 정리]

1편: Python SDK — completion(), 모델 ID 체계, 스트리밍, 임베딩
2편: Router — 폴백·재시도, 로드밸런싱, 비용 추적, 예산, 캐싱
3편: Proxy 서버 — Docker 배포, 가상 키, 팀 관리, Langfuse 연동
4편: 고급 통합 — LangChain·LangGraph, 가드레일, Prometheus, 보안

[권장 도입 순서]
1. pip install litellm → SDK로 시작
2. Router 추가 → 폴백·비용 추적
3. Proxy 배포 → 팀 공유 게이트웨이
4. Langfuse + Prometheus → 옵저버빌리티
5. 트래픽 증가 시 Redis + 다중 인스턴스
6. 1000+ RPS 필요 시 Bifrost 전환 고려

---

관련 글

• LiteLLM 완전 가이드 1편 — Python SDK 기본
• LiteLLM 완전 가이드 2편 — Router, 폴백, 비용 추적
• LiteLLM 완전 가이드 3편 — Proxy 서버 팀 게이트웨이 구축
• OpenRouter 완전 가이드 1편 — 클라우드 대안