프로덕션 배포: Docker + FastAPI + LangServe

# pip install langserve fastapi uvicorn
from fastapi import FastAPI
from langserve import add_routes
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

app = FastAPI(title="OHS AI", version="1.0")

# 체인 정의
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 OHS 도움 챗봇입니다."),
    ("user", "{question}"),
])
llm = ChatAnthropic(model="claude-sonnet-4-6")
chain = prompt | llm

# 자동으로 /chat/invoke, /chat/stream, /chat/batch 엔드포인트 생성
add_routes(app, chain, path="/chat")


# 실행: uvicorn app:app --reload
# 자동 docs: http://localhost:8000/docs
# 자동 playground: http://localhost:8000/chat/playground

FROM python:3.11-slim AS builder

WORKDIR /app

# 의존성 설치 (캐시 최적화)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 코드 복사
COPY . .

# 권한 (보안)
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

# Healthcheck
HEALTHCHECK --interval=30s --timeout=3s \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - REDIS_URL=redis://redis:6379
      - CHROMA_URL=http://chroma:8000
    depends_on:
      redis:
        condition: service_healthy
      chroma:
        condition: service_started
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    volumes:
      - redis-data:/data

  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8001:8000"
    volumes:
      - chroma-data:/chroma/chroma

volumes:
  redis-data:
  chroma-data:

# .env (절대 git commit X)
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...
LANGCHAIN_API_KEY=ls__...

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    anthropic_api_key: str
    openai_api_key: str
    langchain_api_key: str
    redis_url: str = "redis://localhost:6379"

    class Config:
        env_file = ".env"


settings = Settings()

import hashlib
import json
import redis
from langchain_core.caches import BaseCache

r = redis.Redis(host="redis", port=6379, decode_responses=True)


class RedisLLMCache(BaseCache):
    def lookup(self, prompt: str, llm_string: str):
        key = hashlib.sha256(f"{prompt}|{llm_string}".encode()).hexdigest()
        cached = r.get(f"llm:{key}")
        return json.loads(cached) if cached else None

    def update(self, prompt: str, llm_string: str, return_val):
        key = hashlib.sha256(f"{prompt}|{llm_string}".encode()).hexdigest()
        r.setex(f"llm:{key}", 3600, json.dumps(return_val))


import langchain
langchain.llm_cache = RedisLLMCache()


# 같은 질문 재호출 시 LLM 안 씀 — 즉시 응답 + 비용 0

# .github/workflows/deploy.yml
name: Deploy

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Build and push
        uses: docker/build-push-action@v5
        with:
          push: true
          tags: ghcr.io/myorg/ohs-ai:${{ github.sha }}

      - name: Deploy to Vercel/Fly.io
        run: |
          # fly deploy --image ghcr.io/myorg/ohs-ai:${{ github.sha }}
          # 또는 Vercel API

from fastapi import FastAPI
from prometheus_client import Counter, Histogram, generate_latest

app = FastAPI()

requests_total = Counter("requests_total", "Total requests", ["endpoint"])
request_duration = Histogram("request_duration_seconds", "Request duration")


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type="text/plain")


@app.middleware("http")
async def track_requests(request, call_next):
    requests_total.labels(endpoint=request.url.path).inc()
    with request_duration.time():
        return await call_next(request)