RAG 最佳实践

构建生产级 RAG 系统需要考虑性能、成本、可维护性等多个维度。本章总结实践经验，帮助你避坑提效。

文档处理最佳实践

分块策略

推荐配置:
┌─────────────────────────────────────────────────────────────┐
│  文档类型        │  块大小    │  重叠     │  特殊处理       │
├─────────────────────────────────────────────────────────────┤
│  技术文档        │  800字     │  200字    │  保持代码完整   │
│  法律文档        │  1500字    │  300字    │  按条款切分     │
│  FAQ/问答        │  300字     │  0字      │  按问答对切分   │
│  新闻/博客       │  500字     │  100字    │  按段落切分     │
│  API 文档        │  600字     │  150字    │  保持接口完整   │
└─────────────────────────────────────────────────────────────┘

元数据管理

每个分块应保留丰富的元数据：

{
    "chunk_id": "unique-id",
    "source": "document.pdf",
    "page": 5,
    "chunk_index": 12,
    "total_chunks": 50,
    "doc_type": "policy",
    "department": "hr",
    "created_at": "2024-01-15",
    "language": "zh",
    "title": "员工手册第三章"
}

文档清洗

def clean_document(text: str) -> str:
    """文档清洗"""
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text)
    
    # 移除特殊字符
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)
    
    # 统一标点
    text = text.replace(''', "'").replace('"', '"')
    
    # 移除页眉页脚（简单示例）
    lines = text.split('\n')
    lines = [l for l in lines if not l.strip().startswith('第')]
    
    return '\n'.join(lines)

检索优化

混合检索权重调优

# 权重调优经验值
WEIGHT_CONFIGS = {
    "general": {"bm25": 0.3, "vector": 0.7},      # 通用场景
    "technical": {"bm25": 0.4, "vector": 0.6},    # 技术文档（术语多）
    "conversational": {"bm25": 0.2, "vector": 0.8}, # 对话式（语义重要）
}

检索数量选择

初步检索数量 vs 重排序数量:

简单问题:  初步 10-20 → 重排序 3-5
中等问题:  初步 20-30 → 重排序 5-7
复杂问题:  初步 30-50 → 重排序 7-10

经验: 初步检索数量 = 重排序数量 × 4

过滤策略

# 使用元数据过滤提高精度
def build_filter(query_metadata: dict) -> dict:
    """构建过滤条件"""
    filters = []
    
    if query_metadata.get("department"):
        filters.append({"department": {"$eq": query_metadata["department"]}})
    
    if query_metadata.get("year"):
        filters.append({"year": {"$gte": query_metadata["year"]}})
    
    if query_metadata.get("doc_type"):
        filters.append({"doc_type": {"$in": query_metadata["doc_type"]}})
    
    return {"$and": filters} if len(filters) > 1 else filters[0] if filters else {}

性能优化

缓存策略

from functools import lru_cache
import hashlib

class RAGCache:
    def __init__(self, maxsize: int = 1000):
        self.cache = {}
        self.maxsize = maxsize
    
    def get_query_hash(self, query: str) -> str:
        """生成查询哈希"""
        return hashlib.md5(query.encode()).hexdigest()
    
    def get(self, query: str):
        """获取缓存"""
        key = self.get_query_hash(query)
        return self.cache.get(key)
    
    def set(self, query: str, result):
        """设置缓存"""
        if len(self.cache) >= self.maxsize:
            # 简单的 LRU：删除最早的一半
            keys = list(self.cache.keys())
            for k in keys[:len(keys)//2]:
                del self.cache[k]
        
        key = self.get_query_hash(query)
        self.cache[key] = result

批量处理

# 批量索引优化
def batch_index_documents(documents: List[Document], batch_size: int = 100):
    """批量索引"""
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        vectorstore.add_documents(batch)
        print(f"已索引 {min(i+batch_size, len(documents))}/{len(documents)}")

并行处理

from concurrent.futures import ThreadPoolExecutor
import asyncio

async def parallel_retrieve(queries: List[str], retriever):
    """并行检索"""
    tasks = [retriever.aretrieve(q) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

评估与监控

评估指标

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

def evaluate_rag(test_dataset):
    """评估 RAG 系统"""
    results = evaluate(
        test_dataset,
        metrics=[
            faithfulness,        # 忠实度：回答是否基于文档
            answer_relevancy,    # 相关性：回答是否回答问题
            context_precision,   # 精确度：检索文档的相关性
            context_recall       # 召回率：是否检索到所有相关信息
        ]
    )
    return results

质量监控

import logging
from datetime import datetime

class RAGMonitor:
    def __init__(self):
        self.logger = logging.getLogger("rag")
        self.stats = {
            "total_queries": 0,
            "avg_latency": 0,
            "avg_retrieval_time": 0
        }
    
    def log_query(self, query: str, latency: float, retrieval_time: float, sources: int):
        """记录查询"""
        self.stats["total_queries"] += 1
        
        # 更新平均值
        n = self.stats["total_queries"]
        self.stats["avg_latency"] = (self.stats["avg_latency"] * (n-1) + latency) / n
        self.stats["avg_retrieval_time"] = (self.stats["avg_retrieval_time"] * (n-1) + retrieval_time) / n
        
        # 记录日志
        self.logger.info({
            "timestamp": datetime.now().isoformat(),
            "query": query[:100],
            "latency": latency,
            "retrieval_time": retrieval_time,
            "sources": sources
        })

错误处理

常见错误及解决方案

class RAGErrorHandler:
    @staticmethod
    def handle_empty_results(query: str):
        """处理无检索结果"""
        return {
            "answer": "抱歉，我在知识库中没有找到相关信息。请尝试用其他方式描述您的问题。",
            "sources": [],
            "suggestion": "您可以：\n1. 使用更通用的关键词\n2. 联系管理员添加相关文档"
        }
    
    @staticmethod
    def handle_llm_error(error: Exception):
        """处理 LLM 错误"""
        return {
            "answer": "生成回答时出现问题，请稍后重试。",
            "error": str(error)
        }
    
    @staticmethod
    def handle_timeout():
        """处理超时"""
        return {
            "answer": "请求超时，请简化您的问题或稍后重试。",
            "error": "timeout"
        }

重试机制

import tenacity

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=10),
    retry=tenacity.retry_if_exception_type(Exception)
)
def query_with_retry(rag, question: str):
    """带重试的查询"""
    return rag.query(question)

成本优化

模型选择

成本对比（处理 1M tokens）:

嵌入模型:
- text-embedding-3-small: $0.02 (推荐)
- text-embedding-3-large: $0.13
- BGE-large-zh (本地): $0

LLM:
- gpt-4o-mini: $0.15/$0.60 (推荐)
- gpt-4o: $2.50/$10.00
- Claude Haiku: $0.25/$1.25

Token 优化

def optimize_context(documents: List[Document], max_tokens: int = 4000):
    """优化上下文长度"""
    # 按重要性排序（如果有分数）
    # 截断到 token 限制
    
    total_tokens = 0
    selected = []
    
    for doc in documents:
        doc_tokens = count_tokens(doc.page_content)
        if total_tokens + doc_tokens <= max_tokens:
            selected.append(doc)
            total_tokens += doc_tokens
        else:
            break
    
    return selected

def count_tokens(text: str) -> int:
    """估算 token 数量"""
    # 中文约 1.5 字/token，英文约 4 字符/token
    return len(text) // 2  # 简单估算

安全考虑

敏感信息过滤

def filter_sensitive_info(text: str) -> str:
    """过滤敏感信息"""
    # 过滤手机号
    text = re.sub(r'1[3-9]\d{9}', '***', text)
    
    # 过滤身份证
    text = re.sub(r'\d{17}[\dXx]', '***', text)
    
    # 过滤邮箱
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '***', text)
    
    return text

访问控制

def check_document_access(user_id: str, document_metadata: dict) -> bool:
    """检查文档访问权限"""
    # 实现基于角色的访问控制
    user_role = get_user_role(user_id)
    doc_access_level = document_metadata.get("access_level", "public")
    
    access_hierarchy = {
        "public": 0,
        "internal": 1,
        "confidential": 2,
        "secret": 3
    }
    
    return access_hierarchy.get(user_role, 0) >= access_hierarchy.get(doc_access_level, 0)

小结

RAG 最佳实践要点：

1. 分块要适中：平衡语义完整性和检索精度
2. 元数据要丰富：支持过滤和追溯
3. 混合检索优先：综合语义和关键词优势
4. 重排序提精度：精排显著提升效果
5. 缓存减成本：避免重复计算
6. 监控不可少：持续评估和优化
7. 安全要重视：过滤敏感信息，控制访问

下一步

• 速查表：常用代码片段和配置

参考资料

• LangChain Production Best Practices
• RAGAS Evaluation
• Pinecone Best Practices