RAG生产排坑指南：幻觉、漏检、延迟三大难题

# ❌ 容易幻觉的Prompt
prompt = f"""基于以下资料回答问题：
{context}
问题：{question}
"""

# ✅ 抗幻觉的Prompt
prompt = f"""你是一个严格的信息提取助手。请遵循以下规则：

1. **只使用**以下参考资料回答问题
2. 如果参考资料中没有相关信息，直接回答"根据现有资料无法回答此问题"
3. 不要添加任何参考资料中未提及的信息
4. 不要进行推理、猜测或补全
5. 回答中需要引用具体的资料来源

参考资料：
---
{context}
---

问题：{question}

回答格式：
[来源：文档名] 回答内容...
"""

def answer_with_confidence(question, retrieved_docs):
    """带置信度的回答"""
    
    # 第一步：评估检索质量
    relevance_prompt = f"""评估以下文档与问题的相关性(0-10):
    问题: {question}
    文档: {retrieved_docs[0].page_content[:500]}
    只返回数字。"""
    
    relevance_score = int(llm.invoke(relevance_prompt).strip())
    
    if relevance_score < 4:
        return {
            "answer": "抱歉，知识库中没有找到与您问题相关的信息。",
            "confidence": 0.2,
            "should_answer": False,
        }
    
    # 第二步：生成回答并自我验证
    answer = rag_chain.invoke(question)
    
    # 第三步：验证回答是否忠于上下文
    verify_prompt = f"""判断以下回答的每个陈述是否能在参考资料中找到支撑。
    
    参考资料: {retrieved_docs}
    回答: {answer}
    
    返回JSON: {{"faithful": true/false, "unsupported_claims": [...]}}"""
    
    verification = json.loads(llm.invoke(verify_prompt))
    
    return {
        "answer": answer if verification["faithful"] else "无法确认回答准确性",
        "confidence": relevance_score / 10.0,
        "should_answer": verification["faithful"],
    }

def generate_with_citations(question, retrieved_docs):
    """强制引用来源的生成"""
    
    # 给每个文档编号
    numbered_context = "\n\n".join([
        f"[{i+1}] {doc.page_content}" 
        for i, doc in enumerate(retrieved_docs)
    ])
    
    prompt = f"""基于以下编号的参考资料回答问题。每个陈述后必须标注来源编号[1]。

参考资料：
{numbered_context}

问题：{question}

要求：
- 每个事实陈述后标注来源编号，如"GraphRAG由微软提出[1]"
- 如果某个陈述来自多个文档，标注所有来源，如"[1][3]"
- 无法找到来源的陈述不要写出

回答："""
    
    return llm.invoke(prompt)

┌────────────────────────────────────────────────┐
│              漏检原因诊断流程                     │
├────────────────────────────────────────────────┤
│                                                  │
│  检查1: Embedding模型是否匹配文档语言？           │
│    └─ 否 → 换多语言Embedding模型                 │
│    └─ 是 → 继续                                  │
│                                                  │
│  检查2: 查询和文档的表述差异大吗？                 │
│    └─ 大 → 加Query Rewriting                    │
│    └─ 小 → 继续                                  │
│                                                  │
│  检查3: top_k是否足够？                          │
│    └─ 否 → 增大top_k                            │
│    └─ 是 → 继续                                  │
│                                                  │
│  检查4: 分块是否破坏了关键信息？                   │
│    └─ 是 → 调整分块策略                          │
│    └─ 否 → 继续                                  │
│                                                  │
│  检查5: 是否需要Reranker？                       │
│    └─ 是 → 加Cross-Encoder Reranker             │
│                                                  │
└────────────────────────────────────────────────┘

def query_rewrite(user_query, chat_history=None):
    """多策略查询改写"""
    
    # 策略1：同义词扩展
    rewrite_prompt = f"""将以下查询改写为3个不同表述的查询，保持语义不变。
    
    原始查询: {user_query}
    
    返回JSON列表。"""
    
    rewrites = json.loads(llm.invoke(rewrite_prompt))
    
    # 策略2：考虑对话历史的指代消解
    if chat_history:
        resolve_prompt = f"""基于对话历史，解析查询中的指代词。
        
        对话历史: {chat_history[-3:]}
        当前查询: {user_query}
        
        返回解析后的完整查询。"""
        resolved = llm.invoke(resolve_prompt)
        rewrites.append(resolved)
    
    # 策略3：HyDE（假设性文档Embedding）
    hyde_prompt = f"""假设你找到了完美回答，写一段答案。这段答案将用于检索。
    
    问题: {user_query}
    
    答案:"""
    hyde_doc = llm.invoke(hyde_prompt)
    rewrites.append(hyde_doc)
    
    return rewrites

