RAG 知识库优化：别让 AI 一本正经地胡说八道

数据准备  →  检索优化  →  生成约束  →  评估监控

@startuml

top to bottom direction
skinparam defaultTextAlignment center
skinparam shadowing false
skinparam rectangle {
    RoundCorner 15
}

rectangle "用户提问" as User

rectangle "Query 理解\n& 改写" as QueryRewrite
rectangle "检索/召回\n(Retrieval)" as Retrieval
rectangle "重排序\n(Reranking)" as Reranking
rectangle "上下文组装\n& Prompt" as PromptAssembly
rectangle "LLM 生成\n(Generation)" as Generation
rectangle "后处理 &\n引用溯源" as PostProcess

User --> QueryRewrite
QueryRewrite --> Retrieval
Retrieval --> Reranking
Reranking --> PromptAssembly
PromptAssembly --> Generation
Generation --> PostProcess

@enduml

# 固定长度分块——省事但粗暴
chunks = [text[i:i+512] for i in range(0, len(text), 512)]

# 按语义边界分块——多花几行代码，少踩很多坑
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100,
    separators=["\n## ", "\n### ", "\n\n", "\n", "。", ".", " "],
    length_function=len
)
chunks = splitter.split_text(text)

def preprocess_document(doc):
    doc = remove_headers_footers(doc)
    tables = extract_tables(doc)
    images = extract_and_describe_images(doc)
    metadata = extract_metadata(doc)
    return doc, tables, images, metadata

文档层 (Document)  →  摘要索引 (用于粗召回)
  │
  ├── 段落层 (Section)  →  主索引 (用于精确检索)
  │     │
  │     └── 句子层 (Sentence)  →  细粒度索引 (用于精确匹配)
  │
  └── 表格/图片  →  结构化索引 (单独处理)

from typing import List

def hybrid_search(query: str, top_k: int = 10) -> List[dict]:
    vector_results = vector_store.similarity_search(query, k=top_k * 2)
    bm25_results = bm25_index.search(query, k=top_k * 2)

    # Reciprocal Rank Fusion
    fused = reciprocal_rank_fusion(
        [vector_results, bm25_results],
        weights=[0.6, 0.4]
    )
    return fused[:top_k]


def reciprocal_rank_fusion(result_lists, weights=None, k=60):
    scores = {}
    if weights is None:
        weights = [1.0] * len(result_lists)

    for results, weight in zip(result_lists, weights):
        for rank, doc in enumerate(results):
            doc_id = doc["id"]
            if doc_id not in scores:
                scores[doc_id] = 0
            scores[doc_id] += weight * (1.0 / (k + rank + 1))

    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

def query_rewrite(original_query: str, llm) -> List[str]:
    prompt = f"""请将以下用户问题改写为3个不同角度的检索查询：

    原始问题：{original_query}

    要求：
    1. 一个保持原意但更精确的版本
    2. 一个使用同义词/近义词的版本  
    3. 一个更宽泛的版本
    """
    queries = llm.generate(prompt)
    return [original_query] + queries

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

def rerank(query: str, candidates: List[str], top_k: int = 5):
    pairs = [[query, doc] for doc in candidates]
    scores = reranker.predict(pairs)

    ranked = sorted(
        zip(candidates, scores), 
        key=lambda x: x[1], 
        reverse=True
    )
    return ranked[:top_k]

BM25 / 关键词检索   →  擅长命中专有名词、编号、错误码
向量检索 / Dense    →  擅长命中语义相近的表达

RRF_score(d) = Σ 1 / (k + rank_i(d))

results = vector_store.similarity_search(
    query,
    k=10,
    filter={
        "department": "finance",
        "doc_type": "policy",
        "updated_after": "2025-01-01"
    }
)

RAG_SYSTEM_PROMPT = """你是一个专业的知识库问答助手。请严格基于以下检索到的参考资料回答用户问题。

## 规则
1. **只基于参考资料回答**，不要使用你的训练知识
2. 如果参考资料不足以回答问题，明确说"根据现有资料无法回答"
3. 回答中引用来源，格式为 [来源: 文档名称]
4. 如果多个来源有矛盾，指出差异并说明各自来源
5. 保持回答简洁、结构化

## 参考资料
{context}

## 用户问题
{question}
"""

