RAG 深度解析：从数学原理到生产级工程实战

目录

• 第一章 RAG 核心原理：大模型为什么要「开卷考试」
• 第二章 文档处理与智能切分
• 第三章 向量化与索引构建
• 第四章 检索策略深入：从单路到混合
• 第五章 生成与接地策略
• 第六章 Agentic RAG：让模型自主决策检索
• 第七章 评估体系：用数据证明 RAG 有效
• 第八章 生产工程化：从 Demo 到上线
• 第九章 总结与演进路线
• 附录 A：环境搭建与依赖安装

第一章 RAG 核心原理：大模型为什么要「开卷考试」

1.1 参数化记忆的天花板

大语言模型（LLM）的全部「知识」都压缩在权重矩阵里——我们称之为参数化记忆。这种记忆有两个根本性的限制：

知识截止问题（Knowledge Cutoff）

模型训练结束后，世界仍在变化。政策法规更新、产品版本迭代、公司制度修订……这些发生在训练截止日之后的信息，模型无法凭空知道。即使重新训练，从数据收集到模型上线的周期往往以月计，永远追不上现实世界的变化速度。

领域盲区问题（Domain Gap）

模型训练语料是公开互联网数据的采样，你公司的内部文档、产品手册、客户数据、私有知识——这些都不在训练集中。当用户问「我们公司的差旅报销标准是什么」时，模型只能编造一个听起来合理的答案，这就是幻觉（Hallucination）。

知识更新成本问题

即便发现了知识盲区，通过重新训练来补充知识的成本极高：GPT-4 级别的模型训练一次需要数千万美元和数周时间。而 RAG 只需要更新知识库里的文档，成本趋近于零。

1.2 RAG 的核心思想：检索增强生成

RAG（Retrieval-Augmented Generation）的核心思路极其朴素：

先帮模型「查资料」，再让模型基于资料「作答」。

类比考试场景：闭卷考试考的是记忆力，而开卷考试考的是理解力和表达力。RAG 就是给大模型一场开卷考试——参考书是你的私有知识库，模型的任务是从参考书中找到相关内容，并用自己的语言能力组织成自然流畅的回答。

从信息论的视角来看，这个过程可以形式化为：

传统 LLM 生成：P(Y | X)    — 给定问题 X，生成回答 Y
RAG 生成：   P(Y | X, C)   — 给定问题 X 和检索到的上下文 C，生成回答 Y

引入上下文 C 后，生成的条件熵从 H(Y|X) 降低到 H(Y|X,C)。上下文中包含了回答问题所需的证据，模型不再需要「猜」，而是基于证据「推导」，幻觉概率显著降低。

1.3 RAG 三段式架构

一个完整的 RAG 系统包含三个阶段：

索引阶段（Indexing，离线）

将原始文档（PDF、Word、HTML、Markdown 等）解析为纯文本，按语义或结构切分成适当大小的文本块（chunk），对每个文本块生成向量（Embedding），将向量和原文存入向量数据库，同时构建 BM25 稀疏索引。

检索阶段（Retrieval，在线）

用户提问后，将问题向量化，在向量数据库中搜索语义最相似的 Top-K 文本块；同时对问题进行分词，在 BM25 索引中搜索关键词匹配的 Top-K 文本块；将两路结果融合排序（如 RRF 算法），可选用交叉编码器（Cross-Encoder）精排，最终选出最相关的文本块作为上下文。

生成阶段（Generation，在线）

将检索到的上下文和用户问题组装成 Prompt，通过 System Prompt 约束模型只能基于提供的上下文回答，模型流式生成带引用标注的回答。

1.4 RAG vs Fine-tuning(微调) vs Long-context(长上下文)：三选一还是组合拳

工程选型原则：

1. 知识库频繁更新、需要可溯源 → RAG 是首选
2. 风格/格式/特定领域能力 → Fine-tuning 更合适
3. 单文档深度分析 → Long-context 直接塞入最简单
4. 生产系统 → RAG + Fine-tuning 组合：RAG 提供实时知识，Fine-tuning 让模型更擅长在 RAG 上下文中作答

第二章 文档处理与智能切分

本章目标：把 PDF、Markdown、HTML 等异构文档变成「机器能检索、模型能理解」的文本块序列。

2.1 为什么切分是 RAG 质量的第一个拐点

许多 RAG 系统在上线后表现不佳，追查根源往往不是模型不行，而是切分做得太粗糙。切分质量直接影响两件事：

• 检索精度：切得太大，chunk 内噪声多，向量被「平均化」，语义模糊；切得太小，上下文断裂，检索到了也无法回答问题。
• 生成质量：送进 LLM 的上下文由多个 chunk 拼接而成，chunk 的粒度决定了上下文的信息密度。

2.2 四种主流切分策略

策略一：递归字符切分（Recursive Character Splitting）

最基础也最通用的方法。按照一组优先级分隔符（\n\n → \n → 。 → ； → “）逐层尝试切分，直到每个 chunk 不超过目标大小。

def recursive_split(text: str, chunk_size: int = 500, overlap: int = 50,
          separators: list[str] | None = None) -> list[str]:
"""递归字符切分：按优先级分隔符逐层切分，直到满足 chunk_size 约束。"""
if separators is None:
    separators = ["\n\n", "\n", "。", "；", ".", ";", " ", ""]

if len(text) <= chunk_size:
return [text]

  chunks = []
# 找到当前层能用的分隔符
for sep in separators:
if sep == "":
# 最后兜底：硬切
for i in range(0, len(text), chunk_size - overlap):
        chunks.append(text[i:i + chunk_size])
return chunks
if sep in text:
      parts = text.split(sep)
      current_chunk = ""
for part in parts:
# 尝试拼接，超过 chunk_size 就断开
        candidate = current_chunk + sep + part if current_chunk else part
if len(candidate) <= chunk_size:
          current_chunk = candidate
else:
if current_chunk:
            chunks.append(current_chunk)
# 如果单个 part 超过 chunk_size，递归切分
if len(part) > chunk_size:
            sub_chunks = recursive_split(
              part, chunk_size, overlap, separators[separators.index(sep)+1:]
            )
            chunks.extend(sub_chunks)
            current_chunk = ""
else:
            current_chunk = part
if current_chunk:
        chunks.append(current_chunk)
return chunks

return [text]

策略二：语义切分（Semantic Splitting）

不依赖字符边界，而是计算相邻句子的 Embedding 相似度，在相似度骤降处断开。适合段落主题变化明显的长文本。

import numpy as np

def semantic_split(text: str, embed_fn, threshold: float = 0.5,
           max_chunk_size: int = 800) -> list[str]:
"""语义切分：在 Embedding 相似度骤降处断开。

  Args:
    text: 输入文本
    embed_fn: Embedding 函数，接受 list[str] 返回 list[list[float]]
    threshold: 相似度低于此值时断开
    max_chunk_size: 单个 chunk 最大字符数
  """
import re
# 先按句子切分
  sentences = re.split(r'(?<=[。！？.!?])', text)
  sentences = [s.strip() for s in sentences if s.strip()]

if len(sentences) <= 1:
return [text]

# 批量获取句子 Embedding
  embeddings = embed_fn(sentences)

  chunks = []
  current_chunk_sentences = [sentences[0]]

for i in range(1, len(sentences)):
# 计算当前句与前一句的余弦相似度
    sim = cosine_similarity(embeddings[i - 1], embeddings[i])
    current_text = "".join(current_chunk_sentences)

# 相似度骤降 或 超过最大长度 → 断开
if sim < threshold or len(current_text) + len(sentences[i]) > max_chunk_size:
      chunks.append("".join(current_chunk_sentences))
      current_chunk_sentences = [sentences[i]]
else:
      current_chunk_sentences.append(sentences[i])

if current_chunk_sentences:
    chunks.append("".join(current_chunk_sentences))

return chunks


def cosine_similarity(a: list[float], b: list[float]) -> float:
"""计算两个向量的余弦相似度。"""
  va, vb = np.array(a), np.array(b)
return float(np.dot(va, vb) / (np.linalg.norm(va) * np.linalg.norm(vb) + 1e-8))

策略三：结构化切分（Structure-aware Splitting）

针对 Markdown 文档，利用标题层级（#、##、###）作为天然的语义边界。每个 chunk 都携带其所属的标题层级信息。

import re

def markdown_structured_split(text: str, chunk_size: int = 500) -> list[dict]:
"""Markdown 结构化切分：以标题为界，每个 chunk 携带标题层级元数据。

  Returns:
    list[dict]: 每个元素包含 {"content": str, "headers": dict}
  """
# 按标题行切分
  sections = re.split(r'\n(?=#{1,3}\s)', text)

  chunks = []
for section in sections:
    section = section.strip()
if not section:
continue

# 提取标题层级
    headers = {}
    lines = section.split("\n")
    body_lines = []
