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

LangChain RAG 三大场景代码整合版（带详细注释）

场景1：基础RAG链（无对话历史，仅单次检索）

# 1. 安装依赖（首次运行需执行，后续可跳过）
# 作用：安装RAG必需的工具库：
# - langchain：核心链功能
# - langchain-openai：OpenAI的模型和嵌入工具
# - faiss-cpu：轻量级向量数据库（存储文本向量，用于检索）
# - tiktoken：OpenAI的分词工具（辅助文本处理）
%pip install --upgrade --quiet langchain langchain-openai faiss-cpu tiktoken

# 2. 导入所需库
from operator import itemgetter  # 用于从字典中提取指定键的值（简化数据处理）
from langchain_community.vectorstores import FAISS  # 向量库（存储文本向量，支持快速检索）
from langchain_core.output_parsers import StrOutputParser  # 将AI的输出（AIMessage对象）转换为纯字符串
from langchain_core.prompts import ChatPromptTemplate  # 构建提示模板（定义AI的行为规则）
from langchain_core.runnables import RunnablePassthrough  # LCEL组件：传递输入数据（不修改，直接传给下一环）
from langchain_openai import ChatOpenAI, OpenAIEmbeddings  # ChatOpenAI：大语言模型；OpenAIEmbeddings：文本转向量工具

# 3. 初始化核心组件（RAG的基础"零件"）
# 3.1 向量库：存储外部文本的向量表示（计算机可理解的数字形式）
# 输入：
# - 文本列表：["harrison worked at kensho"]（我们要让AI参考的外部知识，这里仅1条）
# - 嵌入工具：OpenAIEmbeddings()（将文本转换为向量的工具）
# 作用：把文本转成向量后存储，方便后续用"语义相似性"快速找到相关文本
vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"],  # 待检索的外部文本（核心知识源）
    embedding=OpenAIEmbeddings()    # 文本→向量的转换工具
)

# 3.2 检索器：从向量库中"查找与问题语义相似的文本"
# 作用：接收用户问题，返回向量库中最相似的文本（默认返回1条）
retriever = vectorstore.as_retriever()

# 3.3 提示模板：定义AI的回答规则（强制AI"只看检索到的文本回答，不瞎编"）
# 占位符说明：
# - {context}：检索器返回的相关文本（外部知识）
# - {question}：用户的问题
prompt = ChatPromptTemplate.from_template("""
Answer the question based only on the following context:
{context}
Question: {question}
""")

# 3.4 AI模型：用于生成回答的大语言模型（这里用OpenAI的模型）
model = ChatOpenAI()

# 4. 构建基础RAG链（串联所有组件，形成完整流程）
# 链的逻辑（数据流向）：
# 输入（用户问题）→ 处理为{context: 检索结果, question: 用户问题} → 填充提示模板 → AI生成回答 → 转换为字符串输出
chain = (
    # 第一步：处理输入，生成后续步骤需要的字典
    # - "context"：用检索器（retriever）获取与问题相关的文本
    # - "question"：用RunnablePassthrough()直接传递用户输入的问题（不修改）
    {"context": retriever, "question": RunnablePassthrough()}
    # 第二步：将上一步的字典（context+question）传入提示模板，生成完整提示（如"基于context:xxx回答question:xxx"）
    | prompt
    # 第三步：AI模型（model）根据完整提示生成回答（返回AIMessage对象）
    | model
    # 第四步：将AI的输出（AIMessage）转换为纯字符串（方便阅读）
    | StrOutputParser()
)

# 5. 测试链（验证是否能基于外部文本回答）
# 测试1：基础提问（问"harrison在哪工作"）
result1 = chain.invoke("where did harrison work?")
print("基础RAG回答：", result1)  # 预期输出：'Harrison worked at Kensho.'（AI仅用向量库中的文本回答）

# 测试2：扩展功能（指定回答语言，如意大利语）
# 5.1 修改提示模板，新增{language}占位符（支持指定回答语言）
multi_lang_prompt = ChatPromptTemplate.from_template("""
Answer the question based only on the following context:
{context}
Question: {question}
Answer in the following language: {language}
""")

# 5.2 构建多语言RAG链（输入需包含question和language两个参数）
multi_lang_chain = (
    # 输入处理：从用户输入的字典中提取question、language，并用question检索context
    {
        "context": itemgetter("question") | retriever,  # 用question作为检索关键词，获取context
        "question": itemgetter("question"),            # 提取用户输入中的"question"键值
        "language": itemgetter("language")             # 提取用户输入中的"language"键值
    }
    | multi_lang_prompt  # 填充多语言提示模板
    | model              # AI生成对应语言的回答
    | StrOutputParser()  # 转换为字符串
)

