LangChain学习日志: RAG

尝试使用LangChain来实现RAG（检索增强生成）功能。

RAG简介

本文将简单介绍RAG（Retrieval-Augmented Generation，检索增强生成）在LangChain中的实现

RAG是一种将检索技术与生成技术相结合的架构，当用户提出问题时，RAG 系统首先从外部知识库中检索出与问题相关的文档片段，然后将这些片段与用户的问题一起喂给大模型，以基于参考资料给出更准确、更实时的回答。

在大语言模型（LLM）的视角下，人类知识被分为两种形态：

参数化记忆（Parametric Memory）：指模型在预训练阶段通过海量文本学习到的、固化在神经元权重（Parameters）中的知识。
非参数化记忆（Non-parametric Memory）：指模型之外的结构化或非结构化知识库（如维基百科、企业私有文档）。

RAG的本质，就是将这两种记忆结合，通过一个微分访问机制（Differentiable Access Mechanism），让模型在生成回答前，先去“翻阅”外部的非参数化记忆。

RAG 的数学形式化定义

在原始论文中（即来自的 NeurIPS2020 的著名论文Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks），RAG 被定义为一个隐变量模型（Latent Variable Model）。

假设用户输入为 $x$ ，系统检索到的文档片段为 $z$ （作为隐变量），最终生成的答案为 $y$ 。RAG 的核心目标是计算在给定输入 $x$ 的情况下，生成输出 $y$ 的条件概率 $p(y|x)$ 。这一过程由检索组件（Retriever）和生成组件（Generator）组成。

Retriever

检索器的任务是计算给定输入 $x$ 时，文档 $z$ 的被选中的概率为：

p_\eta(z|x) \propto \exp(d(z)^\top q(x))

其中：

$q(x)$ ：是由 BERT 构建的查询编码器（Query Encoder）生成的稠密向量。
$d(z)$ ：是由 BERT 构建的文档编码器（Document Encoder）生成的文档向量。
此公式本质上是计算两个向量的点积相似度。

Generator

生成器的任务是根据输入 $x$ 和检索到的上下文 $z$ ，计算生成 token $y_i$ 的概率：

p_\theta(y_i | x, z, y_{1:i-1})

这通常由一个预训练的 Seq2Seq 模型（如论文中的 BART）来完成。

两种 RAG 建模范式

论文提出了两种处理隐变量 $z$ 的方式，这直接决定了 RAG 系统的运行逻辑。

RAG-Sequence：序列化检索增强

该模型假设一整个序列的生成只依赖于同一篇文档 $z$ 。模型会先为每个检索到的文档计算完整的生成概率，最后进行加权求和。

p_{\text{RAG-Sequence}}(y|x) \approx \sum_{z \in \text{top-k}(p(\cdot|x))} p_\eta(z|x) \prod_i^N p_\theta(y_i|x, z, y_{1:i-1})

RAG-Token：Token 级检索增强

该模型更加灵活，允许在生成每一个词（Token）时，都参考不同的文档。这意味着最终生成的句子可能是由多篇文档共同支撑的。

p_{\text{RAG-Token}}(y|x) \approx \prod_i^N \sum_{z \in \text{top-k}(p(\cdot|x))} p_\eta(z|x) p_\theta(y_i|x, z, y_{1:i-1})

技术流程

基于上述原理，一个标准的 RAG 工作流可以被拆解为以下五个步骤，这也是后续章节我们将使用 LangChain 实现的核心流程：

索引（Indexing）：将非参数化记忆（文档）切块并转化为向量 $d(z)$ 。
检索（Retrieval）：根据用户查询 $q(x)$ ，在向量空间中寻找最相似的 $Top-K$ 个文档 $z$ 。
增强（Augmentation）：将 $x$ 与 $z$ 拼接，构建出内容丰富的 Prompt。
生成（Generation）：LLM 基于增强后的信息，计算 $p(y|x, z)$ 并输出答案。
引用（Citation）：由于我们可以追踪到 $z$ ，系统可以给出答案的出处，显著提升可信度。

大概了解RAG之后，我们来构建一个最最简单的基于LangChain的RAG应用的例子⬇️

Infra

在这个例子中，我们使用Python + Postgresql进行构建。

首先，需要一个安装了 pgvector 扩展的 PostgreSQL 实例。先安装pgvector，以我的mac为例。

1	brew install pgvector

然后对我们的目标数据库添加拓展：

1	CREATE EXTENSION IF NOT EXISTS vector;

我们需要安装 LangChain 的核心包、PostgreSQL 驱动以及阿里通义的 SDK。

1 2	pip install langchain langchain-community langchain-postgres pip install psycopg2-binary dashscope

外加一些文档加载库

1	pip install pypdf docx2txt

数据预处理入库

这一步实现了 RAG 原理中的非参数化记忆（Non-parametric Memory）构建过程。我们需要将原始文档转化为检索器（Retriever）可以识别的向量 $d(z)$ 。