# 对所有改写查询并行检索，合并去重
def multi_query_retrieve(question, vector_store, top_k=5):
    queries = query_rewrite(question)
    
    all_docs = []
    for q in queries:
        docs = vector_store.similarity_search(q, k=top_k)
        all_docs.extend(docs)
    
    # 去重（按内容hash）
    seen = set()
    unique_docs = []
    for doc in all_docs:
        h = hash(doc.page_content)
        if h not in seen:
            seen.add(h)
            unique_docs.append(doc)
    
    return unique_docs[:top_k * 2]

from sentence_transformers import CrossEncoder

# 第一阶段：向量检索（高召回，低精度）
candidates = vector_store.similarity_search(question, k=20)

# 第二阶段：Cross-Encoder重排
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
scores = reranker.predict(
    [(question, doc.page_content) for doc in candidates]
)

# 按重排分数排序
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
final_docs = [doc for doc, score in ranked[:5]]

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 稠密检索（语义）
dense_retriever = vector_store.as_retriever(
    search_kwargs={"k": 10}
)

# 稀疏检索（关键词）
bm25_retriever = BM25Retriever.from_documents(all_documents)
bm25_retriever.k = 10

# 混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4],  # 语义检索权重更高
)

docs = ensemble_retriever.invoke(question)

用户提问 → 总延迟 ≈ 3-6s
├── Embedding计算: 100-200ms (10%)
├── 向量检索: 50-150ms (5%)
├── Reranker: 100-300ms (8%)
├── Prompt组装: 50ms (2%)
├── LLM生成: 2000-4000ms (65%)
├── 后处理: 100ms (5%)
└── 网络传输: 200-500ms (5%)

# 最简单也最有效的优化：流式输出
def stream_rag_response(question):
    """流式返回RAG结果"""
    docs = retriever.invoke(question)
    context = "\n\n".join([d.page_content for d in docs])
    
    prompt = f"基于以下资料回答：\n{context}\n\n问题：{question}"
    
    # 流式返回，用户看到第一个token只需500ms
    for chunk in llm.stream(prompt):
        yield chunk.content

import asyncio

async def parallel_rag(question):
    """并行执行检索和Query改写"""
    
    # 并行：查询改写 + 直接检索
    rewrite_task = asyncio.create_task(query_rewrite_async(question))
    search_task = asyncio.create_task(vector_store.asimilarity_search(question, k=10))
    
    rewrites, initial_docs = await asyncio.gather(rewrite_task, search_task)
    
    # 并行：对每个改写查询检索
    multi_search_tasks = [
        vector_store.asimilarity_search(q, k=5) 
        for q in rewrites
    ]
    multi_results = await asyncio.gather(*multi_search_tasks)
    
    # 合并去重
    all_docs = merge_and_dedupe(initial_docs, *multi_results)
    
    # Reranker（此时可以并行计算分数）
    ranked_docs = await rerank_async(question, all_docs[:20])
    
    # 生成
    context = "\n\n".join([d.page_content for d in ranked_docs[:5]])
    answer = await llm.ainvoke(f"基于以下资料回答：\n{context}\n\n问题：{question}")
    
    return answer

from hashlib import md5

class RAGCache:
    def __init__(self, ttl=3600):
        self.cache = {}  # 生产环境用Redis
        self.ttl = ttl
    
    def _key(self, question):
        # 归一化后hash，减少缓存miss
        normalized = question.strip().lower()
        return md5(normalized.encode()).hexdigest()
    
    def get(self, question):
        key = self._key(question)
        entry = self.cache.get(key)
        if entry and time.time() - entry["time"] < self.ttl:
            return entry["answer"]
        return None
    
    def set(self, question, answer):
        key = self._key(question)
        self.cache[key] = {"answer": answer, "time": time.time()}

# 语义缓存：对相似查询返回缓存
class SemanticCache:
    def __init__(self, threshold=0.95):
        self.entries = []  # [(embedding, question, answer, time)]
        self.threshold = threshold
    
    def get(self, question, embedding):
        for emb, q, ans, t in self.entries:
            if time.time() - t > 3600:
                continue
            sim = cosine_similarity([embedding], [emb])[0][0]
            if sim >= self.threshold:
                return ans
        return None

# 方案A：用小模型做简单问答，大模型做复杂问答
def route_model(question, retrieved_docs):
    complexity = assess_complexity(question, retrieved_docs)
    if complexity == "simple":
        return ChatOpenAI(model="gpt-4o-mini")  # 快2-3倍
    return ChatOpenAI(model="gpt-4o")

# 方案B：Speculative Decoding
# 用小模型生成草稿，大模型验证，减少大模型计算量
# 2026年主流LLM推理框架已原生支持