RAG 增强记忆

引言

检索增强生成（Retrieval-Augmented Generation, RAG）是当前 Agent 记忆系统最重要的技术范式。RAG 通过在生成前检索外部知识，让 LLM 能够访问其训练数据之外的信息，有效解决幻觉问题并实现知识的动态更新。

RAG 的动机

LLM 的固有局限：

1. 知识截止：训练数据有截止日期
2. 幻觉：对不确定的问题可能编造答案
3. 领域知识缺失：缺少特定组织/领域的私有知识
4. 不可追溯：无法指出信息来源

RAG 的解决思路：先检索，再生成。

Naive RAG 管线

graph TB
    subgraph 离线索引阶段
        D[文档集合] --> L[文档加载<br/>Document Loading]
        L --> S[文本分块<br/>Chunking]
        S --> E[向量嵌入<br/>Embedding]
        E --> I[(向量数据库<br/>Vector Store)]
    end

    subgraph 在线查询阶段
        Q[用户查询] --> QE[查询嵌入]
        QE --> R[检索<br/>Top-K]
        I --> R
        R --> C[上下文组装]
        Q --> C
        C --> G[LLM 生成]
        G --> A[回答]
    end

步骤 1：文档加载

from langchain.document_loaders import (
    PyPDFLoader, TextLoader, WebBaseLoader, 
    UnstructuredMarkdownLoader
)

# PDF
loader = PyPDFLoader("paper.pdf")
docs = loader.load()

# 网页
loader = WebBaseLoader("https://example.com/article")
docs = loader.load()

# 多种格式统一处理
from langchain.document_loaders import DirectoryLoader
loader = DirectoryLoader("./knowledge_base/", glob="**/*.md")
docs = loader.load()

步骤 2：文本分块

分块策略直接影响检索质量：

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # 每块最大 token 数
    chunk_overlap=50,     # 块间重叠
    separators=["\n\n", "\n", "。", ".", " ", ""],  # 分割优先级
    length_function=len,
)

chunks = splitter.split_documents(docs)

分块策略比较：

策略	优点	缺点
固定大小	简单一致	可能切断语义
递归分割	尊重文档结构	块大小不均匀
语义分块	语义完整	计算成本高
文档结构分块	保留标题/章节	依赖文档格式

步骤 3：嵌入与索引

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

步骤 4：检索

retriever = vectorstore.as_retriever(
    search_type="similarity",  # or "mmr" for diversity
    search_kwargs={"k": 5}
)

relevant_docs = retriever.get_relevant_documents("什么是 Agent 的记忆系统？")

步骤 5：生成

from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=retriever,
    return_source_documents=True,
)

result = qa_chain.invoke({"query": "什么是 Agent 的记忆系统？"})
print(result["result"])
print(result["source_documents"])  # 来源文档

Advanced RAG 技术

查询优化

查询重写（Query Rewriting）

def rewrite_query(original_query, llm):
    prompt = f"""请将以下用户查询改写为更适合语义搜索的形式。
    生成 3 个不同角度的查询变体。

    原始查询: {original_query}

    改写查询:"""
    return llm.generate(prompt)

# "怎么让 Agent 记住东西" -> 
# ["Agent 记忆系统的实现方法", "LLM 长期记忆存储技术", "对话历史持久化方案"]

HyDE（Hypothetical Document Embeddings）

先让 LLM 生成一个假设性答案，再用这个答案去检索：

def hyde_retrieval(query, llm, retriever):
    # 1. 生成假设性答案
    hypothetical_answer = llm.generate(
        f"请回答以下问题（即使你不确定）：{query}"
    )

    # 2. 用假设性答案的嵌入去检索
    results = retriever.get_relevant_documents(hypothetical_answer)
    return results

原理：假设性答案与真实文档在语义空间中更接近，比原始查询的检索效果更好。

检索优化

多路检索（Ensemble Retrieval）

from langchain.retrievers import EnsembleRetriever, BM25Retriever

# 关键词检索
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 向量检索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 集成
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

最大边际相关性（MMR）

在相关性和多样性之间取平衡：

\[\text{MMR} = \arg\max_{d_i \in R \setminus S} \left[ \lambda \cdot \text{sim}(d_i, q) - (1-\lambda) \cdot \max_{d_j \in S} \text{sim}(d_i, d_j) \right]\]

其中 \(\lambda\) 控制相关性与多样性的平衡。

重排序（Reranking）

检索后使用交叉编码器（Cross-Encoder）对候选结果精排：

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

def rerank(query, documents, top_k=3):
    pairs = [(query, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)

    # 按分数排序
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:top_k]]

Cohere Rerank API：

import cohere

co = cohere.Client("your-api-key")
results = co.rerank(
    query="Agent 记忆系统",
    documents=[doc.page_content for doc in candidates],
    top_n=3,
    model="rerank-english-v3.0"
)

高级 RAG 模式

Self-RAG（自反思 RAG）

graph TD
    Q[查询] --> D{需要检索吗？}
    D -->|是| R[检索文档]
    D -->|否| G1[直接生成]
    R --> E{文档相关吗？}
    E -->|是| G2[基于文档生成]
    E -->|否| R2[重新检索/改写查询]
    G2 --> V{回答有支撑吗？}
    V -->|是| O[输出]
    V -->|否| G2