LangChain 提供了丰富的 document_loaders，以最常见的txt、pdf、word为例：

txt

from langchain_community.document_loaders import TextLoader

loader = TextLoader("./data/example.txt", encoding="utf-8")
txt_docs = loader.load()

pdf

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("./data/example.pdf")
pdf_docs = loader.load()

word

from langchain_community.document_loaders import Docx2txtLoader

loader = Docx2txtLoader("./data/example.docx")
word_docs = loader.load()

为了演示，我在网上找了本精美的家用菜谱大全pdf：

看饿了。

对于中文 RAG，我们使用BGE (BAAI General Embedding)开源模型，先下载：

1	pip install sentence-transformers

使用 langchain 加载pdf文档，切分，并存入数据库。

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_postgres.vectorstores import PGVector

FILE_PATH = "/Users/fangtianyao/Downloads/家庭实用菜谱大全.pdf"
CONNECTION_STRING = "postgresql+psycopg2://postgres:postgres@localhost:5432/rag"
COLLECTION_NAME = "recipe_collection_bge"

def main():
    # 1. 加载本地 Embedding 模型 (BGE-small-zh)
    embeddings = HuggingFaceEmbeddings(
        model_name="BAAI/bge-small-zh-v1.5",
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': True}
    )

    # 2. 加载 PDF
    loader = PyPDFLoader(FILE_PATH)
    documents = loader.load()

    # 3. 切分文档
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=50)
    chunks = text_splitter.split_documents(documents)

    # 4. 存入 PostgreSQL
    vector_store = PGVector.from_documents(
        embedding=embeddings,
        documents=chunks,
        collection_name=COLLECTION_NAME,
        connection=CONNECTION_STRING,
        use_jsonb=True,
    )

if __name__ == "__main__":
    main()

LangChain会自动创建 langchain_pg_embedding 和 langchain_pg_collection 两个数据库，其中 embedding 字段会被定义为 vector 类型。

简单的RAG对话

我们参考文档https://docs.langchain.com/oss/python/integrations/chat/tongyi ，基于阿里云的 ChatTongyi 构建RAG服务。可以看到，阿里云的模型可选择性很高。

申请一个阿里云百炼的 API-Key ，这里就不再赘述。我们先测试一下LangChain调用阿里云API的能力：

import os
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 设置 API Key
os.environ["DASHSCOPE_API_KEY"] = "********"

# 初始化模型
llm = ChatTongyi(model_name="qwen-max")

# 简单的测试链路
prompt = ChatPromptTemplate.from_template("讲个关于{topic}的笑话")
chain = prompt | llm | StrOutputParser()

print(chain.invoke({"topic": "程序员"}))

输出如下：

当然可以，这里有一个程序员可能会会心一笑的笑话：

为什么程序员不喜欢在户外工作？
因为那里有太多的 bugs（虫子/错误）！

这里的“bugs”一词双关，既指自然界中的昆虫，也指编程中遇到的程序错误。希望这个小笑话能给你带来一丝轻松和快乐！

好无聊的笑话……，不过至少证明我们测试成功了！那么现在，我们可以引入之前构建的向量数据库。导入包，连接数据库，并初始化和入库时一致的 Embedding 模型：

import os

from langchain_community.chat_models import ChatTongyi
from langchain_postgres.vectorstores import PGVector
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

os.environ["DASHSCOPE_API_KEY"] = "******"

# 数据库配置
CONNECTION_STRING = "postgresql+psycopg2://postgres:postgres@localhost:5432/rag"
COLLECTION_NAME = "recipe_collection_bge"

# 初始化 Embedding 模型
print("正在加载本地 BGE 模型...")
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 连接现有的向量数据库
print(f"正在连接数据库集合: {COLLECTION_NAME}...")
vector_store = PGVector(
    connection=CONNECTION_STRING,
    embeddings=embeddings,
    collection_name=COLLECTION_NAME,
    use_jsonb=True,
)

# 创建检索器 (Retriever)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

其中，我们通过 retriever = vector_store.as_retriever(search_kwargs={"k": 3}) 创建了检索器，而参数 "k": 3 代表了表示每次检索最相似的 3 条记录。随后，定义 LLM 和 Prompt：

# 定义 LLM 和 Prompt (LCEL 风格) 
# 初始化通义千问 (Qwen-Max)
llm = ChatTongyi(model_name="qwen-max")

