🤖 AI 서비스 성능 개선 실전 가이드

import { NextRequest } from 'next/server';

export async function POST(req: NextRequest) {
  const { messages } = await req.json();

  const upstream = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
    },
    body: JSON.stringify({
      model: 'gpt-4o-mini',
      messages,
      stream: true,          // 스트리밍 활성화
      max_tokens: 1024,
    }),
  });

  // 업스트림 SSE를 클라이언트에 그대로 파이프
  return new Response(upstream.body, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive',
    },
  });
}

export async function streamChat(messages: Message[], onChunk: (text: string) => void) {
  const res = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages }),
  });

  const reader = res.body!.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const lines = decoder.decode(value).split('\n');
    for (const line of lines) {
      if (!line.startsWith('data: ') || line === 'data: [DONE]') continue;
      const delta = JSON.parse(line.slice(6))
        ?.choices?.[0]?.delta?.content;
      if (delta) onChunk(delta);
    }
  }
}

import anthropic

client = anthropic.Anthropic()

# 긴 시스템 프롬프트에 cache_control 적용
# 첫 호출 이후 cache hit 시 TTFT 최대 85% 감소
response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": LONG_SYSTEM_PROMPT,           # 수천 토큰의 컨텍스트
            "cache_control": {"type": "ephemeral"} # 캐시 마킹
        }
    ],
    messages=[{"role": "user", "content": user_query}],
)

# 캐시 적중 확인
usage = response.usage
print(f"Cache read tokens: {usage.cache_read_input_tokens}")
print(f"Cache creation tokens: {usage.cache_creation_input_tokens}")

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def classify_batch(texts: list[str]) -> list[str]:
    """여러 텍스트를 동시 분류 — 개별 호출 대비 3~5배 처리량"""
    tasks = [
        client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "텍스트를 긍정/부정/중립으로 분류하세요."},
                {"role": "user",   "content": text},
            ],
            max_tokens=10,
        )
        for text in texts
    ]
    responses = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in responses]

# 실행
results = asyncio.run(classify_batch(["좋아요", "별로예요", "평범해요"]))

services:
  vllm:
    image: vllm/vllm-openai:latest
    runtime: nvidia
    environment:
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
    command: >
      --model meta-llama/Llama-3.1-8B-Instruct
      --quantization awq                  # AWQ INT4 양자화
      --max-model-len 8192               # 컨텍스트 길이
      --max-num-seqs 256                 # 최대 동시 시퀀스 (연속 배칭)
      --gpu-memory-utilization 0.90      # GPU 메모리 90% 활용
      --enable-chunked-prefill           # 긴 프롬프트 청크 처리
    ports:
      - "8000:8000"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

import asyncio
from functools import lru_cache
from qdrant_client import AsyncQdrantClient
from openai import AsyncOpenAI

openai_client = AsyncOpenAI()
qdrant = AsyncQdrantClient(url="http://localhost:6333")

# 임베딩 캐싱 — 동일 쿼리 재호출 방지
@lru_cache(maxsize=1024)
def cache_key(query: str) -> str:
    return query.lower().strip()

embedding_cache: dict[str, list[float]] = {}

async def embed_with_cache(text: str) -> list[float]:
    key = cache_key(text)
    if key not in embedding_cache:
        res = await openai_client.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        )
        embedding_cache[key] = res.data[0].embedding
    return embedding_cache[key]

async def retrieve(query: str, top_k: int = 5) -> list[dict]:
    # 임베딩 + 검색 병렬화
    query_vec = await embed_with_cache(query)

    results = await qdrant.search(
        collection_name="docs",
        query_vector=query_vec,
        limit=top_k,
        with_payload=True,
        # HNSW 파라미터: ef 높을수록 정확하지만 느림 (기본 128)
        search_params={"hnsw_ef": 64},
    )
    return [r.payload for r in results]

import http from 'k6/http';
import { check, sleep } from 'k6';
import { Trend, Counter } from 'k6/metrics';

// AI 전용 커스텀 메트릭
const ttft    = new Trend('ai_time_to_first_token', true);
const totalTok = new Counter('ai_total_tokens');

export const options = {
  stages: [
    { duration: '1m', target: 10  }, // 워밍업
    { duration: '3m', target: 50  }, // 정상 부하
    { duration: '2m', target: 100 }, // 최대 부하
    { duration: '1m', target: 0   }, // 쿨다운
  ],
  thresholds: {
    'http_req_duration':          ['p(95)<15000'], // AI는 15s 이내
    'ai_time_to_first_token':     ['p(95)<1000'],  // TTFT p95 < 1s
    'http_req_failed':            ['rate<0.01'],
  },
};

const PROMPTS = [
  "Redis 캐싱 전략을 3가지 설명해주세요.",
  "API 응답 시간이 느릴 때 점검할 항목은?",
  "Kubernetes HPA 설정 방법을 알려주세요.",
];

export default function () {
  const prompt = PROMPTS[Math.floor(Math.random() * PROMPTS.length)];
  const start  = Date.now();

  const res = http.post(
    'http://localhost:8000/v1/chat/completions',
    JSON.stringify({
      model: 'meta-llama/Llama-3.1-8B-Instruct',
      messages: [{ role: 'user', content: prompt }],
      max_tokens: 256,
      stream: false,
    }),
    { headers: { 'Content-Type': 'application/json' } },
  );

  check(res, {
    'status 200':         (r) => r.status === 200,
    'has content':        (r) => r.json()?.choices?.[0]?.message?.content?.length > 0,
  });

  const body = res.json();
  if (body?.usage) {
    totalTok.add(body.usage.completion_tokens);
    // non-streaming: TTFT ≈ 전체 응답 시간 (보수적 측정)
    ttft.add(Date.now() - start);
  }

  sleep(Math.random() * 2 + 1); // 1~3초 think time
}

# vLLM Prometheus 메트릭 수집 설정
scrape_configs:
  - job_name: 'vllm'
    static_configs:
      - targets: ['vllm:8000']
    metrics_path: '/metrics'

# 핵심 PromQL 알림 규칙
groups:
  - name: ai-service
    rules:
      - alert: HighTTFT
        expr: histogram_quantile(0.95, vllm:time_to_first_token_seconds_bucket) > 1
        for: 5m
        annotations:
          summary: "TTFT p95가 1초를 초과했습니다"

      - alert: QueueBacklog
        expr: vllm:num_requests_waiting > 20
        for: 2m
        annotations:
          summary: "추론 큐 대기 요청이 20개를 초과했습니다"

      - alert: LowGPUUtilization
        expr: vllm:gpu_cache_usage_perc < 0.3
        for: 10m
        annotations:
          summary: "GPU 캐시 활용률이 30% 미만입니다 — 스케일 다운 고려"