https://python.langchain.com.cn/docs/expression_language/cookbook/retrieval

一、文档核心目标

原文只讲一件事：用 LCEL 构建 RAG 链——让 LLM 基于“外部检索到的文本”（而非仅自身训练数据）生成回答，分3个递进场景：

1. 基础 RAG 链（无对话历史，仅单次检索）；
2. 对话式 RAG 链（支持对话历史，处理指代问题）；
3. 带内存+返回源文档的 RAG 链（自动存历史，显示回答来源）。

二、场景1：基础 RAG 链（原文前半段代码）

步骤1：安装依赖（原文第一行代码）

作用：安装 RAG 必需的工具（向量库、嵌入、LLM 等），直接复制运行即可。

%pip install --upgrade --quiet  langchain langchain-openai faiss-cpu tiktoken

步骤2：导入库+初始化核心组件（原文代码逐行对应）

这步是“搭基础零件”，每个组件的作用和数据格式都明确标注：

# 导入需要的工具：itemgetter（提取字典键值）、FAISS（向量库）、输出解析器等
from operator import itemgetter
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 1. 初始化向量库：存储文本的“向量”，用于后续检索
# 输入：文本列表（原文仅1条："harrison worked at kensho"）、嵌入工具（OpenAIEmbeddings）
# 作用：把文本转成向量，存到FAISS向量库中
vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"],  # 要检索的“外部文本”（核心数据源）
    embedding=OpenAIEmbeddings()    # 把文本转向量的工具
)

# 2. 初始化检索器：从向量库中“找和问题相似的文本”
# 作用：接收问题，返回向量库中最相似的文本（默认返回1条）
retriever = vectorstore.as_retriever()

# 3. 初始化提示模板：强制LLM“仅基于检索到的context回答”
# 占位符：{context}（检索到的文本）、{question}（用户问题）
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)  # 生成可调用的提示模板

# 4. 初始化LLM：用于生成回答（原文用ChatOpenAI）
model = ChatOpenAI()

步骤3：构建+测试基础 RAG 链（原文核心链代码）

3.1 构建链：明确数据流向

# 链结构（数据流向：输入→处理→输出）：
# 1. 输入处理：拆分出2个键——
#    - "context"：用retriever检索与问题相似的文本（自动关联问题）
#    - "question"：用RunnablePassthrough()直接传递用户输入的问题
# 2. 填充提示：把context和question传入prompt模板
# 3. 生成回答：model基于填充后的提示生成文本
# 4. 解析输出：StrOutputParser()把model的AIMessage转成纯字符串
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

3.2 测试链：输入→输出完全对应原文

# 输入：用户问题“where did harrison work?”（询问harrison在哪工作）
# 运行链：retriever会从向量库中检索到"harrison worked at kensho"作为context
result = chain.invoke("where did harrison work?")

# 输出：原文明确给出的结果（LLM仅基于context回答）
print(result)  # 输出：'Harrison worked at Kensho.'

步骤4：基础 RAG 链扩展（多参数输入：支持指定语言）

原文还提供了“按指定语言回答”的扩展案例，核心是给链加“language”参数：

# 1. 修改提示模板：新增{language}占位符，指定回答语言
template = """Answer the question based only on the following context:
{context}
Question: {question}
Answer in the following language: {language}
"""
prompt = ChatPromptTemplate.from_template(template)

# 2. 构建链：用itemgetter提取输入字典中的"question""language"键
# 数据流向：输入字典→提取3个键（context/question/language）→填充prompt→生成回答
chain = (
    {
        "context": itemgetter("question") | retriever,  # 用question去检索context
        "question": itemgetter("question"),            # 提取输入中的"question"
        "language": itemgetter("language"),            # 提取输入中的"language"
    }
    | prompt
    | model
    | StrOutputParser()
)

# 测试：输入字典（含question和language）
# 原文输入：{"question": "where did harrison work", "language": "italian"}（意大利语）
result = chain.invoke({"question": "where did harrison work", "language": "italian"})