# 编写提示词模板
template = """你是一个专业的智能大厨。请基于下面的【参考资料】回答用户的问题。
如果参考资料里没有提到的内容，请诚实地说“资料中未提及”，不要编造。

【参考资料】：
{context}

【用户问题】：
{question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 辅助函数：将检索到的文档列表合并成一个字符串
def format_docs(docs):
    return "\n\n".join([doc.page_content for doc in docs])

构建 RAG 链：

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

这是最重要的一步。上述代码使用的是 LCEL（LangChain Expression Language，表达式语言），| 符号pipeline中每一道工序的间隔。我们把这个链拆解成三个阶段：

STEP1 数据并行处理：

1	{"context": retriever \| format_docs, "question": RunnablePassthrough()}

这一部分是一个字典。当你调用 rag_chain.invoke("如何做红烧肉？") 时，这个初始字符串会同时喂给两个分支：

question 分支：
- RunnablePassthrough() 不对输入做任何处理，直接把“如何做红烧肉？”原样传给 question 字段。
context 分支（这是一个子链）：
- 第一步：retriever 接收到问题，去 PostgreSQL 向量库里寻找最相似的 $K$ 个文档片段（此时输出的是一个 Document 对象列表）。
- 第二步：通过 | 传给 format_docs 函数。这个函数把多个文档对象里的文字提炼出来，拼接成一个大字符串。
- 最终 context 字段得到了库里的菜谱知识。

STEP2 推理

1	... \| prompt \| llm

注入 Prompt (| prompt)：
- 此时，前面的输出是一个字典：{"context": "...", "question": "如何做红烧肉？"}。ChatPromptTemplate 会自动寻找对应的变量，把它们填入你定义的模板中，生成一段发给 AI 的完整指令。
调用大模型 (| llm)：
- 将填充好的指令传给通义千问 ChatTongyi。

STEP3 输出解析：

1	... \| StrOutputParser()

StrOutputParser() 会从LLM返回的 AIMessage 对象中提取 content 属性。

最后，只要调用 rag_chain.stream()，就能获得流式输出的回答。给出完整的代码：

import os
from langchain_community.chat_models import ChatTongyi
from langchain_postgres.vectorstores import PGVector
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

os.environ["DASHSCOPE_API_KEY"] = "sk-873463eacac84219b830314fb5f10992"

# 数据库配置
CONNECTION_STRING = "postgresql+psycopg2://postgres:postgres@localhost:5432/rag"
COLLECTION_NAME = "recipe_collection_bge"

# 初始化 Embedding 模型
print("正在加载本地 BGE 模型...")
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 连接现有的向量数据库
print(f"正在连接数据库集合: {COLLECTION_NAME}...")
vector_store = PGVector(
    connection=CONNECTION_STRING,
    embeddings=embeddings,
    collection_name=COLLECTION_NAME,
    use_jsonb=True,
)

# 创建检索器 (Retriever)
# k=3 表示每次检索最相似的 3 条菜谱
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# 定义 LLM 和 Prompt (LCEL 风格) 
# 初始化通义千问 (Qwen-Max)
llm = ChatTongyi(model_name="qwen-max")

# 编写提示词模板
template = """你是一个专业的智能大厨。请基于下面的【参考资料】回答用户的问题。
如果参考资料里没有提到的内容，请诚实地说“资料中未提及”，不要编造。

【参考资料】：
{context}

【用户问题】：
{question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 辅助函数：将检索到的文档列表合并成一个字符串
def format_docs(docs):
    return "\n\n".join([doc.page_content for doc in docs])

# 构建 RAG 链 (The Chain) 
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 执行对话
if __name__ == "__main__":
    
    while True:
        question = input("\n请问有什么可以帮您: ")
        if question.lower() in ["quit", "exit"]:
            break
        
        print("\n正在思考并检索数据库...", end="")
        
        try:
            # 使用流式输出 (Stream) 提升体验
            print("\n回答: ", end="")
            for chunk in rag_chain.stream(question):
                print(chunk, end="", flush=True)
            print("\n")
        except Exception as e:
            print(f"\n发生错误: {e}")

询问他菜谱中的菜“毛蛤拌菠菜”怎么做：

请问有什么可以帮您: 我想制作毛蛤拌菠菜

正在思考并检索数据库...
回答: 您想制作的是“毛蛤拌菠菜”。根据提供的参考资料，下面是具体的步骤：

### 材料
- 毛蛤·250克
- 菠菜·250克
- 生姜·1小块
- 大蒜·4瓣

### 调料
- 香油·2小匙
- 香醋·1小匙
- 精盐·1小匙
- 味精· 小匙

### 做法
1. **准备材料**：首先将毛蛤煮熟后取出肉备用；同时将生姜和大蒜清洗干净后切成末。
2. **处理菠菜**：将菠菜择洗干净，然后放入沸水中焯烫片刻。之后用凉水冲凉以去除多余的热量，并且沥干水分后切成段。
3. **调拌**：将处理好的毛蛤、切好的菠菜以及之前准备的调料（包括盐、香醋、味精、香油、姜末和蒜末）混合在一起充分搅拌均匀。
4. **完成**：最后将调拌好的食材装盘即可享用。

按照以上步骤操作，您就能做出一道味道鲜美、清淡爽口的毛蛤拌菠菜了。希望这对您有所帮助！

对比菜谱中的做法：

非常成功✌️