RAG 项目从 Claude 4.7 迁移到 DeepSeek V4 Pro：成本降低 95% 的实战复盘

四、完整 RAG Pipeline 实现

下面是一个完整的生产级 RAG Pipeline，整合了文档处理、向量检索和答案生成：

# complete_rag_pipeline.py
import re
import json
from typing import List, Dict, Tuple
from langchain.text_splitter import RecursiveCharacterTextSplitter
from vector_store import VectorStoreManager
from hybrid_rag_engine import HybridRAGEngine, ModelProvider

class CompleteRAGPipeline:
    def __init__(self):
        self.vector_store = VectorStoreManager("production_rag")
        self.engine = HybridRAGEngine(RAGConfig())
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""]
        )
    
    def ingest_documents(self, documents: List[Dict[str, str]]):
        """
        文档摄入流程：
        1. 文本分块
        2. 并行嵌入
        3. 批量写入向量数据库
        """
        all_chunks = []
        all_embeddings = []
        batch_size = 100
        
        for doc in documents:
            # 智能分块
            chunks = self.text_splitter.split_text(doc["content"])
            
            # 批量嵌入请求，减少 API 调用次数
            for i in range(0, len(chunks), batch_size):
                batch = chunks[i:i+batch_size]
                embeddings = asyncio.run(
                    self.engine.embed_documents(batch)
                )
                all_chunks.extend(batch)
                all_embeddings.extend(embeddings)
        
        # 构建文档对象
        doc_objects = [
            {
                "id": f"{doc['id']}_chunk_{j}",
                "content": chunk,
                "metadata": {"source": doc.get("source", "unknown")}
            }
            for j, chunk in enumerate(all_chunks)
        ]
        
        # 批量写入
        self.vector_store.upsert_documents(doc_objects, all_embeddings)
        print(f"成功摄入 {len(doc_objects)} 个文档块")
    
    def retrieve_and_generate(
        self, 
        query: str, 
        top_k: int = 5,
        return_sources: bool = True
    ) -> Dict:
        """
        检索+生成流程：
        1. 查询嵌入
        2. 向量相似度搜索
        3. 上下文组装
        4. LLM 生成答案
        """
        # 步骤1：查询嵌入
        query_embedding = asyncio.run(
            self.engine.embed_documents([query])
        )[0]
        
        # 步骤2：向量搜索
        search_results = self.vector_store.search(
            query_embedding, 
            top_k=top_k,
            score_threshold=0.72  # 过滤低相关度结果
        )
        
        # 步骤3：上下文组装
        context = "\n\n".join([
            f"[来源 {i+1}] {r['content']}"
            for i, r in enumerate(search_results)
        ])
        
        # 步骤4：动态选择模型生成
        prompt = f"""基于以下参考资料回答问题。如果资料不足以回答，请明确说明。

参考资料：
{context}

问题：{query}

回答："""
        
        response = asyncio.run(
            self.engine.generate_answer(
                query=prompt,
                context=context,
                quality_requirement=0.75  # 低于此值使用 Claude
            )
        )
        
        result = {
            "answer": response["text"],
            "model_used": response["provider"],
            "tokens_consumed": response["tokens_used"]
        }
        
        if return_sources:
            result["sources"] = [
                {
                    "content": r["content"][:200] + "...",
                    "score": round(r["score"], 3)
                }
                for r in search_results
            ]
        
        return result

使用示例
async def main():
    pipeline = CompleteRAGPipeline()
    
    # 摄入文档
    documents = [
        {
            "id": "doc_001",
            "content": "深度学习是机器学习的一个分支，专注于使用多层神经网络进行特征学习。",
            "source": "技术百科"
        },
        {
            "id": "doc_002", 
            "content": "RAG（检索增强生成）结合了检索系统和语言模型的优势，提高生成质量。",
            "source": "AI 教程"
        }
    ]
    pipeline.ingest_documents(documents)
    
    # 查询
    result = pipeline.retrieve_and_generate("什么是深度学习和 RAG？")
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

# 错误示例
claude_embedding = await engine.embed_with_claude("文本")
假设返回 [1024 维]
vector_store.upsert([claude_embedding])  # 向量维度不匹配

正确解决方案：重新生成嵌入或使用维度转换
async def migrate_embeddings(documents: List[Dict]):
    """迁移时重新生成所有嵌入"""
    # 获取旧向量
    old_vectors = vector_store.get_all_vectors()
    
    # 重新用 DeepSeek 生成
    texts = [doc["content"] for doc in documents]
    new_embeddings = await engine.embed_documents(texts)  # 1536 维
    
    # 批量更新
    for doc, new_emb in zip(documents, new_embeddings):
        vector_store.update_vector(doc["id"], new_emb)

# 错误示例：大批量直接处理
for batch in all_batches:
    embeddings = await engine.embed_documents(batch)  # 超时风险
    save_to_db(embeddings)  # 如果超时会丢失整个批次

正确解决方案：添加重试和断点续传
async def robust_batch_processing(
    batches: List[List[str]], 
    checkpoint_file: str = "checkpoint.json"
):
    """带断点续传的批量处理"""
    # 读取已完成的批次
    completed = set()
    if os.path.exists(checkpoint_file):
        with open(checkpoint_file) as f:
            completed = json.load(f)
    
    for i, batch in enumerate(batches):
        if str(i) in completed:
            continue
            
        for retry in range(3):
            try:
                embeddings = await engine.embed_documents(batch)
                save_batch_results(i, embeddings)
                completed.add(str(i))
                
                # 保存断点
                with open(checkpoint_file, 'w') as f:
                    json.dump(list(completed), f)
                break
            except Exception as e:
                if retry == 2:
                    print(f"批次 {i} 连续失败: {e}")
                    raise
                await asyncio.sleep(2 ** retry)  # 指数退避

# 错误示例：无限制并发
tasks = [process(doc) for doc in all_docs]
await asyncio.gather(*tasks)  # 可能触发限流

正确解决方案：使用信号量控制并发
class RateLimitedEngine:
    def __init__(self, max_concurrent: int = 50, rpm_limit: int = 1000):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rpm_tracker = RPMTracker(limit=rpm_limit)
    
    async def limited_request(self, text: str):
        async with self.semaphore:
            await self.rpm_tracker.wait_if_needed()
            
            try:
                return await self.engine.embed_documents([text])
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:
                    # 收到限流响应，主动退避
                    await asyncio.sleep(5)
                    return await self.limited_request(text)
                raise

class RPMTracker:
    """RPM（每分钟请求数）追踪器"""
    def __init__(self, limit: int):
        self.limit = limit
        self.requests = []
    
    async def wait_if_needed(self):
        now = time.time()
        # 清理超过 1 分钟的记录
        self.requests = [t for t in self.requests if now - t < 60]
        
        if len(self.requests) >= self.limit:
            wait_time = 60 - (now - self.requests[0])
            await asyncio.sleep(wait_time)
        
        self.requests.append(now)