for line in lines:
      m = re.match(r'^(#{1,3})\s+(.+)

策略四：Contextual Chunking（上下文增强切分）

核心洞察：一个 chunk 被检索到后，LLM 看到的是孤立的片段，缺乏「这段在讲什么主题」的背景。Contextual Chunking 在每个 chunk 前面拼接其所属章节的摘要或标题链，让模型看到 chunk 时能理解它的上下文位置。

def contextual_chunking(chunks: list[dict]) -> list[dict]:
"""为每个 chunk 拼接标题前缀，形成上下文增强版本。

  输入：markdown_structured_split 的输出
  输出：每个 chunk 的 content 前面加上标题链
  """
  result = []
for chunk in chunks:
    headers = chunk.get("headers", {})
# 构建标题链：Header 1 > Header 2 > Header 3
    header_chain = " > ".join(
      headers.get(f"Header {i}", "")
for i in range(1, 4)
if headers.get(f"Header {i}")
    )

# 拼接上下文前缀
if header_chain:
      enhanced_content = f"[章节: {header_chain}]\n{chunk['content']}"
else:
      enhanced_content = chunk["content"]

    result.append({
"content": enhanced_content,
"raw_content": chunk["content"],
"headers": headers,
    })

return result

2.3 完整的文档处理流水线

将上述切分策略封装成一个统一的文档处理器。现行默认参数：chunk_size=500、overlap=0（与 [config.py](../ragbackend/config.py) / [splitter.py](../ragbackend/splitter.py) 一致）。CLI 入库入口：python -m ragbackend.main ingest <文件>，内部调用 DocumentProcessor.process_file()。

import os
import hashlib
from dataclasses import dataclass, field
from typing import Literal


@dataclass
class DocumentChunk:
"""一个文本块的完整数据结构。"""
  chunk_id: str       # 唯一标识：{file_id}_chunk_{index}
  file_id: str      # 所属文件 ID
  source_name: str    # 原始文件名
  content: str      # 用于向量化和检索的内容（含上下文前缀）
  raw_content: str    # 原始文本（不含前缀）
  chunk_index: int    # 在文件中的序号
  headers: dict = field(default_factory=dict)  # 标题层级元数据
  metadata: dict = field(default_factory=dict)  # 其他元数据


class DocumentProcessor:
"""统一的文档处理器，支持多种输入格式和切分策略。

  Usage:
    processor = DocumentProcessor(chunk_size=500, strategy="markdown")
    chunks = processor.process_file("docs/policy.md")
  """

def __init__(
    self,
    chunk_size: int = 500,
    overlap: int = 0,
    strategy: Literal["recursive", "semantic", "markdown"] = "markdown",
    embed_fn=None,
):
self.chunk_size = chunk_size
self.overlap = overlap
self.strategy = strategy
self.embed_fn = embed_fn  # 语义切分时使用

def process_file(self, file_path: str) -> list[DocumentChunk]:
"""处理单个文件，返回 DocumentChunk 列表。"""
# 1. 读取文件内容
    text = self._read_file(file_path)
    file_id = self._generate_file_id(file_path)
    source_name = os.path.basename(file_path)

# 2. 根据策略切分
if self.strategy == "markdown" and file_path.endswith(".md"):
      raw_chunks = markdown_structured_split(text, self.chunk_size)
      enhanced_chunks = contextual_chunking(raw_chunks)
elif self.strategy == "semantic" and self.embed_fn:
      texts = semantic_split(text, self.embed_fn, max_chunk_size=self.chunk_size)
      enhanced_chunks = [{"content": t, "raw_content": t, "headers": {}} for t in texts]
else:
      texts = recursive_split(text, self.chunk_size, self.overlap)
      enhanced_chunks = [{"content": t, "raw_content": t, "headers": {}} for t in texts]

# 3. 封装为 DocumentChunk
    result = []
for i, chunk in enumerate(enhanced_chunks):
      doc_chunk = DocumentChunk(
        chunk_id=f"{file_id}_chunk_{i}",
        file_id=file_id,
        source_name=source_name,
        content=chunk["content"],
        raw_content=chunk["raw_content"],
        chunk_index=i,
        headers=chunk.get("headers", {}),
        metadata={"chunk_index": i, "source_name": source_name},
      )
      result.append(doc_chunk)

return result

def _read_file(self, file_path: str) -> str:
"""读取文件内容，根据扩展名选择解析方式。"""
    ext = os.path.splitext(file_path)[1].lower()

if ext == ".md":
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
elif ext == ".txt":
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
elif ext == ".html":
return self._parse_html(file_path)
elif ext == ".pdf":
return self._parse_pdf(file_path)
else:
raise ValueError(f"不支持的文件格式: {ext}")

def _parse_html(self, file_path: str) -> str:
"""简单的 HTML 解析，去除标签保留文本。"""
from html.parser import HTMLParser

class TextExtractor(HTMLParser):
def __init__(self):
super().__init__()
self.text_parts = []
self.skip = False

def handle_starttag(self, tag, attrs):
if tag in ("script", "style"):
self.skip = True

def handle_endtag(self, tag):
if tag in ("script", "style"):
self.skip = False

def handle_data(self, data):
if not self.skip:
self.text_parts.append(data)

with open(file_path, "r", encoding="utf-8") as f:
      html = f.read()

    parser = TextExtractor()
    parser.feed(html)
return "\n".join(parser.text_parts)

def _parse_pdf(self, file_path: str) -> str:
"""PDF 解析：优先使用 pypdf 提取文本。"""
try:
from pypdf import PdfReader
      reader = PdfReader(file_path)
      pages = []
for page in reader.pages:
        text = page.extract_text()
if text:
          pages.append(text)
return "\n\n".join(pages)
except ImportError:
raise ImportError("请安装 pypdf: pip install pypdf")

def _generate_file_id(self, file_path: str) -> str:
"""基于文件路径和内容生成唯一 ID。"""
with open(file_path, "rb") as f:
      content_hash = hashlib.md5(f.read()).hexdigest()[:12]
return f"{os.path.basename(file_path)}_{content_hash}"

2.4 chunk_size 调参经验

overlap 的经验法则：

• overlap=0 适合结构化切分（标题边界已经够清晰）
• overlap=50-100 适合递归字符切分（减少边界处信息丢失）
• overlap > 100 通常不推荐（增加索引体积，收益递减）

第三章 向量化与索引构建

本章目标：理解 Embedding 原理，构建稠密（向量）和稀疏（BM25）双索引，为混合检索奠定基础。

3.1 Embedding 原理：从词袋到语义空间

词袋模型的局限

传统的文本表示方法（如 TF-IDF、BM25）将文本表示为词频向量，维度等于词汇表大小（通常 10K-100K 维），向量极度稀疏，且无法捕捉语义关系——「电脑」和「计算机」在词袋模型中是完全不同的维度。

Transformer Embedding 的突破

现代 Embedding 模型（如 BERT、GTE、BGE）通过 Transformer 编码器将整个文本映射到一个低维稠密向量空间（通常 768-3072 维）。在这个空间中，语义相似的文本距离更近。

"公司差旅报销标准" → [0.12, -0.34, 0.56, ...]  (1024维)
"出差费用报销规定" → [0.11, -0.31, 0.54, ...]  (语义相近，向量也相近)
"今天天气真好"   → [-0.89, 0.42, -0.11, ...]  (语义不同，向量距离远)

3.2 Bi-Encoder vs Cross-Encoder

这是理解 RAG 检索架构的关键区分：

Bi-Encoder（双塔模型）:
  Query  → [Encoder] → q_vec ─┐
                ├─ cosine_similarity(q_vec, d_vec) → score
  Doc  → [Encoder] → d_vec ─┘

Cross-Encoder（交叉编码器）:
  [Query, Doc] → [Encoder] → score（直接输出相关性分数）

RAG 的黄金组合：Bi-Encoder 大规模召回 + Cross-Encoder 少量候选精排。

3.3 DashScope Embedding API 实战

DashScope 提供 text-embedding-v3 模型，支持 512/1024/2048 三种维度，最大输入 8192 tokens。

源码：[ragbackend/embedding.py](../ragbackend/embedding.py)。API Key 由 [config.py](../ragbackend/config.py) 的 _load_env() 自动加载，无需手写 dashscope.api_key = ...。

重要默认参数（[config.py](../ragbackend/config.py)）：

# 摘自 ragbackend/embedding.py，以仓库为准
from ragbackend.embedding import get_embeddings, get_single_embedding
from ragbackend.config import settings

# settings.embedding_batch_size == 10
embeddings = get_embeddings(["问题 A", "问题 B"], dimension=settings.embedding_dimension)
vec = get_single_embedding("单条文本")

入库与问答时通过 get_embeddings(texts) 分批调用；失败时抛出 RuntimeError 并附带 DashScope 错误码。

3.4 向量数据库选型

Demo 和小团队用 Chroma，生产环境根据规模选 Milvus/Qdrant。

3.5 Chroma 向量存储实现

import chromadb
from chromadb.config import Settings


class VectorStore:
"""基于 Chroma 的向量存储，支持批量写入和元数据过滤检索。

  Usage:
    store = VectorStore(collection_name="my_knowledge_base")
    store.add_chunks(chunks, embeddings)
    results = store.search(query_embedding, top_k=10)
  """

def __init__(self, collection_name: str = "knowledge_base", persist_dir: str = "./chroma_db"):
self.client = chromadb.PersistentClient(path=persist_dir)
self.collection = self.client.get_or_create_collection(
      name=collection_name,
      metadata={"hnsw:space": "cosine"},  # 使用余弦相似度
    )

def add_chunks(self, chunks: list[dict], embeddings: list[list[float]]) -> None:
"""批量写入文本块和对应的向量。

    Args:
      chunks: [{"chunk_id": str, "content": str, "metadata": dict}, ...]
      embeddings: 对应的向量列表
    """
    ids = [c["chunk_id"] for c in chunks]
    documents = [c["content"] for c in chunks]
    metadatas = []
for c in chunks:
      meta = c.get("metadata", {})
# Chroma 的 metadata 值只支持 str/int/float/bool
# 将 dict/list 转为 JSON 字符串
      clean_meta = {}
for k, v in meta.items():
if isinstance(v, (str, int, float, bool)):
          clean_meta[k] = v
else:
import json
          clean_meta[k] = json.dumps(v, ensure_ascii=False)
      metadatas.append(clean_meta)

# Chroma 单次最多约 5461 条，分批写入
    batch_size = 500
for i in range(0, len(ids), batch_size):
self.collection.upsert(
        ids=ids[i:i + batch_size],
        documents=documents[i:i + batch_size],
        embeddings=embeddings[i:i + batch_size],
        metadatas=metadatas[i:i + batch_size],
      )

print(f"已写入 {len(ids)} 个文本块到集合 '{self.collection.name}'")

def search(
    self,
    query_embedding: list[float],
    top_k: int = 10,
    where: dict | None = None,
) -> list[dict]:
"""向量相似度搜索。

    Args:
      query_embedding: 查询向量
      top_k: 返回数量
      where: 元数据过滤条件，如 {"source_name": "policy.md"}

    Returns:
      [{"chunk_id": str, "content": str, "score": float, "metadata": dict}, ...]
    """
    kwargs = {
"query_embeddings": [query_embedding],
"n_results": top_k,
"include": ["documents", "metadatas", "distances"],
    }
if where:
      kwargs["where"] = where

    results = self.collection.query(**kwargs)

    hits = []
for i in range(len(results["ids"][0])):
# Chroma 返回的是 distance（1 - similarity），转换为 similarity
      distance = results["distances"][0][i]
      similarity = 1.0 - distance

      hits.append({
"chunk_id": results["ids"][0][i],
"content": results["documents"][0][i],
"score": similarity,
"metadata": results["metadatas"][0][i] if results["metadatas"] else {},
      })

return hits

def delete_by_file(self, file_id: str) -> None:
"""删除指定文件的所有文本块。"""
self.collection.delete(where={"file_id": file_id})

def count(self) -> int:
"""返回集合中的文本块总数。"""
return self.collection.count()

def get_all_documents(self) -> list[dict]:
"""导出所有文档（用于重建 BM25 索引）。"""
    results = self.collection.get(include=["documents", "metadatas"])
    docs = []
for i in range(len(results["ids"])):
      docs.append({
"chunk_id": results["ids"][i],
"content": results["documents"][i],
"metadata": results["metadatas"][i] if results["metadatas"] else {},
      })
return docs

3.6 BM25 稀疏索引：关键词精确匹配

BM25（Best Matching 25）是经典的文本相关性打分算法，是 TF-IDF 的改进版本。其核心公式为：

BM25(q, d) = Σ IDF(qi) × [ f(qi, d) × (k1 + 1) ] / [ f(qi, d) + k1 × (1 - b + b × |d|/avgdl) ]

其中：

• f(qi, d) 是词 qi 在文档 d 中的词频
• |d| 是文档长度，avgdl 是平均文档长度
• k1（通常 1.2-2.0）控制词频饱和度
• b（通常 0.75）控制文档长度归一化

BM25 的优势在于精确匹配：当用户搜索特定编号（如「AI-2024-03号文」）、专有名词、产品型号时，BM25 比语义搜索更准确。

import jieba
import bm25s


class BM25Index:
"""基于 bm25s 的稀疏检索索引，使用 jieba 中文分词。

  Usage:
    index = BM25Index()
    index.build(documents)
    results = index.search("差旅报销标准", top_k=10)
  """

def __init__(self, k1: float = 1.5, b: float = 0.75):
self.k1 = k1
self.b = b
self.retriever = None
self.documents = []  # 原始文档列表
self.doc_ids = []  # 文档 ID 列表

def tokenize(self, text: str) -> list[str]:
"""使用 jieba 分词。"""
return list(jieba.cut(text))

def build(self, documents: list[dict]) -> None:
"""从文档列表构建 BM25 索引。

    Args:
      documents: [{"chunk_id": str, "content": str, "metadata": dict}, ...]
    """
self.documents = documents
self.doc_ids = [d["chunk_id"] for d in documents]

# 对所有文档分词
    corpus_tokens = [self.tokenize(d["content"]) for d in documents]

# 构建 BM25 索引
self.retriever = bm25s.BM25(k1=self.k1, b=self.b)
self.retriever.index(corpus_tokens)

print(f"BM25 索引构建完成：{len(documents)} 个文档")

def search(self, query: str, top_k: int = 10) -> list[dict]:
"""BM25 关键词搜索。

    Args:
      query: 查询文本
      top_k: 返回数量

    Returns:
      [{"chunk_id": str, "content": str, "score": float, "metadata": dict}, ...]
    """
if self.retriever is None or not self.documents:
return []

    query_tokens = self.tokenize(query)
    results, scores = self.retriever.retrieve(
      query_tokens, k=min(top_k, len(self.documents))
    )

    hits = []
for i in range(len(results[0])):
      doc_idx = results[0][i]
      score = float(scores[0][i])
      doc = self.documents[doc_idx]

      hits.append({
"chunk_id": doc["chunk_id"],
"content": doc["content"],
"score": score,
"metadata": doc.get("metadata", {}),
      })

return hits

def rebuild_from_vector_store(self, vector_store) -> None:
"""从向量数据库导出文档重建 BM25 索引（保证双索引一致性）。"""
    docs = vector_store.get_all_documents()
self.build(docs)

3.7 双索引一致性保障

混合检索的质量前提是稠密索引和稀疏索引看到同一批文档。工程上最容易踩的坑是：更新了向量库但忘了重建 BM25，导致两路检索结果不一致，融合后质量静默下降。

class DualIndexManager:
"""双索引管理器：确保 Chroma 和 BM25 始终同步。

  所有写入/删除操作都通过这个管理器，保证一致性。
  """

def __init__(self, vector_store: VectorStore, bm25_index: BM25Index):
self.vector_store = vector_store
self.bm25_index = bm25_index

def ingest(self, chunks: list[dict], embed_fn) -> None:
"""完整的入库流程：向量化 → 写入 Chroma → 重建 BM25。"""
# 1. 批量获取 Embedding
    texts = [c["content"] for c in chunks]
    embeddings = embed_fn(texts)

# 2. 写入向量库
self.vector_store.add_chunks(chunks, embeddings)

# 3. 从向量库全量导出，重建 BM25（保证一致性）
self.bm25_index.rebuild_from_vector_store(self.vector_store)

print(f"双索引更新完成：Chroma {self.vector_store.count()} 条，"
f"BM25 {len(self.bm25_index.documents)} 条")

def delete_file(self, file_id: str) -> None:
"""删除文件后同步重建 BM25。"""
self.vector_store.delete_by_file(file_id)
self.bm25_index.rebuild_from_vector_store(self.vector_store)

def initialize_on_startup(self) -> None:
"""启动时：如果 Chroma 有数据，自动重建 BM25。"""
if self.vector_store.count() > 0:
print("检测到 Chroma 有历史数据，正在重建 BM25 索引...")
self.bm25_index.rebuild_from_vector_store(self.vector_store)
print("BM25 索引重建完成")
else:
print("Chroma 为空，跳过 BM25 重建")

CLI 行为：[main.py](../ragbackend/main.py) 在 ingest / ask / eval 启动时均调用 dual_manager.initialize_on_startup()，若 Chroma 已有历史数据则自动重建 BM25，避免双索引不一致。

第四章 检索策略深入：从单路到混合

本章目标：实现生产级的混合检索系统，包括并行召回、RRF 融合、交叉编码器精排和查询增强策略。

4.1 为什么必须混合检索

单路检索永远有盲区。混合检索让两条路各自发挥优势，再通过融合算法取并集。

4.2 RRF 融合算法：不校准分数的排序融合

稠密相似度和 BM25 分数的量纲完全不同（一个是 0-1 的余弦相似度，一个是无界的 BM25 分数），直接加权求和很脆弱——权重难以调节，且不同查询下的分数分布差异很大。

Reciprocal Rank Fusion（RRF） 巧妙地绕开了分数校准问题：它只看排名，不看分数。

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

其中 rank_i(d) 是文档 d 在第 i 路检索结果中的排名（从 1 开始），k 是平滑常数（默认 60）。

直觉理解：一个文档如果在两路检索中都排在前列，它的 RRF 分数就高；如果只有一路命中，分数就低。k=60 意味着排名 60 以后的文档贡献几乎可以忽略。

from typing import Any


def reciprocal_rank_fusion(
  ranked_lists: list[list[dict[str, Any]]],
  k: int = 60,
) -> list[dict[str, Any]]:
"""Reciprocal Rank Fusion：融合多路检索结果。

  Args:
    ranked_lists: 多路检索结果，每路是按相关性排序的文档列表
    k: 平滑常数，默认 60

  Returns:
    融合后按 RRF 分数降序排列的文档列表
  """
  rrf_scores: dict[str, float] = {}
  doc_map: dict[str, dict[str, Any]] = {}

for rank_list in ranked_lists:
for rank, doc in enumerate(rank_list, start=1):
      doc_id = doc["chunk_id"]

# 累加 RRF 分数
      rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + 1.0 / (k + rank)

# 保留文档信息（取第一次出现的版本）
if doc_id not in doc_map:
        doc_map[doc_id] = doc.copy()
else:
# 合并元数据（两路可能返回不同的 metadata 信息）
        existing_meta = doc_map[doc_id].get("metadata", {})
        new_meta = doc.get("metadata", {})
        existing_meta.update(new_meta)

# 按 RRF 分数降序排列
  sorted_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)

