从零搭建RAG系统2026：端到端实战指南

┌─────────────────────────────────────────────┐
│                  用户接口                     │
├─────────────────────────────────────────────┤
│               API网关                        │
├──────────┬──────────┬───────────────────────┤
│ 查询处理  │ 检索引擎  │   生成引擎             │
├──────────┴──────────┴───────────────────────┤
│            数据处理流水线                      │
├─────────────────────────────────────────────┤
│     向量数据库  │  文档存储  │  缓存层          │
└─────────────────────────────────────────────┘

# 2026年推荐技术栈
tech_stack = {
    "embedding_model": "text-embedding-3-large",  # 或BGE-large-zh
    "vector_db": "Qdrant",  # 或Milvus
    "reranker": "bge-reranker-large",
    "llm": "GPT-4o-mini",  # 或开源模型
    "framework": "LangChain",  # 或LlamaIndex
    "cache": "Redis",
    "document_store": "PostgreSQL",
}

class DocumentProcessor:
    def __init__(self):
        self.chunker = RecursiveChunker(max_tokens=500)
        self.embedder = EmbeddingModel("text-embedding-3-large")
    
    async def process(self, documents):
        chunks = []
        for doc in documents:
            # 1. 解析文档
            text = await self.parse(doc)
            
            # 2. 分块
            doc_chunks = self.chunker.chunk(text)
            
            # 3. 添加元数据
            for i, chunk in enumerate(doc_chunks):
                chunks.append({
                    "id": f"{doc.id}-chunk-{i}",
                    "text": chunk,
                    "embedding": await self.embedder.embed(chunk),
                    "metadata": {
                        "doc_id": doc.id,
                        "doc_title": doc.title,
                        "chunk_index": i,
                        "source": doc.source
                    }
                })
        
        return chunks

class RetrievalEngine:
    def __init__(self):
        self.vector_store = QdrantClient(host="localhost", port=6333)
        self.reranker = Reranker("bge-reranker-large")
    
    async def search(self, query, top_k=20, rerank_top_k=5):
        # 1. 向量检索
        query_embedding = await self.embedder.embed(query)
        results = self.vector_store.search(
            collection_name="documents",
            query_vector=query_embedding,
            limit=top_k
        )
        
        # 2. 重排序
        reranked = await self.reranker.rerank(query, results, top_k=rerank_top_k)
        
        return reranked

class GenerationEngine:
    def __init__(self):
        self.llm = LLM("gpt-4o-mini")
        self.cache = RedisCache()
    
    async def generate(self, query, retrieved_docs):
        # 1. 检查缓存
        cache_key = hash(query + str([d.id for d in retrieved_docs]))
        cached = await self.cache.get(cache_key)
        if cached:
            return cached
        
        # 2. 构建prompt
        context = self.format_context(retrieved_docs)
        prompt = f"""
        基于以下参考信息回答问题。如果参考信息不足以回答，请说明。
        
        参考信息:
        {context}
        
        问题: {query}
        
        回答:
        """
        
        # 3. 生成
        answer = await self.llm.generate(prompt)
        
        # 4. 缓存
        await self.cache.set(cache_key, answer, ttl=3600)
        
        return answer

class RAGSystem:
    def __init__(self):
        self.processor = DocumentProcessor()
        self.retriever = RetrievalEngine()
        self.generator = GenerationEngine()
    
    async def ingest(self, documents):
        """导入文档"""
        chunks = await self.processor.process(documents)
        await self.retriever.vector_store.upsert(chunks)
    
    async def query(self, question):
        """查询"""
        # 1. 检索
        docs = await self.retriever.search(question)
        
        # 2. 生成
        answer = await self.generator.generate(question, docs)
        
        return {
            "answer": answer,
            "sources": [{"title": d.metadata["doc_title"], "text": d.text[:200]} for d in docs]
        }

# 1. 缓存热门查询
# 2. 预计算embedding
# 3. 并行检索
# 4. 流式生成
async def query_stream(self, question):
    docs = await self.retriever.search(question)
    async for token in self.generator.generate_stream(question, docs):
        yield token

# 1. 查询改写
query_rewritten = await self.rewrite_query(question)

# 2. 多路检索
vector_results = await self.vector_search(query)
keyword_results = await self.keyword_search(query)
fused = self.fuse(vector_results, keyword_results)

# 3. 自适应检索
if self.needs_multi_hop(question):
    docs = await self.multi_hop_retrieve(question)
else:
    docs = await self.simple_retrieve(question)

# docker-compose.yml
version: '3.8'
services:
  rag-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - QDRANT_URL=http://qdrant:6333
      - REDIS_URL=redis://redis:6379
    depends_on:
      - qdrant
      - redis
  
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_data:/qdrant/storage
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  
volumes:
  qdrant_data:

# 关键监控指标
metrics = {
    "query_latency_p50": "中位查询延迟",
    "query_latency_p99": "99%查询延迟",
    "retrieval_accuracy": "检索准确率",
    "answer_quality": "回答质量评分",
    "cache_hit_rate": "缓存命中率",
    "error_rate": "错误率",
    "token_cost": "Token消耗"
}