RAG 系统生产部署全流程

┌─────────────────────────────────────────────────────────────────┐
│                      RAG 生产架构                                │
├────────────────┬────────────────────────────────────────────────┤
│  数据层        │  文档加载 → 清洗 → 分块 → Embedding → 向量库    │
│  检索层        │  混合检索(向量+BM25) → 重排 → 上下文组装        │
│  生成层        │  Prompt 模板 → LLM 调用 → 流式输出 → 后处理     │
│  基础设施层    │  API网关 → 负载均衡 → 缓存 → 监控 → 日志        │
└────────────────┴────────────────────────────────────────────────┘

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import numpy as np

es = Elasticsearch(["http://es:9200"])

# 创建同时支持文本和向量的索引
mapping = {
    "mappings": {
        "properties": {
            "content": {
                "type": "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_smart"
            },
            "content_vector": {
                "type": "dense_vector",
                "dims": 1024,
                "index": True,
                "similarity": "cosine"
            },
            "metadata": {
                "type": "object",
                "enabled": False
            }
        }
    }
}

es.indices.create(index="rag_docs", body=mapping)

def hybrid_search(query_text, query_vector, top_k=5):
    """混合检索：BM25 + 向量检索 + RRF 融合"""
    body = {
        "query": {
            "bool": {
                "should": [
                    # BM25 文本检索
                    {
                        "match": {
                            "content": {
                                "query": query_text,
                                "boost": 1.0
                            }
                        }
                    }
                ]
            }
        },
        "knn": {
            "field": "content_vector",
            "query_vector": query_vector,
            "k": top_k * 2,
            "num_candidates": top_k * 10
        },
        "size": top_k,
        "_source": ["content", "metadata"]
    }
    
    results = es.search(index="rag_docs", body=body)
    return results["hits"]["hits"]

# docker-compose.yml
version: '3.8'

services:
  # API 网关
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - rag-api
    restart: always

  # RAG API 服务
  rag-api:
    build: .
    environment:
      - QDRANT_URL=http://qdrant:6333
      - ES_URL=http://elasticsearch:9200
      - REDIS_URL=redis://redis:6379
      - LLM_API_KEY=${LLM_API_KEY}
      - EMBEDDING_MODEL=bge-m3
      - EMBEDDING_URL=http://embedding-server:8080/embed
    depends_on:
      - qdrant
      - elasticsearch
      - redis
      - embedding-server
    restart: always
    deploy:
      resources:
        limits:
          memory: 4G
          cpus: "2.0"

  # 向量数据库
  qdrant:
    image: qdrant/qdrant:v1.12.0
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    restart: always

  # Elasticsearch（混合检索）
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.14.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
    volumes:
      - es_data:/usr/share/elasticsearch/data
    restart: always

  # Redis（缓存 + 限流）
  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 1gb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    restart: always

  # Embedding 服务
  embedding-server:
    image: ghcr.io/huggingface/text-embeddings-inference:cpu-1.5
    command: ["--model-id", "BAAI/bge-m3", "--port", "8080"]
    volumes:
      - embedding_models:/data
    restart: always

  # 监控
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

volumes:
  qdrant_data:
  es_data:
  redis_data:
  embedding_models:

# Prometheus 指标埋点
from prometheus_client import Counter, Histogram, Gauge

rag_search_duration = Histogram(
    'rag_search_duration_seconds',
    'RAG retrieval duration',
    ['search_type']  # hybrid/vector/keyword
)

rag_generation_duration = Histogram(
    'rag_generation_duration_seconds',
    'LLM generation duration'
)

rag_context_relevance = Gauge(
    'rag_context_relevance_score',
    'Average context relevance score'
)

# 使用示例
with rag_search_duration.labels('hybrid').time():
    results = hybrid_search(query, vector)