# 原文输出：意大利语回答
print(result)  # 输出：'Harrison ha lavorato a Kensho.'

三、场景2：对话式 RAG 链（支持对话历史，原文“Conversational Retrieval Chain”部分）

解决“指代问题”（如用户问“where did he work?”，“he”需结合历史才知是“Harrison”），核心是“把‘历史+跟进问题’改写成独立问题”。

步骤1：导入新增库+初始化关键组件（原文代码）

# 导入处理对话历史和文档合并的工具
from langchain.schema import format_document
from langchain_core.messages import AIMessage, HumanMessage, get_buffer_string
from langchain_core.runnables import RunnableParallel
from langchain.prompts.prompt import PromptTemplate

# 1. 独立问题生成提示：把“对话历史+跟进问题”改写成无指代的独立问题
# 占位符：{chat_history}（对话历史）、{question}（跟进问题）
_template = """Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question, in its original language.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(_template)

# 2. 回答提示：和基础RAG一致，仅基于context回答
ANSWER_PROMPT = ChatPromptTemplate.from_template("""Answer the question based only on the following context:
{context}
Question: {question}
""")

# 3. 文档合并函数：把检索到的多个Document对象（含page_content）合并成纯文本
# 作用：检索器返回的是Document列表，需转成字符串才能传入prompt
def _combine_documents(
    docs, 
    document_prompt=PromptTemplate.from_template("{page_content}"),  # 提取Document的page_content
    document_separator="\n\n"  # 多个文档用空行分隔
):
    doc_strings = [format_document(doc, document_prompt) for doc in docs]  # 逐个提取内容
    return document_separator.join(doc_strings)  # 合并成单个字符串

步骤2：构建对话式 RAG 链（分2步：生成独立问题→检索+回答）

2.1 第一步：生成独立问题（解决指代）

# _inputs链：输入“question+chat_history”→输出“standalone_question”（无指代的独立问题）
_inputs = RunnableParallel(
    standalone_question=RunnablePassthrough.assign(
        # 把chat_history（Message列表）转成字符串（如“Human: ...\nAI: ...”）
        chat_history=lambda x: get_buffer_string(x["chat_history"])
    )
    | CONDENSE_QUESTION_PROMPT  # 传入“chat_history+question”到提示
    | ChatOpenAI(temperature=0)  # 用LLM生成独立问题（temperature=0确保结果稳定）
    | StrOutputParser()  # 转成纯字符串
)

2.2 第二步：检索+生成回答

# _context链：用独立问题检索→合并文档→准备回答的输入
_context = {
    "context": itemgetter("standalone_question") | retriever | _combine_documents,  # 独立问题→检索→合并文档
    "question": lambda x: x["standalone_question"]  # 用独立问题作为最终提问
}

# 完整链：生成独立问题→检索+回答→LLM生成
conversational_qa_chain = _inputs | _context | ANSWER_PROMPT | ChatOpenAI()

步骤3：测试对话式链（2个案例，完全对应原文）

案例1：无对话历史（直接提问）

# 输入：question=“where did harrison work?”，chat_history=空列表
result1 = conversational_qa_chain.invoke(
    {
        "question": "where did harrison work?",
        "chat_history": [],
    }
)

# 原文输出：AIMessage对象（content是回答）
print(result1)  # 输出：AIMessage(content='Harrison was employed at Kensho.')

案例2：有对话历史（处理指代“he”）

# 输入：question=“where did he work?”（he指代Harrison），chat_history含历史对话
result2 = conversational_qa_chain.invoke(
    {
        "question": "where did he work?",
        "chat_history": [
            HumanMessage(content="Who wrote this notebook?"),  # 历史：用户问“谁写的笔记本”
            AIMessage(content="Harrison")                     # 历史：AI答“Harrison”
        ],
    }
)

# 原文输出：AI正确识别“he”是Harrison
print(result2)  # 输出：AIMessage(content='Harrison worked at Kensho.')

四、场景3：带内存+返回源文档的 RAG 链（原文最后部分）

新增2个功能：① 用ConversationBufferMemory自动存对话历史（不用每次传chat_history）；② 返回检索到的源文档（docs），显示回答来源。

步骤1：初始化内存组件（原文代码）

from langchain.memory import ConversationBufferMemory

# 内存：存储对话历史（Message对象），需指定input_key/output_key匹配链的输入输出
memory = ConversationBufferMemory(
    return_messages=True,  # 存储Message对象（而非纯文本）
    output_key="answer",   # 内存的“输出”对应链的“answer”键
    input_key="question"   # 内存的“输入”对应链的“question”键
)

步骤2：构建完整链（分4步，原文逐段对应）

# 1. 第一步：加载内存→从内存中获取chat_history，添加到输入字典
loaded_memory = RunnablePassthrough.assign(
    chat_history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"),
    # 逻辑：memory.load_memory_variables({})→返回{"history": [Message列表]}→提取history
)

# 2. 第二步：生成独立问题（复用场景2的逻辑）
standalone_question = {
    "standalone_question": {
        "question": lambda x: x["question"],
        "chat_history": lambda x: get_buffer_string(x["chat_history"])
    }
    | CONDENSE_QUESTION_PROMPT
    | ChatOpenAI(temperature=0)
    | StrOutputParser(),
}

# 3. 第三步：检索源文档→返回Document列表（不合并，后续要显示来源）
retrieved_documents = {
    "docs": itemgetter("standalone_question") | retriever,  # 独立问题→检索源文档
    "question": lambda x: x["standalone_question"],        # 传递独立问题
}

# 4. 第四步：生成回答+传递源文档（最终输出“answer+docs”）
final_inputs = {
    "context": lambda x: _combine_documents(x["docs"]),  # 合并文档为context（用于回答）
    "question": itemgetter("question"),
}
answer = {
    "answer": final_inputs | ANSWER_PROMPT | ChatOpenAI(),  # 生成回答
    "docs": itemgetter("docs"),                              # 传递源文档（不处理，直接返回）
}

# 完整链：加载内存→生成独立问题→检索文档→生成回答+返回文档
final_chain = loaded_memory | standalone_question | retrieved_documents | answer

步骤3：测试链+手动保存内存（原文强调“内存不自动保存”）

案例1：第一次提问（无历史）

# 输入：仅question（不用传chat_history，内存自动加载空历史）
inputs1 = {"question": "where did harrison work?"}
result1 = final_chain.invoke(inputs1)

# 原文输出：字典（含answer和docs）
print(result1)
# 输出：
# {'answer': AIMessage(content='Harrison was employed at Kensho.'),
#  'docs': [Document(page_content='harrison worked at kensho')]}  # 源文档（回答来源）

# 原文关键注释：内存不自动保存，需手动调用save_context
memory.save_context(inputs1, {"answer": result1["answer"].content})

# 验证内存：内存中已存储第一次对话
print(memory.load_memory_variables({}))
# 输出：{'history': [HumanMessage(content='where did harrison work?'),
#                  AIMessage(content='Harrison was employed at Kensho.')]}

案例2：第二次提问（有历史，指代“he”）

# 输入：question=“but where did he really work?”（he指代Harrison）
inputs2 = {"question": "but where did he really work?"}
result2 = final_chain.invoke(inputs2)

# 原文输出：AI正确识别“he”，且返回源文档
print(result2)
# 输出：
# {'answer': AIMessage(content='Harrison actually worked at Kensho.'),
#  'docs': [Document(page_content='harrison worked at kensho')]}

五、关键补充：原文没丢任何内容，这些细节要注意

1. itemgetter的作用：仅用于“从输入字典中提取指定键的值”，比如itemgetter("question")就是“拿输入里‘question’对应的值”，无其他复杂逻辑；
2. 内存手动保存：原文反复注释“memory does not save automatically”，必须在每次invoke后调用memory.save_context(inputs, {"answer": ...})，否则历史会丢；
3. 源文档格式：docs是Document对象列表，每个对象的page_content字段是实际检索到的文本（原文仅1条，实际场景会有多条）。