Self-RAG（Asai et al., 2023）训练模型自主决定何时检索、评估检索质量、验证生成结果。

CRAG（Corrective RAG）

def corrective_rag(query, retriever, llm):
    docs = retriever.get_relevant_documents(query)

    # 评估检索质量
    relevance = llm.evaluate(f"以下文档是否与查询相关？\n查询: {query}\n文档: {docs}")

    if relevance == "correct":
        # 直接使用检索结果
        return generate(query, docs)
    elif relevance == "ambiguous":
        # 结合检索结果和网络搜索
        web_results = web_search(query)
        return generate(query, docs + web_results)
    else:  # incorrect
        # 完全依赖网络搜索
        web_results = web_search(query)
        return generate(query, web_results)

多跳 RAG（Multi-hop RAG）

对于需要多步推理的复杂问题：

def multi_hop_rag(query, retriever, llm, max_hops=3):
    context = []
    current_query = query

    for hop in range(max_hops):
        # 检索
        docs = retriever.get_relevant_documents(current_query)
        context.extend(docs)

        # 判断是否有足够信息回答
        can_answer = llm.evaluate(
            f"基于以下信息能否回答问题？\n问题: {query}\n信息: {context}"
        )

        if can_answer:
            break

        # 生成后续查询
        current_query = llm.generate(
            f"为了回答'{query}'，还需要什么信息？已知: {context}"
        )

    return llm.generate(f"基于以下信息回答问题。\n问题: {query}\n信息: {context}")

Graph RAG

结合知识图谱的结构化检索：

# 使用 LLM 从文本中提取实体和关系
def build_knowledge_graph(documents, llm):
    triples = []
    for doc in documents:
        extracted = llm.extract(
            f"从以下文本中提取(主语, 关系, 宾语)三元组：\n{doc}"
        )
        triples.extend(extracted)
    return triples

# 结合图检索和向量检索
def graph_rag(query, graph_store, vector_store, llm):
    # 提取查询中的实体
    entities = llm.extract_entities(query)

    # 图检索：获取实体的邻居关系
    graph_context = graph_store.get_neighbors(entities, depth=2)

    # 向量检索：获取语义相关的文档
    vector_context = vector_store.similarity_search(query, k=5)

    # 合并上下文生成回答
    return llm.generate(query, graph_context + vector_context)

RAG 评估

核心指标

检索质量

• Recall@K：前 K 个结果中包含正确答案的比例

\[\text{Recall@K} = \frac{|\text{检索到的相关文档} \cap \text{前K结果}|}{|\text{所有相关文档}|}\]

• MRR（Mean Reciprocal Rank）：正确答案首次出现的位置的倒数的平均值

\[\text{MRR} = \frac{1}{|Q|} \sum_{i=1}^{|Q|} \frac{1}{\text{rank}_i}\]

• NDCG（Normalized DCG）：考虑排序位置的检索质量指标

生成质量

• Faithfulness：生成的回答是否忠实于检索到的文档
• Answer Relevance：回答是否与问题相关
• Context Relevance：检索到的上下文是否与问题相关

RAGAS 评估框架

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision],
)
print(results)

生产级 RAG 架构

graph TB
    subgraph 数据管线
        DS[数据源] --> ETL[ETL Pipeline]
        ETL --> CP[分块处理]
        CP --> EM[嵌入生成]
        EM --> VS[(向量数据库)]
        ETL --> MD[(元数据存储)]
    end

    subgraph 查询管线
        UQ[用户查询] --> QP[查询处理<br/>重写/扩展]
        QP --> HR[混合检索<br/>向量+关键词]
        VS --> HR
        HR --> RR[重排序]
        RR --> CTX[上下文组装]
        CTX --> LLM[LLM 生成]
        LLM --> PP[后处理<br/>引用/验证]
        PP --> OUT[最终回答]
    end

    subgraph 反馈循环
        OUT --> FB[用户反馈]
        FB --> AN[分析优化]
        AN --> DS
    end

实践建议

RAG 优化清单

• [ ] 选择合适的分块大小（通常 256-512 tokens）
• [ ] 使用递归分割保留文档结构
• [ ] 实现混合搜索（向量 + BM25）
• [ ] 添加重排序步骤提升精度
• [ ] 对查询进行重写/扩展
• [ ] 添加元数据过滤缩小搜索范围
• [ ] 实现评估管线持续监控质量
• [ ] 处理多模态内容（表格、图片）

常见陷阱

1. 分块太大或太小：太大稀释信号，太小丢失上下文
2. 忽略元数据：时间戳、来源、类型等元数据是重要的过滤维度
3. 不做重排序：初始检索结果通常需要精排
4. 不评估：没有评估就无法改进

延伸阅读

• API 编排与工具选择 - RAG 作为 Agent 工具的编排
• Lewis, P., et al. (2020). "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks"
• Asai, A., et al. (2023). "Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection"
• Gao, Y., et al. (2024). "Retrieval-Augmented Generation for Large Language Models: A Survey"

---

评论