# 附加 RRF 分数到文档中
  results = []
for doc_id in sorted_ids:
    doc = doc_map[doc_id]
    doc["rrf_score"] = rrf_scores[doc_id]
    results.append(doc)

return results


def weighted_rrf(
  ranked_lists: list[list[dict[str, Any]]],
  weights: list[float],
  k: int = 60,
) -> list[dict[str, Any]]:
"""加权 RRF：为不同检索路赋予不同权重。

  Args:
    ranked_lists: 多路检索结果
    weights: 每路的权重，如 [1.0, 0.8] 表示稠密路权重更高
    k: 平滑常数
  """
  rrf_scores: dict[str, float] = {}
  doc_map: dict[str, dict[str, Any]] = {}

for weight, rank_list in zip(weights, ranked_lists):
for rank, doc in enumerate(rank_list, start=1):
      doc_id = doc["chunk_id"]
      rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + weight / (k + rank)
if doc_id not in doc_map:
        doc_map[doc_id] = doc.copy()

  sorted_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)
  results = []
for doc_id in sorted_ids:
    doc = doc_map[doc_id]
    doc["rrf_score"] = rrf_scores[doc_id]
    results.append(doc)

return results

4.3 Cross-Encoder 精排

Bi-Encoder 适合大规模召回（毫秒级），Cross-Encoder 适合对少量候选进行精细排序（百毫秒级，但更准确）。