def build_context(chunks: List[dict], max_tokens: int = 4000) -> str:
    context_parts = []
    current_tokens = 0

    for i, chunk in enumerate(chunks):
        chunk_tokens = count_tokens(chunk["text"])
        if current_tokens + chunk_tokens > max_tokens:
            break

        context_parts.append(
            f"[参考{i+1}] (来源: {chunk['source']}, "
            f"更新: {chunk['date']})\n{chunk['text']}"
        )
        current_tokens += chunk_tokens

    return "\n\n---\n\n".join(context_parts)

CITATION_PROMPT = """回答问题时，请在每个关键信息后标注来源编号。

格式示例：
公司的年假政策规定，入职满1年的员工享有5天年假[1]，
满5年的员工享有10天年假[2]。

最后列出参考来源：
[1] 《员工手册v3.2》第四章第二节
[2] 《2025年度假期政策更新》
"""

def reorder_for_attention(chunks: List[dict]) -> List[dict]:
    """最匹配的放开头和结尾，次匹配的放中间"""
    if len(chunks) <= 2:
        return chunks

    sorted_chunks = sorted(chunks, key=lambda x: x["score"], reverse=True)
    result = []
    left, right = [], []

    for i, chunk in enumerate(sorted_chunks):
        if i % 2 == 0:
            left.append(chunk)
        else:
            right.append(chunk)

    return left + list(reversed(right))

eval_dataset = [
    {
        "question": "公司的报销流程是什么？",
        "expected_answer": "提交申请→主管审批→财务审核→打款",
        "expected_sources": ["报销制度v2.1"],
        "category": "policy"
    },
    # ... 至少 50-100 条覆盖不同场景
]

class RAGEvaluator:
    def evaluate(self, question, generated, expected, retrieved_docs):
        return {
            # 检索质量
            "retrieval_precision": self.calc_precision(retrieved_docs, expected_sources),
            "retrieval_recall": self.calc_recall(retrieved_docs, expected_sources),

            # 生成质量
            "answer_relevance": self.llm_judge_relevance(question, generated),
            "faithfulness": self.llm_judge_faithfulness(generated, retrieved_docs),
            "correctness": self.llm_judge_correctness(generated, expected),

            # 实用指标
            "has_citation": bool(re.search(r'\[.*?\]', generated)),
            "response_length": len(generated),
            "latency_ms": self.last_latency
        }

def agentic_rag(question: str):
    # Step 1: LLM 分析问题，决定检索策略
    plan = llm.generate(f"""分析这个问题需要什么信息：
    问题：{question}

    输出：
    - 需要检索的子问题列表
    - 每个子问题的检索策略 (向量/关键词/结构化查询)
    """)

    # Step 2: 执行多轮检索
    all_contexts = []
    for sub_query in plan.sub_queries:
        results = search(sub_query)
        all_contexts.extend(results)

    # Step 3: 判断信息是否充分
    if llm.judge_sufficient(question, all_contexts):
        return llm.generate_answer(question, all_contexts)
    else:
        additional = llm.generate_followup_queries(question, all_contexts)
        # ... 继续检索

向量检索 → 找到相关段落
    +
知识图谱 → 找到关联实体和关系
    ↓
更完整的上下文

import hashlib

class RAGCache:
    def __init__(self, ttl=3600):
        self.cache = {}
        self.ttl = ttl

    def get_or_compute(self, query, search_fn, generate_fn):
        key = hashlib.md5(query.encode()).hexdigest()

        if key in self.cache and not self.is_expired(key):
            return self.cache[key]

        results = search_fn(query)
        answer = generate_fn(query, results)

        self.cache[key] = {"answer": answer, "sources": results}
        return self.cache[key]

□ 数据准备
  □ 文档清洗 (去噪声、格式统一)
  □ 语义分块 (非固定长度)
  □ 分块重叠 (10-20%)
  □ 元数据保留 (来源、日期、分类)
  □ 多层索引架构

□ 检索优化
  □ 混合检索 (向量 + BM25)
  □ Query 改写与扩展
  □ Reranking 重排序
  □ 元数据过滤
  □ 匹配的 Embedding 模型

□ 生成优化
  □ 严格的 Prompt 约束
  □ 上下文数量控制 (3-5 个)
  □ 注意力友好的排列顺序
  □ 引用溯源
  □ 兜底策略 (无法回答时的处理)

□ 评估监控
  □ 评估数据集 (50-100 条+)
  □ 多维度指标 (检索+生成+实用)
  □ 持续评估 pipeline
  □ 用户反馈收集与分析
  □ 线上监控告警