# 5.3 测试多语言功能（指定意大利语回答）
result2 = multi_lang_chain.invoke({"question": "where did harrison work", "language": "italian"})
print("意大利语回答：", result2)  # 预期输出：'Harrison ha lavorato a Kensho.'

场景2：对话式RAG链（支持对话历史，处理指代问题）

# 1. 安装依赖（若已安装可跳过）
%pip install --upgrade --quiet langchain langchain-openai faiss-cpu tiktoken

# 2. 导入所需库（包含场景1的基础库+处理对话历史的新增库）
from operator import itemgetter
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate  # PromptTemplate用于构建纯文本提示
from langchain_core.runnables import RunnableParallel, RunnablePassthrough  # RunnableParallel：并行处理多个输入
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.schema import format_document  # 格式化Document对象（提取文本内容）
from langchain_core.messages import AIMessage, HumanMessage, get_buffer_string  # 处理对话消息的工具

# 3. 初始化基础组件（复用场景1的核心组件，避免重复创建）
vectorstore = FAISS.from_texts(["harrison worked at kensho"], embedding=OpenAIEmbeddings())  # 向量库
retriever = vectorstore.as_retriever()  # 检索器
model = ChatOpenAI()  # AI模型

# 4. 新增对话处理组件（解决"指代问题"的关键）
# 4.1 独立问题生成提示：将"对话历史+跟进问题"改写为"无指代的独立问题"
# 例如：历史是"用户问谁写的笔记本，AI答Harrison"，跟进问题是"他在哪工作"→改写为"Harrison在哪工作"
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template("""
Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question.
Chat History:
{chat_history}  # 对话历史（如"Human: ...\nAI: ..."）
Follow Up Input: {question}  # 用户的跟进问题（含指代，如"where did he work?"）
Standalone question:  # 输出：无指代的独立问题（如"where did Harrison work?"）
""")

# 4.2 回答提示：与场景1一致（强制AI仅基于检索到的context回答）
ANSWER_PROMPT = ChatPromptTemplate.from_template("""
Answer the question based only on the following context:
{context}
Question: {question}
""")

# 4.3 文档合并函数：将检索器返回的"Document对象列表"转换为"纯文本字符串"
# 原因：检索器返回的是带格式的Document对象（如[Document(page_content="xxx")]），AI无法直接读取，需提取文本
def _combine_documents(docs):
    # 遍历Document列表，提取每个对象的page_content（纯文本），用空行分隔拼接
    return "\n\n".join([format_document(doc, PromptTemplate.from_template("{page_content}")) for doc in docs])

# 5. 构建对话式RAG链（分两步：先解决指代，再检索回答）
# 5.1 第一步：生成独立问题（核心：消除指代歧义）
_inputs = RunnableParallel(
    # 输出键：standalone_question（独立问题）
    standalone_question=RunnablePassthrough.assign(
        # 将chat_history（Message列表）转换为字符串（如"Human: ...\nAI: ..."），方便LLM理解
        chat_history=lambda x: get_buffer_string(x["chat_history"])
    )
    | CONDENSE_QUESTION_PROMPT  # 传入"chat_history+question"到提示
    | ChatOpenAI(temperature=0)  # 用LLM生成独立问题（temperature=0确保结果稳定，不随机）
    | StrOutputParser()  # 转换为纯字符串
)

# 5.2 第二步：基于独立问题检索+生成回答
_context = {
    # "context"：用独立问题检索→合并文档为纯文本
    "context": itemgetter("standalone_question") | retriever | _combine_documents,
    # "question"：用独立问题作为最终提问（替代含指代的原问题）
    "question": lambda x: x["standalone_question"]
}

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

# 6. 测试对话式链（验证是否能处理指代）
# 测试1：无对话历史（直接提问，类似场景1）
result1 = conversational_qa_chain.invoke({
    "question": "where did harrison work?",  # 无指代的问题
    "chat_history": []  # 空历史
})
print("无历史回答：", result1.content)  # 预期输出：'Harrison was employed at Kensho.'

# 测试2：有对话历史（处理指代"he"）
result2 = conversational_qa_chain.invoke({
    "question": "where did he work?",  # "he"需要结合历史才能理解（指Harrison）
    "chat_history": [  # 对话历史：明确"he"的指代
        HumanMessage(content="Who wrote this notebook?"),  # 用户：谁写的笔记本？
        AIMessage(content="Harrison")                     # AI：Harrison
    ]
})
print("带指代回答：", result2.content)  # 预期输出：'Harrison worked at Kensho.'（成功识别"he"指Harrison）

场景3：带内存+返回源文档的RAG链（自动存历史，显示回答来源）

# 1. 安装依赖（若已安装可跳过）
%pip install --upgrade --quiet langchain langchain-openai faiss-cpu tiktoken