接线说明：精排使用 本地 Cross-Encoder（sentence-transformers），模型路径由 .env 的 RERANKER_MODEL_PATH 指定；CLI 中 [main.py](../ragbackend/main.py) 的 create_reranker() 在 USE_RERANKER=true 时加载并注入检索器。

# 摘自 ragbackend/reranker.py，以仓库为准
from ragbackend.reranker import CrossEncoderReranker, create_reranker

reranker = create_reranker()  # 读取 RERANKER_MODEL_PATH
retriever = HybridRetriever(..., reranker=reranker)

4.4 高级检索策略

集成状态（对照 [advanced_retrieval.py](../ragbackend/advanced_retrieval.py) 与 [retriever.py](../ragbackend/retriever.py)）：

策略	代码位置	默认 CLI
查询改写 rewrite	_prepare_queries() 已接入	需 RetrievalConfig(query_strategy="rewrite") + llm_call_fn
HyDE	hyde_retrieval()	未接入默认 ask 路径
Step-back	step_back_retrieval()	未接入默认 ask 路径
Parent-Child	ParentChildIndex	未接入默认 ask 路径

4.4.1 查询改写（Query Rewriting）

用户的原始查询可能表述不清、过于笼统，或者遗漏了关键信息。让 LLM 改写查询，生成多个角度的查询，分别检索后融合结果。

def query_rewriting(query: str, llm_call_fn, num_rewrites: int = 3) -> list[str]:
"""让 LLM 从不同角度改写查询，生成多个检索查询。

  Args:
    query: 用户原始查询
    llm_call_fn: LLM 调用函数，接受 prompt 返回文本
    num_rewrites: 改写数量

  Returns:
    包含原始查询和改写查询的列表
  """
  prompt = f"""你是一个查询改写助手。请将以下用户问题改写为 {num_rewrites} 个不同角度的搜索查询，
以提高在知识库中检索到相关信息的概率。

原始问题：{query}

要求：
1. 保持原始问题的核心意图
2. 每个改写使用不同的表述角度或关键词
3. 如果问题涉及专有名词，保留原名并尝试同义词
4. 每行输出一个查询，不要编号

请直接输出 {num_rewrites} 行改写后的查询："""

  response = llm_call_fn(prompt)
  rewrites = [line.strip() for line in response.strip().split("\n") if line.strip()]

# 始终包含原始查询
return [query] + rewrites[:num_rewrites]

4.4.2 HyDE：假设性文档嵌入

HyDE（Hypothetical Document Embeddings）是一种巧妙的检索增强策略：

1. 先让 LLM 生成一个「假设性的答案」（不需要准确，只需要格式和语义上接近真实答案）
2. 将这个假设答案作为查询去检索（因为假设答案和真实文档的语义更接近）
3. 用检索到的真实文档替换假设答案，送给 LLM 生成最终回答

def hyde_retrieval(query: str, llm_call_fn, embed_fn, vector_store,
           top_k: int = 10) -> list[dict]:
"""HyDE 检索：先生成假设答案，再用假设答案的向量去检索。

  原理：用户问题是疑问句，知识库文档是陈述句，二者语义空间有差距。
  假设答案是陈述句，与知识库文档的语义更接近，检索效果更好。
  """
# 第一步：生成假设性答案
  hyde_prompt = f"""请回答以下问题。如果你不确定答案，请根据你的一般知识给出最合理的推测。
不需要标注不确定性，直接陈述即可。

问题：{query}

回答："""

  hypothetical_answer = llm_call_fn(hyde_prompt)

# 第二步：将假设答案向量化（而不是将问题向量化）
  hyde_embedding = embed_fn([hypothetical_answer])[0]

# 第三步：用假设答案的向量检索真实文档
  results = vector_store.search(hyde_embedding, top_k=top_k)

# 附加 HyDE 信息用于调试
for r in results:
    r["retrieval_method"] = "hyde"
    r["hypothetical_answer"] = hypothetical_answer

return results

4.4.3 Step-back Prompting

当用户的问题非常具体时，先让 LLM 抽象出一个更宏观的问题，检索宏观层面的背景信息，再结合具体问题的检索结果一起生成答案。

def step_back_retrieval(query: str, llm_call_fn, retrieve_fn) -> list[dict]:
"""Step-back 检索：同时检索原始问题和抽象化问题。

  例：原始问题 "2024年Q3差旅报销新标准中打车费上限是多少"
    Step-back: "公司差旅报销政策的一般规定和标准"
  """
  step_back_prompt = f"""请将以下具体问题抽象为一个更宏观的背景问题。
宏观问题应该能帮助检索到相关的背景信息和政策框架。

具体问题：{query}

请输出一个宏观背景问题（一行）："""

  abstract_query = llm_call_fn(step_back_prompt).strip()

# 并行检索两个查询
  specific_results = retrieve_fn(query, top_k=5)
  abstract_results = retrieve_fn(abstract_query, top_k=5)

# 标记来源
for r in specific_results:
    r["retrieval_method"] = "specific"
for r in abstract_results:
    r["retrieval_method"] = "step_back"

# RRF 融合
return reciprocal_rank_fusion([specific_results, abstract_results])

4.4.4 Parent-Child Retrieval

核心思想：用小块（child）做精确检索，但送进 LLM 的是包含该小块的更大段落（parent），保留完整上下文。

class ParentChildIndex:
"""Parent-Child 索引：小块检索，大块送生成。

  构建时：
  1. 将文档切分为大块（parent，如 1500 字）
  2. 将每个大块再切分为小块（child，如 300 字）
  3. 只对小块建向量索引
  4. 维护 child → parent 的映射关系

  检索时：
  1. 在小块索引中搜索 Top-K
  2. 找到命中的小块对应的父块
  3. 返回父块内容（去重）
  """

def __init__(self, vector_store: VectorStore):
self.vector_store = vector_store
self.parent_map: dict[str, dict] = {}  # child_id → parent_doc

def build(self, documents: list[str], embed_fn,
        parent_size: int = 1500, child_size: int = 300) -> None:
"""构建 Parent-Child 索引。"""
    child_chunks = []

for doc_idx, doc_text in enumerate(documents):
# 切分为 parent 块
      parent_chunks = recursive_split(doc_text, chunk_size=parent_size, overlap=0)

for p_idx, parent_text in enumerate(parent_chunks):
        parent_id = f"doc_{doc_idx}_parent_{p_idx}"
        parent_doc = {
"parent_id": parent_id,
"content": parent_text,
        }

# 将 parent 切分为 child 块
        children = recursive_split(parent_text, chunk_size=child_size, overlap=0)

for c_idx, child_text in enumerate(children):
          child_id = f"{parent_id}_child_{c_idx}"

# 建立 child → parent 映射
self.parent_map[child_id] = parent_doc

          child_chunks.append({
"chunk_id": child_id,
"content": child_text,
"metadata": {
"parent_id": parent_id,
"child_index": c_idx,
            },
          })

# 只对 child 块建向量索引
    texts = [c["content"] for c in child_chunks]
    embeddings = embed_fn(texts)
self.vector_store.add_chunks(child_chunks, embeddings)

print(f"Parent-Child 索引构建完成：{len(self.parent_map)} 个 child 块")

