关于AI生产最佳实践的几点记录

docker run --gpus all \
  -p 8000:8000 \
  --shm-size=16g \
  -v /data/models:/models \
  vllm/vllm-openai:latest \
  --model /models/llama-3-8b-instruct \
  --tensor-parallel-size 1 \
  --gpu-memory-utilization 0.9 \
  --max-model-len 4096 \
  --max-num-seqs 256 \
  --dtype auto \
  --quantization awq

from vllm import LLM, SamplingParams

llm = LLM(
    model="path/to/quantized-model",
    quantization="awq",
    max_model_len=4096,
    gpu_memory_utilization=0.9,
    tensor_parallel_size=1
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512
)

outputs = llm.generate(["写一段AI生产环境监控的实践总结"], sampling_params)
for output in outputs:
    print(output.outputs[0].text)

from vllm import LLM
import gc
import torch

def reset_worker():
    global llm
    del llm
    gc.collect()
    torch.cuda.empty_cache()
    llm = LLM(...)

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_inference_with_retry(prompt, max_retries=3):
    try:
        return call_inference(prompt, timeout=10)
    except TimeoutError:
        raise  # 让重试策略自己处理

http {
    limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;

    server {
        location /v1/chat/completions {
            limit_req zone=api_limit burst=20 nodelay;
            proxy_pass http://llm_service;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_read_timeout 30s;
            proxy_send_timeout 30s;
        }
    }
}

# prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'vllm'
    static_configs:
      - targets: ['localhost:8000']
    metrics_path: '/metrics'

def log_generation_metrics(prompt, response, user_feedback, metrics):
    metrics['response_length'] = len(response)
    metrics['user_feedback'] = user_feedback
    metrics['has_moderation_flag'] = check_moderation(response)
    # 发送到你的监控系统

# docker-compose.yml
version: '3.8'
services:
  vllm-1:
    image: vllm/vllm-openai:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

  vllm-2:
    image: vllm/vllm-openai:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

  nginx:
    image: nginx:latest
    ports:
      - "8000:8000"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf

def should_use_new_model(user_id, traffic_percentage=0.01):
    hash_val = hashlib.md5(user_id.encode()).hexdigest()
    return int(hash_val[:8], 16) % 10000 < traffic_percentage * 10000

def get_model_for_user(user_id):
    if should_use_new_model(user_id, traffic_percentage=0.01):
        return new_model
    else:
        return old_model

def check_and_rollback():
    current_error_rate = get_metric('error_rate')
    if current_error_rate > 0.2:
        logger.error(f"Error rate too high: {current_error_rate}, rolling back...")
        rollback_to_previous_version()
        send_alert("Auto rollback triggered due to high error rate")

def batch_prompts(prompts, max_batch_size=8):
    for i in range(0, len(prompts), max_batch_size):
        yield prompts[i:i+max_batch_size]

def process_batch(batch):
    return llm.generate(batch, sampling_params)

import hashlib
import json

def cache_key(prompt, params):
    params_str = json.dumps(params, sort_keys=True)
    return f"llm:{hashlib.md5((prompt + params_str).encode()).hexdigest()}"

def get_cached_response(prompt, params):
    key = cache_key(prompt, params)
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    return None

def cache_response(prompt, params, response, ttl=3600):
    key = cache_key(prompt, params)
    redis.setex(key, ttl, json.dumps(response))