# 2. 导入所需库（包含前两场景的库+内存组件）
from operator import itemgetter
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough  # RunnableLambda：包装函数为可运行组件
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.schema import Document, format_document
from langchain_core.messages import AIMessage, HumanMessage, get_buffer_string
from langchain.memory import ConversationBufferMemory  # 存储对话历史的内存组件

# 3. 初始化基础组件（复用前两场景的核心组件）
vectorstore = FAISS.from_texts(["harrison worked at kensho"], embedding=OpenAIEmbeddings())  # 向量库
retriever = vectorstore.as_retriever()  # 检索器
model = ChatOpenAI()  # AI模型
# 独立问题生成提示（复用场景2）
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template("""
Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:
""")
# 回答提示（复用场景2）
ANSWER_PROMPT = ChatPromptTemplate.from_template("""
Answer the question based only on the following context:
{context}
Question: {question}
""")
# 文档合并函数（复用场景2）
def _combine_documents(docs):
    return "\n\n".join([format_document(doc, PromptTemplate.from_template("{page_content}")) for doc in docs])

# 4. 新增内存组件：存储对话历史（避免每次手动传入chat_history）
memory = ConversationBufferMemory(
    return_messages=True,  # 存储格式：Message对象列表（如[HumanMessage, AIMessage]），而非纯文本
    output_key="answer",   # 内存中"AI输出"对应的键名（需与链的输出键匹配）
    input_key="question"   # 内存中"用户输入"对应的键名（需与链的输入键匹配）
)

# 5. 构建完整链（四步流程：加载内存→生成独立问题→检索文档→生成回答+返回源文档）
final_chain = (
    # 步骤1：加载内存，获取对话历史（自动从memory中读取，无需手动传chat_history）
    RunnablePassthrough.assign(
        # 逻辑：memory.load_memory_variables({})→返回{"history": [Message列表]}→提取"history"键的值
        chat_history=RunnableLambda(memory.load_memory_variables) | itemgetter("history")
    )
    # 步骤2：生成独立问题（复用场景2的逻辑，解决指代）
    | {
        "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对象，不合并，用于后续显示来源）
    | {
        "docs": itemgetter("standalone_question") | retriever,  # 用独立问题检索，返回Document列表
        "question": lambda x: x["standalone_question"]  # 传递独立问题
    }
    # 步骤4：生成回答+返回源文档（最终输出包含answer和docs）
    | {
        # "answer"键：生成AI回答
        "answer": {
            "context": lambda x: _combine_documents(x["docs"]),  # 合并文档为context
            "question": itemgetter("question")  # 用独立问题提问
        }
        | ANSWER_PROMPT  # 填充提示
        | ChatOpenAI(),  # AI生成回答
        # "docs"键：传递检索到的源文档（显示回答来源）
        "docs": itemgetter("docs")
    }
)

# 6. 测试链（需手动保存内存，当前版本不支持自动保存）
# 测试1：第一次提问（无历史）
inputs1 = {"question": "where did harrison work?"}  # 输入仅需用户问题（内存自动加载空历史）
result1 = final_chain.invoke(inputs1)

# 打印结果：包含回答和源文档（证明回答基于哪条文本）
print("第一次回答：", result1["answer"].content)  # 预期输出：'Harrison was employed at Kensho.'
print("第一次检索到的源文档：", [doc.page_content for doc in result1["docs"]])  # 预期输出：['harrison worked at kensho']

# 关键步骤：手动保存当前对话到内存（否则下次调用会丢失历史）
# 格式：save_context(用户输入字典, {"answer": AI回答内容})
memory.save_context(inputs1, {"answer": result1["answer"].content})

# 测试2：第二次提问（有历史，处理指代"he"）
inputs2 = {"question": "but where did he really work?"}  # "he"指历史中的Harrison
result2 = final_chain.invoke(inputs2)

print("第二次回答：", result2["answer"].content)  # 预期输出：'Harrison actually worked at Kensho.'
print("第二次检索到的源文档：", [doc.page_content for doc in result2["docs"]])  # 预期输出：['harrison worked at kensho']

核心说明：

1. 注释重点：每个组件的作用、数据流向（输入→输出）、关键参数含义（如return_messages=True）都有解释，帮助理解"为什么这么写"；
2. 连贯性：三个场景逐步递进，后一场景复用前一场景的组件，注释中明确标注"复用自XX场景"，强化整体逻辑；
3. 初学者友好：避免抽象术语，用"外部知识"代替"外部文本"，用"消除指代歧义"解释"生成独立问题"的目的，降低理解门槛。

直接复制代码运行，结合注释观察每一步的输出，能清晰看到从"基础检索"到"带记忆的对话检索"的完整演进过程。