def search(self, query_embedding: list[float], top_k: int = 5) -> list[dict]:
"""检索：child 命中 → 返回 parent 内容（去重）。"""
    child_results = self.vector_store.search(query_embedding, top_k=top_k * 3)

# 映射回 parent，去重
    seen_parents = set()
    parent_results = []

for child in child_results:
      parent = self.parent_map.get(child["chunk_id"])
if parent and parent["parent_id"] not in seen_parents:
        seen_parents.add(parent["parent_id"])
        parent_results.append({
"chunk_id": parent["parent_id"],
"content": parent["content"],
"score": child["score"],
"matched_child": child["content"][:100] + "...",
        })

if len(parent_results) >= top_k:
break

return parent_results

4.5 完整的混合检索器

将以上策略整合为统一检索器。完整实现见 [ragbackend/retriever.py](../ragbackend/retriever.py)。

与文档旧版差异（以源码为准）：

# 摘自 ragbackend/retriever.py，以仓库为准
from ragbackend.retriever import HybridRetriever, RetrievalConfig
from ragbackend.llm import make_llm_call_fn
from ragbackend.config import settings

config = RetrievalConfig(
  recall_k=settings.recall_k,     # 默认 5
  top_k=settings.top_k,
  use_reranker=settings.use_reranker,
  query_strategy="rewrite",     # 启用查询改写
)
retriever = HybridRetriever(
  vector_store=vector_store,
  bm25_index=bm25_index,
  config=config,
  embed_fn=lambda texts: get_embeddings(texts),
  llm_call_fn=make_llm_call_fn(settings.llm_model),
# reranker=CrossEncoderReranker(),  # 可选，须显式传入
)
results = retriever.retrieve("差旅报销标准")
timing = retriever.get_timing()  # vector_ms / bm25_ms / rrf_ms / rerank_ms / retrieve_total_ms

检索流程：_prepare_queries → 并行稠密/稀疏召回 → RRF 融合 → 可选精排 → 返回 Top-K。超时或单路失败时自动降级为可用一路的结果。

第五章 生成与接地策略

本章目标：将检索到的上下文安全、准确地送入 LLM，实现带引用的流式生成，并设计健壮的拒答机制。

5.1 Prompt 工程：接地约束是核心

RAG 生成阶段最关键的不是模型有多强，而是 Prompt 是否足够严格地约束模型只基于检索上下文作答。一个松散的 Prompt 会让模型在上下文不足时自由发挥，产生幻觉。

# 生产级 System Prompt 模板
RAG_SYSTEM_PROMPT = """你是一个专业的知识问答助手。你必须严格遵守以下规则：

## 回答规则
1. **仅依据提供的「参考资料」回答问题**，不要使用你自己的知识
2. 如果参考资料中没有足够的信息来回答问题，请直接回复：
   "根据知识库中的资料，暂时无法回答此问题。"
3. **绝对禁止编造**信息、数据、日期或来源
4. 回答时请引用资料来源，格式为 [来源: 文件名]

## 引用格式
- 在回答的关键事实后面标注引用，例如：报销上限为 500 元 [来源: 差旅报销制度.md]
- 如果多个来源支持同一结论，标注所有来源

## 回答风格
- 简洁、准确、结构化
- 优先使用列表和表格呈现结构化信息
- 不确定的内容明确标注"""


def build_rag_prompt(query: str, contexts: list[dict]) -> tuple[str, str]:
"""构建 RAG 的 system prompt 和 user prompt。

  Args:
    query: 用户问题
    contexts: 检索到的文档列表

  Returns:
    (system_prompt, user_prompt) 元组
  """
if not contexts:
return RAG_SYSTEM_PROMPT, query

# 组装上下文
  context_parts = []
for i, ctx in enumerate(contexts, 1):
    source = ctx.get("metadata", {}).get("source_name", "未知来源")
    content = ctx["content"]
    context_parts.append(f"### 参考资料 {i}\n来源: {source}\n内容: {content}")

  context_block = "\n\n".join(context_parts)

  user_prompt = f"""## 参考资料

{context_block}

---

## 用户问题

{query}

请基于以上参考资料回答用户问题。如果参考资料中没有相关信息，请明确说明。"""

return RAG_SYSTEM_PROMPT, user_prompt

5.2 拒答策略：宁可不说，不可乱说

RAG 系统最常见的错误不是「答错了」，而是「在不知道的情况下编造了一个听起来合理的答案」。健壮的拒答策略是 RAG 系统上线的安全底线。

class RefusalPolicy:
"""RAG 拒答策略：在多种情况下拒绝回答，防止幻觉。"""

def __init__(
    self,
    min_contexts: int = 1,      # 最少需要的上下文数量
    min_avg_score: float = 0.3,    # 平均相关性分数阈值
    min_max_score: float = 0.5,    # 最高相关性分数阈值
):
self.min_contexts = min_contexts
self.min_avg_score = min_avg_score
self.min_max_score = min_max_score

def should_refuse(self, contexts: list[dict]) -> tuple[bool, str]:
"""判断是否应该拒答。

    Returns:
      (should_refuse: bool, reason: str)
    """
# 策略 1：空检索 → 拒答
if not contexts:
return True, "未检索到相关资料"

# 策略 2：上下文数量不足
if len(contexts) < self.min_contexts:
return True, "检索到的资料不足以回答问题"

# 策略 3：最高分数低于阈值（所有结果都不太相关）
    scores = [c.get("score", c.get("rrf_score", 0)) for c in contexts]
    max_score = max(scores) if scores else 0
if max_score < self.min_max_score:
return True, f"检索结果相关性不足 (最高分: {max_score:.2f})"

# 策略 4：平均分数低于阈值
    avg_score = sum(scores) / len(scores) if scores else 0
if avg_score < self.min_avg_score:
return True, f"检索结果平均相关性偏低 (平均分: {avg_score:.2f})"

return False, ""


REFUSAL_MESSAGE = "根据知识库中的资料，暂时无法回答此问题。建议尝试换一种方式提问，或联系管理员确认知识库是否包含相关内容。"

5.2.5 统一 LLM 层（llm.py）

生成、查询改写、HyDE、Agent 与 RAGAS Judge 共用 [ragbackend/llm.py](../ragbackend/llm.py) 的模型工厂，避免各处硬编码 DashScope / DeepSeek 分支。

环境变量（[config.py](../ragbackend/config.py) 的 Settings.__post_init__ 读取）：

# 摘自 ragbackend/llm.py，以仓库为准
from ragbackend.llm import create_chat_llm, make_llm_call_fn

llm = create_chat_llm("qwen3.7-max", streaming=False)
llm_call_fn = make_llm_call_fn()  # 默认 settings.llm_model

5.3 流式生成与 Citation

生成器通过 LangChain Chat 模型调用，不再直接使用 dashscope.Generation.call。完整实现见 [ragbackend/generator.py](../ragbackend/generator.py)。

• 默认模型：qwen3.7-max（可通过 .env 的 llm_model 覆盖）
• 非流式：llm.invoke(SystemMessage + HumanMessage)，返回 timing.llm_total_ms
• 流式：llm.stream(...)，SSE 事件形态不变

SSE 事件类型：

[main.py](../ragbackend/main.py) 的 cmd_ask 在直连模式下会打印：检索 {retrieve_total_ms}ms | LLM {llm_total_ms}ms | 合计。

# 摘自 ragbackend/generator.py，以仓库为准
from langchain_core.messages import HumanMessage, SystemMessage
from ragbackend.generator import RAGGenerator
from ragbackend.llm import create_chat_llm
from ragbackend.prompts import build_rag_prompt

class RAGGenerator:
def __init__(self, model: str | None = None, max_tokens: int | None = None):
self.model = model or settings.llm_model  # 默认 qwen3.7-max
self.max_tokens = max_tokens or settings.llm_max_tokens

def generate(self, query: str, contexts: list[dict]) -> dict:
    should_refuse, reason = self.refusal_policy.should_refuse(contexts)
if should_refuse:
return {"answer": REFUSAL_MESSAGE, "refused": True, "timing": {}}

    system_prompt, user_prompt = build_rag_prompt(query, contexts)
    messages = [SystemMessage(content=system_prompt), HumanMessage(content=user_prompt)]

    start = time.time()
    llm = create_chat_llm(self.model, streaming=False, max_tokens=self.max_tokens)
    response = llm.invoke(messages)
    answer = response.content if hasattr(response, "content") else str(response)

return {
"answer": answer,
"citations": self._extract_citations(contexts),
"refused": False,
"timing": {"llm_total_ms": (time.time() - start) * 1000},
"llm_model": self.model,
    }

def stream_generate(self, query: str, contexts: list[dict]):
# 拒答 → yield refusal + done
# 否则 llm.stream(messages) → yield content / timing / citation / done
    ...

5.4 Token 预算管理

LLM 有上下文长度限制（如 qwen3.7-max 支持长上下文，实际仍受 max_context_tokens 预算约束）。检索到的多个 chunk 拼接后可能超出预算。需要合理分配上下文窗口。

def estimate_tokens(text: str) -> int:
"""粗略估算中文文本的 token 数。

  经验法则：中文 1 个字 ≈ 1-2 个 token，英文 1 个单词 ≈ 1-1.5 个 token。
  这里取保守估计。
  """
