5.3 企业级 RAG 部署

从开发原型到生产环境的完整部署指南

一个能在本地跑通的 RAG 系统和企业级生产系统之间，隔着性能优化、高可用、监控运维、安全隔离等多道门槛。本节将详细讲解如何将 Haystack RAG 系统从原型推进到可支撑真实业务负载的生产环境。

1. Docker 容器化

多阶段构建是生产级镜像的最佳实践，分离构建依赖和运行依赖：

# Dockerfile
# ========== 阶段1：构建 ==========
FROM python:3.11-slim AS builder

WORKDIR /app

# 系统构建依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ========== 阶段2：运行 ==========
FROM python:3.11-slim AS runtime

WORKDIR /app

# 只安装运行时系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 创建非 root 用户
RUN groupadd -r appuser && useradd -r -g appuser appuser

# 从构建阶段复制已安装的包
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin

COPY . .
RUN chown -R appuser:appuser /app

USER appuser

EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["gunicorn", "api_server:app", \
     "--bind", "0.0.0.0:8000", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--timeout", "120", \
     "--access-logfile", "-", \
     "--error-logfile", "-"]
​

Docker Compose 编排完整服务栈：

# docker-compose.yml
version: "3.8"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - APP_ENV=production
      - LOG_LEVEL=INFO
      - REDIS_URL=redis://redis:6379/0
      - DB_URL=postgresql://user:pass@postgres:5432/ragdb
      - QDRANT_HOST=qdrant
      - QDRANT_PORT=6333
    depends_on:
      redis:
        condition: service_healthy
      qdrant:
        condition: service_started
    deploy:
      resources:
        limits:
          memory: 4G
          cpus: "2"
      replicas: 2
    restart: unless-stopped
    healthcheck:
      test: curl -f http://localhost:8000/health
      interval: 20s
      timeout: 5s
      retries: 3

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./ssl:/etc/nginx/ssl:ro
    depends_on:
      - api
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: redis-cli ping
      interval: 10s

  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    restart: unless-stopped

  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus_data:/prometheus

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin123
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  redis_data:
  qdrant_data:
  prometheus_data:
  grafana_data:
​

2. 高性能文档存储方案

将内存文档存储替换为生产级的 Qdrant 向量数据库：

from haystack.document_stores.qdrant import QdrantDocumentStore
from qdrant_client import QdrantClient

# Qdrant 配置
QDRANT_CONFIG = {
    "host": "qdrant",        # Docker Compose 服务名
    "port": 6333,
    "collection": "rag_knowledge",
    "embedding_dim": 384,    # 匹配你的嵌入模型维度
    "hnsw_config": {
        "m": 16,             # 每层邻居数，越大越精确但越慢
        "ef_construct": 200  # 构建索引时搜索宽度
    },
    "optimizers_config": {
        "deleted_threshold": 0.2,
        "memmap_threshold": 50000,
        "default_segment_number": 5
    }
}

def create_production_store():
    """创建生产级文档存储"""
    client = QdrantClient(
        host=QDRANT_CONFIG["host"],
        port=QDRANT_CONFIG["port"]
    )
    
    store = QdrantDocumentStore(
        client=client,
        collection_name=QDRANT_CONFIG["collection"],
        embedding_dim=QDRANT_CONFIG["embedding_dim"],
        recreate_index=True,
        hnsw_config=QDRANT_CONFIG["hnsw_config"],
        optimizers_config=QDRANT_CONFIG["optimizers_config"]
    )
    
    return store
​

3. 性能优化策略

语义缓存：避免对相似问题重复计算：

import hashlib
import json
from datetime import timedelta
import redis

class SemanticCache:
    """基于 Redis 的语义缓存层"""
    
    def __init__(self, redis_url: str, ttl: int = 3600, similarity_threshold: float = 0.95):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl
        self.similarity_threshold = similarity_threshold
        self._embedder = SentenceTransformersTextEmbedder(
            model="sentence-transformers/all-MiniLM-L6-v2"
        )
        self._embedder.warm_up()
    
    def _get_key(self, text: str) -> str:
        return f"rag:cache:{hashlib.md5(text.encode()).hexdigest()}"
    
    def get(self, query: str) -> str | None:
        """查询缓存，返回命中的回答或 None"""
        cached = self.redis.get(self._get_key(query))
        if cached:
            data = json.loads(cached)
            return data["answer"]
        
        # 哈希前缀匹配（相似查询）
        query_embedding = self._get_embedding(query)
        pattern = "rag:cache:*"
        for key in self.redis.scan_iter(pattern):
            key_str = key.decode()
            cached_data = json.loads(self.redis.get(key_str))
            if self._cosine_similarity(query_embedding, cached_data["embedding"]) > self.similarity_threshold:
                return cached_data["answer"]
        
        return None
    
    def set(self, query: str, answer: str):
        """写入缓存"""
        embedding = self._get_embedding(query)
        data = {"answer": answer, "embedding": embedding.tolist()}
        self.redis.setex(
            self._get_key(query),
            timedelta(seconds=self.ttl),
            json.dumps(data, ensure_ascii=False)
        )
    
    def _get_embedding(self, text: str):
        result = self._embedder.run(text=text)
        return result["embedding"]
    
    @staticmethod
    def _cosine_similarity(a, b):
        import numpy as np
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
​

异步管道：使用 Haystack 的异步接口提升吞吐量：

import asyncio
from haystack.async_pipeline import AsyncPipeline

async def process_batch(pipeline, queries: list[str], concurrency: int = 5):
    """异步批量处理查询"""
    semaphore = asyncio.Semaphore(concurrency)
    
    async def process_one(query: str):
        async with semaphore:
            return await pipeline.run_async(
                text_embedder={"text": query},
                prompt_builder={"question": query}
            )
    
    tasks = [process_one(q) for q in queries]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results
​

连接池管理：确保数据库和 API 连接高效复用：

from contextlib import asynccontextmanager
import httpx

class ConnectionManager:
    """统一管理外部连接池"""
    
    def __init__(self):
        self._llm_client: httpx.AsyncClient = None
        self._db_pool = None
    
    @asynccontextmanager
    async def lifespan(self):
        """FastAPI 生命周期管理"""
        # 启动时初始化
        self._llm_client = httpx.AsyncClient(
            timeout=120.0,
            limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
        )
        # ... 数据库连接池初始化
        
        yield
        
        # 关闭时清理
        await self._llm_client.aclose()
        # ... 数据库连接池关闭
    
    @property
    def llm_client(self) -> httpx.AsyncClient:
        if not self._llm_client:
            raise RuntimeError("连接管理器未初始化")
        return self._llm_client
​

4. 监控与可观测性

自定义 Metrics 中间件：

import time
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from fastapi import Request, Response

# 定义指标
REQUEST_COUNT = Counter(
    "rag_requests_total", "Total RAG requests", ["endpoint", "status"]
)
REQUEST_LATENCY = Histogram(
    "rag_request_duration_seconds", "Request duration", ["endpoint"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
ACTIVE_REQUESTS = Gauge("rag_active_requests", "Currently active requests")
RETRIEVAL_SCORE = Histogram(
    "rag_retrieval_score", "Top retrieval score per query",
    buckets=[0.0, 0.3, 0.5, 0.7, 0.8, 0.9, 0.95, 1.0]
)
CACHE_HIT_RATE = Counter(
    "rag_cache_hits_total", "Cache hit/miss", ["result"]
)

class MetricsMiddleware:
    """Prometheus 监控中间件"""
    
    async def __call__(self, request: Request, call_next):
        ACTIVE_REQUESTS.inc()
        start_time = time.time()
        
        try:
            response = await call_next(request)
            status = str(response.status_code)
        except Exception:
            status = "500"
            raise
        finally:
            duration = time.time() - start_time
            ACTIVE_REQUESTS.dec()
            REQUEST_COUNT.labels(request.url.path, status).inc()
            REQUEST_LATENCY.labels(request.url.path).observe(duration)
        
        return response
​

Prometheus 配置：

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "rag-api"
    static_configs:
      - targets: ["api:8000"]
    metrics_path: "/metrics"

  - job_name: "qdrant"
    static_configs:
      - targets: ["qdrant:6333"]

  - job_name: "redis"
    static_configs:
      - targets: ["redis:6379"]
​

5. 高可用架构

Nginx 负载均衡配置：

# nginx.conf
upstream rag_api {
    least_conn;
    server api_1:8000 max_fails=3 fail_timeout=30s;
    server api_2:8000 max_fails=3 fail_timeout=30s;
    keepalive 32;
}

server {
    listen 80;
    server_name rag.example.com;
    
    # 请求体限制（大文档上传场景）
    client_max_body_size 50M;
    
    location / {
        proxy_pass http://rag_api;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        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_set_header X-Forwarded-Proto $scheme;
        
        # 超时配置
        proxy_connect_timeout 10s;
        proxy_read_timeout 120s;
        proxy_send_timeout 60s;
    }
    
    location /health {
        proxy_pass http://rag_api/health;
        access_log off;
    }
    
    # 速率限制
    location /ask {
        limit_req zone=api_limit burst=20 nodelay;
        proxy_pass http://rag_api;
    }
    
    limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
}
​

Kubernetes 部署清单（可选的进阶方案）：

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rag-api
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    metadata:
      labels:
        app: rag-api
    spec:
      containers:
      - name: api
        image: rag-api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 15
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        env:
        - name: APP_ENV
          value: "production"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: rag-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: rag-api
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
​

6. 安全配置

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

app = FastAPI()

API_KEY_HEADER = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    """API Key 鉴权中间件"""
    valid_keys = set(os.getenv("API_KEYS", "").split(","))
    if api_key not in valid_keys:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return api_key

# 敏感信息脱敏
from starlette.middleware.base import BaseHTTPMiddleware

class DataSanitizer(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        # 对响应中的敏感信息进行过滤
        return response
​

7. 运维检查清单

架构全景图

本节完整覆盖了从容器化、高性能存储、缓存优化、监控告警到高可用部署的企业级 RAG 系统架构。按照这套方案，你可以构建一个稳定支撑日均十万级查询量的生产级 RAG 服务。