# 简单近似：字符数 × 0.7
return int(len(text) * 0.7)


def budget_contexts(
  contexts: list[dict],
  max_context_tokens: int = 4000,
  reserved_for_answer: int = 2000,
  model_max_tokens: int = 8192,
) -> list[dict]:
"""在 token 预算内选择最优的上下文组合。

  策略：按相关性从高到低逐个加入，直到预算耗尽。

  Args:
    contexts: 按相关性排序的检索结果
    max_context_tokens: 上下文最大 token 数
    reserved_for_answer: 为模型回答预留的 token 数
    model_max_tokens: 模型最大上下文
  """
# System prompt 大约占用 200 tokens
  system_tokens = 200
  available = min(max_context_tokens, model_max_tokens - system_tokens - reserved_for_answer)

  selected = []
  used_tokens = 0

for ctx in contexts:
    ctx_tokens = estimate_tokens(ctx["content"])
if used_tokens + ctx_tokens <= available:
      selected.append(ctx)
      used_tokens += ctx_tokens
else:
# 尝试截断当前 chunk（保留前半部分）
      remaining = available - used_tokens
if remaining > 100:  # 至少保留 100 tokens 才有意义
        truncated_content = ctx["content"][:int(remaining / 0.7)]
        selected.append({**ctx, "content": truncated_content, "truncated": True})
break

return selected

5.5 端到端 RAG Pipeline

将检索和生成组合为完整的 RAG 流水线：

class RAGPipeline:
"""端到端 RAG 流水线：检索 + 上下文预算 + 生成。

  Usage:
    pipeline = RAGPipeline(retriever, generator)

    # 非流式
    result = pipeline.ask("差旅报销标准是什么？")
    print(result["answer"])

    # 流式
    for event in pipeline.ask_stream("差旅报销标准是什么？"):
      if event["type"] == "content":
        print(event["content"], end="", flush=True)
  """

def __init__(
    self,
    retriever: HybridRetriever,
    generator: RAGGenerator,
    max_context_tokens: int = 4000,
):
self.retriever = retriever
self.generator = generator
self.max_context_tokens = max_context_tokens

def ask(self, query: str) -> dict:
"""非流式问答。"""
# 1. 检索
    contexts = self.retriever.retrieve(query)

# 2. 上下文预算管理
    contexts = budget_contexts(contexts, self.max_context_tokens)

# 3. 生成
    result = self.generator.generate(query, contexts)

# 4. 附加检索元信息
    result["retrieval_timing"] = self.retriever.get_timing()
    result["contexts_used"] = len(contexts)

return result

def ask_stream(self, query: str):
"""流式问答。"""
# 1. 检索
    contexts = self.retriever.retrieve(query)

# 2. 上下文预算管理
    contexts = budget_contexts(contexts, self.max_context_tokens)

# 3. 发送检索耗时
yield {
"type": "retrieval_info",
"data": {
"timing": self.retriever.get_timing(),
"contexts_count": len(contexts),
      },
    }

# 4. 流式生成
yield from self.generator.stream_generate(query, contexts)


# ask() 返回字段（与 ragbackend/pipeline.py 一致）：
# - answer, citations, refused, refusal_reason, timing（含 llm_total_ms）
# - retrieval_timing（来自 retriever.get_timing()）
# - contexts_used（送入 LLM 的 chunk 数量，非全文 contexts 列表）
# - llm_model（generator 实际使用的模型名）

编程接入示例见 附录 B 与 [ragbackend/main.py](../ragbackend/main.py)。

第六章 Agentic RAG：让模型自主决策检索

本章目标：将检索能力封装为 Agent 的工具，让 LLM 自主决定是否检索、检索什么、是否需要二次检索。

6.1 Direct RAG vs Agentic RAG

Direct RAG（标准两步式）

用户提问 → [强制检索] → [单轮生成] → 回答

• 优点：路径短、延迟可控、行为可预测
• 缺点：每次提问都检索（简单寒暄也检索）；只能检索一次；无法处理多跳问题

Agentic RAG（智能体式）

用户提问 → [LLM 推理] → 是否需要检索？
  ├─ 是 → [调用检索工具] → [LLM 评估结果] → 够了吗？
  │  ├─ 够 → [生成回答]
  │  └─ 不够 → [改写查询 / 换工具] → 再次检索 → ...
  └─ 否 → [直接回答 / 追问]

• 优点：按需检索、多步推理、自反思、多工具协同
• 缺点：多次 LLM 调用增加延迟和成本、行为更难预测

运行切换（[main.py](../ragbackend/main.py) + [config.py](../ragbackend/config.py)）：

# .env
RAG_USE_AGENT=false   # 或 true
python -m ragbackend.main ask "出差住宿费上限是多少？"

6.2 基于 LangChain 1.x 的 Agentic RAG 实现

LangChain 1.3+ 使用 create_agent（非旧版 AgentExecutor / create_react_agent）。完整实现见 [ragbackend/agent.py](../ragbackend/agent.py)。

# 摘自 ragbackend/agent.py，以仓库为准
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

from ragbackend.llm import create_chat_llm
from ragbackend.agent import (
  build_agent_tools, create_rag_agent, agentic_ask, _parse_agent_output,
)

# Step 1: 注入检索依赖（模块级，供 @tool 使用）
build_agent_tools(retriever, vector_store, embed_fn)

# Step 2: 创建 Agent
agent = create_rag_agent(llm_model=settings.llm_model)
# 内部：create_agent(model=create_chat_llm(...), tools=[...], system_prompt=...)

# Step 3: 调用（messages 格式，非旧版 input/chat_history）
result = agent.invoke({
"messages": [{"role": "user", "content": "出差住宿标准和报销流程分别是什么？"}]
})
parsed = _parse_agent_output(result)
# parsed["answer"] — 最终 AIMessage 文本
# parsed["steps"]   — [{tool, input, observation}, ...]

# 或直接使用封装好的 agentic_ask()
result = agentic_ask("2024年差旅报销新制度和旧制度有什么区别？")
print(result["answer"])
for step in result.get("steps", []):
print(step["tool"], step["input"])

与 Direct RAG 的差异：Agent 可多次调用 search_knowledge_base / search_with_filters，简单寒暄可跳过检索；代价是多次 LLM 调用与更高延迟。

6.3 自反思检索（Self-Reflective Retrieval）

集成状态：以下 self_reflective_retrieval 为 [advanced_retrieval.py](../ragbackend/advanced_retrieval.py) 中的参考实现，未接入默认 ask / agentic_ask 路径。生产接入需自行编排检索 → 评估 → 改写重试循环。

更高级的 Agentic 模式：Agent 检索后不直接回答，而是先评估检索结果的质量，如果不够好则自动改写查询重新检索。

def self_reflective_retrieval(
  query: str,
  retriever: HybridRetriever,
  llm_call_fn,
  max_retries: int = 2,
) -> dict:
"""自反思检索：检索 → 评估 → 必要时改写重试。

  Returns:
    {"answer": str, "contexts": list, "attempts": int, "queries_used": list}
  """
  current_query = query
  queries_used = [query]
  best_contexts = []
  best_score = 0

for attempt in range(max_retries + 1):
# 1. 检索
    contexts = retriever.retrieve(current_query)

# 2. 评估检索结果质量
if contexts:
      avg_score = sum(c.get("rrf_score", c.get("score", 0)) for c in contexts) / len(contexts)
else:
      avg_score = 0

# 保留最好的一轮结果
if avg_score > best_score:
      best_score = avg_score
      best_contexts = contexts

# 3. 如果结果足够好，直接返回
if avg_score > 0.5 and len(contexts) >= 3:
return {
"contexts": contexts,
"attempts": attempt + 1,
"queries_used": queries_used,
"quality": "good",
      }

# 4. 结果不够好，让 LLM 改写查询
if attempt < max_retries:
      rewrite_prompt = f"""我在知识库中搜索以下问题，但结果不够理想：

原始问题：{query}
已尝试的搜索：{', '.join(queries_used)}
当前检索质量评分：{avg_score:.2f}（满分 1.0）

请从不同角度改写搜索查询，帮助找到更相关的信息。
直接输出改写后的查询（一行）："""

      current_query = llm_call_fn(rewrite_prompt).strip()
      queries_used.append(current_query)

return {
"contexts": best_contexts,
"attempts": max_retries + 1,
"queries_used": queries_used,
"quality": "limited",
  }

第七章 评估体系：用数据证明 RAG 有效

本章目标：建立系统化的 RAG 评估方法，用 RAGAS 框架量化检索和生成质量，支撑持续优化。

7.1 为什么「体感测试」不够

许多团队在 RAG 上线后只靠「随机问几个问题看看回答是否合理」来评估质量。这种方法的问题：

1. 不可重复：每次问的问题不同，无法对比前后版本
2. 覆盖面窄：人的直觉只能覆盖常见场景，边界 case 容易遗漏
3. 无法量化：「感觉好了很多」不是工程决策的依据
4. 回归风险：改了切分策略后，某些问题变好了，另一些可能变差了，但你不知道

7.2 RAGAS 四大核心指标

RAGAS（Retrieval-Augmented Generation Assessment）是目前最广泛使用的 RAG 评估框架，核心指标分为检索质量和生成质量两个维度：

检索质量指标

生成质量指标

7.3 构建评测集

评测集是评估的基础。一个好的评测集应覆盖：常见问题、边界 case、多跳问题、否定问题（知识库里没有答案的问题）。

import json
from dataclasses import dataclass, asdict


@dataclass
class EvalSample:
"""评测样本。"""
  question: str       # 用户问题
  ground_truth: str     # 标准答案（人工标注）
  context: str      # 标准上下文（可选，标注时参考的源文档）
  difficulty: str     # easy / medium / hard
  category: str       # 问题类别
  metadata: dict = None   # 额外元数据


def build_eval_dataset() -> list[EvalSample]:
"""构建示例评测集。

  实际项目中应包含 50-200 条样本，覆盖：
  1. 简单事实问答（easy）
  2. 需要综合多个 chunk 的问题（medium）
  3. 多跳推理问题（hard）
  4. 知识库中不存在答案的问题（negative）
  """
  samples = [
    EvalSample(
      question="出差住宿费上限是多少？",
      ground_truth="国内出差住宿费上限为每天 500 元。",
      context="公司差旅报销制度规定，国内出差住宿费上限为每天 500 元。",
      difficulty="easy",
      category="事实查询",
    ),
    EvalSample(
      question="报销流程需要多久？",
      ground_truth="报销审批流程约 3-5 个工作日。",
      context="报销审批流程：提交人 → 直属领导 → 财务部门 → 打款。约 3-5 个工作日。",
      difficulty="easy",
      category="事实查询",
    ),
    EvalSample(
      question="2024 年的餐补标准是多少？",
      ground_truth="2024 年新版制度取消了原有餐补标准，改为每天 150 元的综合补贴。",
      context="2024 年新版报销制度取消了原有的餐补标准，改为每天 150 元的综合补贴。",
      difficulty="medium",
      category="变更查询",
    ),
    EvalSample(
      question="打车费报销需要哪些材料？流程是怎样的？",
      ground_truth="打车费属于交通费，需要附上打车发票和行程单。报销流程为提交人→直属领导→财务部门→打款，约 3-5 个工作日。",
      context="出差打车费用属于交通费范畴，需要附上打车发票和行程单。报销审批流程：提交人→直属领导→财务部门→打款，约3-5个工作日。",
      difficulty="hard",
      category="多跳推理",
    ),
    EvalSample(
      question="公司年假有多少天？",
      ground_truth="根据知识库中的资料，暂时无法回答此问题。",
      context="",
      difficulty="easy",
      category="否定测试",
    ),
  ]

return samples


def save_eval_dataset(samples: list[EvalSample], path: str = "eval_dataset.json"):
"""保存评测集到 JSON 文件。"""
  data = [asdict(s) for s in samples]
with open(path, "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
print(f"评测集已保存到 {path}，共 {len(samples)} 条")

现行 CLI 行为（[evaluator.py](../ragbackend/evaluator.py) + [main.py](../ragbackend/main.py)）：

• python -m ragbackend.main eval 调用内置 build_eval_dataset()，固定 5 条样本（easy / medium / hard / 否定测试各覆盖）。
• [data/eval/eval_dataset.json](../ragbackend/data/eval/eval_dataset.json) 存在但 不会自动加载；扩展评测集需自行调用 save_eval_dataset() 或修改 cmd_eval。
• RAGAS 限制：pipeline.ask() 返回 contexts_used（数量）而非 contexts（全文），因此 evaluate_rag_pipeline 中 retrieved_contexts 常为空，context_precision / context_recall 等指标可能偏低或无效——Faithfulness / Answer Relevancy 仍可参考。若需完整 Context 指标，需在 pipeline 中额外返回检索全文（后续可单独 PR 修复）。

7.4 RAGAS 评估流水线

与 ragas 0.4.3 API 对齐，Judge LLM 与 Embedding 均走 DashScope，避免默认 OpenAI。完整实现见 [ragbackend/evaluator.py](../ragbackend/evaluator.py)。

# 摘自 ragbackend/evaluator.py，以仓库为准
from ragas import EvaluationDataset, SingleTurnSample, evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from langchain_community.embeddings import DashScopeEmbeddings

from ragbackend.llm import create_chat_llm
from ragbackend.config import settings

def _evaluate_with_ragas(results: list[dict], llm_model: str) -> dict:
  dataset = EvaluationDataset(samples=[
    SingleTurnSample(
      user_input=r["question"],
      response=r["answer"],
      retrieved_contexts=r["retrieved_contexts"],
      reference=r["ground_truth"],
    )
for r in results
  ])

  eval_llm = LangchainLLMWrapper(create_chat_llm(llm_model, streaming=False))
  eval_embeddings = LangchainEmbeddingsWrapper(DashScopeEmbeddings(
    model=settings.embedding_model,
    dashscope_api_key=settings.dashscope_api_key,
  ))

  scores = evaluate(
    dataset=dataset,
    metrics=[context_precision, context_recall, faithfulness, answer_relevancy],
    llm=eval_llm,
    embeddings=eval_embeddings,
    raise_exceptions=False,  # 单条失败不中断整批
  )

def _metric(name: str):
try:
return float(scores[name])
except Exception:
if hasattr(scores, "to_pandas"):
        df = scores.to_pandas()
if name in df.columns:
return float(df[name].mean())
return None

return {
"context_precision": _metric("context_precision"),
"context_recall": _metric("context_recall"),
"faithfulness": _metric("faithfulness"),
"answer_relevancy": _metric("answer_relevancy"),
"details": results,
  }

未安装 ragas 时自动降级为 _simple_evaluate()（拒答准确率、回答非空率、平均长度）。

A/B 对比实验（调 chunk_size 等参数时，每次只改一个变量）：

def ab_test_chunk_sizes(documents, eval_samples, chunk_sizes=(200, 500, 800, 1200)):
for size in chunk_sizes:
    processor = DocumentProcessor(chunk_size=size)
# 重新入库 → evaluate_rag_pipeline → 记录 faithfulness / recall / precision
    ...

7.5 评估的工程纪律

1. 每次只改一个变量：切分大小、Embedding 模型、top_k、是否重排、Prompt 模板——不要同时改两个以上
2. 评估前冻结评测集：评测集不能在调参过程中修改，否则等于过拟合测试集
3. 保留基线：每次改动前记录当前版本的指标，改完后对比
4. 关注回归：某个指标提升但另一个下降时，要分析 trade-off 是否合理
5. 接入 CI：核心指标跌破阈值（如 Faithfulness 下降 5%）时阻断发布

第八章 生产工程化：从 Demo 到上线

本章目标：将一个能跑的 RAG Demo 硬化为可监控、可降级、可扩展的生产系统。

说明：[production.py](../ragbackend/production.py) 为参考架构（缓存、熔断、ObservableRAGPipeline 等），默认 python -m ragbackend.main ask 未启用 ObservableRAGPipeline；现行直连模式仅通过 retriever.timing / generator.timing 与 stdout 打印耗时。

8.1 延迟分析与优化

RAG 系统的端到端延迟可以分解为：

TTFT (Time to First Token) = 查询向量化 + 检索延迟 + LLM Prefill
Total Latency = TTFT + 生成耗时（逐 token）

各阶段典型耗时参考：

关键优化策略：

import functools
import hashlib
import time


class QueryCache:
"""查询缓存：相同查询在短时间内直接返回缓存结果。

  生产系统中，大量查询是重复或相似的（如热门 FAQ）。
  """

def __init__(self, ttl_seconds: int = 300, max_size: int = 1000):
self.ttl = ttl_seconds
self.max_size = max_size
self.cache: dict[str, tuple[float, any]] = {}  # key → (timestamp, result)

def get(self, query: str) -> any | None:
"""查询缓存，未命中或过期返回 None。"""
    key = self._make_key(query)
if key in self.cache:
      ts, result = self.cache[key]
if time.time() - ts < self.ttl:
return result
else:
del self.cache[key]  # 过期删除
return None

def put(self, query: str, result: any) -> None:
"""写入缓存。"""
# LRU 淘汰
if len(self.cache) >= self.max_size:
      oldest_key = min(self.cache, key=lambda k: self.cache[k][0])
del self.cache[oldest_key]

    key = self._make_key(query)
self.cache[key] = (time.time(), result)

def _make_key(self, query: str) -> str:
return hashlib.md5(query.strip().lower().encode()).hexdigest()


def timing_decorator(name: str):
"""分阶段耗时追踪装饰器。"""
def decorator(func):
    @functools.wraps(func)
def wrapper(*args, **kwargs):
      start = time.time()
      result = func(*args, **kwargs)
      elapsed_ms = (time.time() - start) * 1000
# 可以接入 metrics 系统（Prometheus、DataDog 等）
print(f"[Timing] {name}: {elapsed_ms:.1f}ms")
return result
return wrapper
return decorator


# 使用示例
@timing_decorator("vector_search")
def cached_vector_search(query: str, cache: QueryCache, vector_store, embed_fn, top_k=10):
  cached = cache.get(f"vector:{query}")
if cached is not None:
return cached

  embedding = embed_fn([query])[0]
  results = vector_store.search(embedding, top_k=top_k)

  cache.put(f"vector:{query}", results)
return results

8.2 降级策略矩阵

生产系统必须为每个组件设计降级方案：

class DegradationManager:
"""降级管理器：集中管理各组件的降级策略。"""

def __init__(self):
self.component_status = {}  # 组件名 → 状态
self.failure_counts = {}  # 组件名 → 连续失败次数
self.circuit_breaker_threshold = 3  # 连续失败 N 次后触发熔断

def record_success(self, component: str):
"""记录组件成功。"""
self.failure_counts[component] = 0
self.component_status[component] = "healthy"

def record_failure(self, component: str):
"""记录组件失败。"""
self.failure_counts[component] = self.failure_counts.get(component, 0) + 1
if self.failure_counts[component] >= self.circuit_breaker_threshold:
self.component_status[component] = "circuit_open"
print(f"[Circuit Breaker] {component} 熔断！连续失败 "
f"{self.failure_counts[component]} 次")

def is_available(self, component: str) -> bool:
"""检查组件是否可用。"""
return self.component_status.get(component, "healthy") != "circuit_open"

def reset(self, component: str):
"""手动重置熔断器。"""
self.failure_counts[component] = 0
self.component_status[component] = "healthy"


# 降级策略矩阵
DEGRADATION_MATRIX = {
"vector_search": {
"failure": "使用 BM25 单路结果",
"timeout": "使用已返回的部分结果",
"circuit_open": "临时禁用向量检索，仅使用 BM25",
  },
"bm25_search": {
"failure": "使用向量单路结果",
"timeout": "使用已返回的部分结果",
"circuit_open": "临时禁用 BM25，仅使用向量检索",
  },
"reranker": {
"failure": "使用 RRF 排序结果",
"timeout": "使用 RRF 排序结果",
"circuit_open": "跳过精排阶段",
  },
"llm_generation": {
"failure": "返回固定话术：'系统繁忙，请稍后重试'",
"timeout": "返回已生成的部分文本 + '（回答被截断）'",
"circuit_open": "切换到备用模型或返回缓存答案",
  },
"embedding_api": {
"failure": "使用本地轻量 Embedding 模型",
"timeout": "重试一次后降级",
"circuit_open": "切换到备用 Embedding 服务",
  },
}

8.3 可观测性：分阶段耗时追踪

现行 ragbackend 已在检索器与生成器中内置 timing 字典；cmd_ask 打印检索 + LLM 合计耗时。与 [production.py](../ragbackend/production.py) 中 RequestTrace 字段对照：

Prometheus / Grafana 接入为演进目标；当前以 stdout 与返回 dict 为主。

import time
from contextlib import contextmanager


class RequestTrace:
"""请求级追踪：记录一次 RAG 请求中各阶段的耗时和关键数据。"""

def __init__(self, request_id: str = ""):
self.request_id = request_id or str(int(time.time() * 1000))
self.spans: dict[str, float] = {}
self.metadata: dict = {}
self._start_times: dict[str, float] = {}

  @contextmanager
def span(self, name: str):
"""上下文管理器：自动记录某个阶段的耗时。

    Usage:
      with trace.span("vector_search"):
        results = vector_store.search(...)
    """
    start = time.time()
try:
yield
finally:
      elapsed_ms = (time.time() - start) * 1000
self.spans[name] = elapsed_ms

def set_metadata(self, key: str, value):
"""记录元数据（如检索数量、模型选择等）。"""
self.metadata[key] = value

def summary(self) -> dict:
"""输出追踪摘要。"""
return {
"request_id": self.request_id,
"spans_ms": self.spans,
"metadata": self.metadata,
"total_ms": sum(self.spans.values()),
    }

def log(self):
"""打印追踪日志。"""
    s = self.summary()
    parts = [f"request={s['request_id']}"]
for name, ms in s["spans_ms"].items():
      parts.append(f"{name}={ms:.0f}ms")
    parts.append(f"total={s['total_ms']:.0f}ms")
print(f"[Trace] {' | '.join(parts)}")


# 在 RAG Pipeline 中使用
class ObservableRAGPipeline(RAGPipeline):
"""可观测的 RAG Pipeline：每个阶段自动记录耗时。"""

def ask(self, query: str) -> dict:
    trace = RequestTrace()

with trace.span("retrieve"):
with trace.span("dense_search"):
# ... 稠密检索
pass
with trace.span("sparse_search"):
# ... 稀疏检索
pass
      contexts = self.retriever.retrieve(query)
      trace.set_metadata("contexts_count", len(contexts))

with trace.span("budget_filter"):
      contexts = budget_contexts(contexts, self.max_context_tokens)
      trace.set_metadata("contexts_after_budget", len(contexts))

with trace.span("generate"):
      result = self.generator.generate(query, contexts)

    trace.set_metadata("refused", result.get("refused", False))
    trace.set_metadata("answer_length", len(result.get("answer", "")))

    result["trace"] = trace.summary()
    trace.log()

return result

8.4 多租户与权限隔离

在企业场景中，不同部门/团队的知识库内容应该互相隔离。检索阶段必须按调用者身份过滤。

class TenantAwareRetriever:
"""租户感知的检索器：在检索阶段按租户过滤。

  关键原则：权限过滤必须在检索阶段完成，
  绝不能把越权文档送进 LLM 的上下文。
  """

def __init__(self, retriever: HybridRetriever):
self.retriever = retriever

def retrieve_for_tenant(
    self,
    query: str,
    tenant_id: str,
    user_permissions: list[str] | None = None,
) -> list[dict]:
"""为指定租户检索，自动过滤无权限文档。

    Args:
      query: 用户查询
      tenant_id: 租户 ID
      user_permissions: 用户有权限访问的文件列表
    """
# 在向量检索中添加元数据过滤
    where_filter = {"tenant_id": tenant_id}

# 如果有文件级权限控制
if user_permissions is not None:
# Chroma 的 $in 过滤
      where_filter["$and"] = [
        {"tenant_id": tenant_id},
        {"file_id": {"$in": user_permissions}},
      ]

# 临时修改 vector_store 的 search 方法，添加过滤条件
# （实际实现中应在 VectorStore.search 中直接支持 where 参数）
    original_search = self.retriever.vector_store.search

def filtered_search(embedding, top_k, where=None):
return original_search(embedding, top_k, where=where_filter)

self.retriever.vector_store.search = filtered_search

try:
      results = self.retriever.retrieve(query)
finally:
# 恢复原始方法
self.retriever.vector_store.search = original_search

return results

8.5 增量更新策略

class IncrementalIngestion:
"""增量入库：只处理新增/修改的文档，减少重建开销。"""

def __init__(self, dual_manager: DualIndexManager):
self.dual_manager = dual_manager
self.file_registry: dict[str, dict] = {}  # file_id → {hash, timestamp}

def ingest_if_changed(self, file_path: str) -> bool:
"""只在文件内容变化时才重新入库。

    Returns:
      True 如果文件被重新入库，False 如果无变化
    """
import hashlib

# 计算文件内容哈希
with open(file_path, "rb") as f:
      content_hash = hashlib.md5(f.read()).hexdigest()

    file_id = os.path.basename(file_path)

# 检查是否有变化
if file_id in self.file_registry:
if self.file_registry[file_id]["hash"] == content_hash:
return False  # 无变化，跳过

# 删除旧版本
if file_id in self.file_registry:
self.dual_manager.delete_file(self.file_registry[file_id]["chunk_id_prefix"])

# 处理并入库
    processor = DocumentProcessor(chunk_size=500)
    chunks = processor.process_file(file_path)
    chunk_dicts = [
      {"chunk_id": c.chunk_id, "content": c.content, "metadata": c.metadata}
for c in chunks
    ]

self.dual_manager.ingest(chunk_dicts, lambda texts: get_embeddings(texts))

# 更新注册表
self.file_registry[file_id] = {
"hash": content_hash,
"chunk_id_prefix": chunks[0].file_id if chunks else file_id,
"timestamp": time.time(),
"chunk_count": len(chunks),
    }

return True

第九章 总结与演进路线

9.1 RAG 落地路线图

9.2 各阶段目标与里程碑

9.3 十条核心工程原则

1. 切分优先于模型：再好的 Embedding 也救不了糟糕的切分
2. 混合优于单路：BM25 + 向量 + RRF 覆盖 90% 的场景
3. 降级优于报错：每个组件都要有降级方案
4. 拒答优于瞎答：宁可说「不知道」也不要编造
5. 测量优于直觉：用 RAGAS 数据说话，不要靠体感
6. 一次改一个变量：A/B 实验的基本纪律
7. 权限在检索侧：不要把越权文档送进 LLM
8. 缓存热门查询：FAQ 场景下缓存能省 80% 的 API 调用
9. 可观测是基础：没有分阶段耗时数据就无法优化
10. 渐进式演进：先 MVP 验证价值，再逐步升级架构

9.4 技术选型速查表

附录 A：环境搭建与依赖安装

# 推荐：一次性安装 ragbackend 全部依赖
pip install -r ragbackend/requirements.txt

# 配置 API Key（或在 ragbackend/.env 中填写）
export DASHSCOPE_API_KEY="your-api-key"
export DEEPSEEK_API_KEY="your-deepseek-key"   # 仅当 llm_model=deepseek-* 时需要

常用环境变量（[config.py](../ragbackend/config.py)）：