构建检索增强生成（RAG）应用：第二部分

来源：Build a Retrieval Augmented Generation (RAG) App: Part 2 | 🦜️🔗 LangChain

---

构建检索增强生成（RAG）应用：第二部分

        在许多问答应用中，我们希望允许用户进行来回对话，这意味着应用需要某种“记忆”以记录过去的问题和答案，并具备将其纳入当前思考的逻辑。

这是一个多部分教程的第二部分：

• 第1部分介绍了RAG并演示了一个最小实现。

• 第2部分（本指南）扩展了实现以支持对话式交互和多步检索过程。

我们在这里重点讲解如何添加管理历史消息的逻辑。这涉及到聊天历史的管理。

我们将介绍两种方法：

• 链式结构（Chains）：最多执行一次检索步骤；

• 智能体（Agents）：允许大语言模型自行决定是否执行多次检索。

📌注意
本教程中的方法利用了现代聊天模型中的工具调用（Tool Calling）能力。可在此页面查看支持该功能的模型列表。

我们仍使用第1部分中提到的外部知识源：Lilian Weng 撰写的博客文章《LLM驱动的自主智能体（LLM Powered Autonomous Agents）》。

---

环境准备

所需组件

我们将从 LangChain 集成套件中选择三个组件：

选择聊天模型：

pip install -qU "langchain[google-genai]"

import getpass, osif not os.environ.get("GOOGLE_API_KEY"):os.environ["GOOGLE_API_KEY"] = getpass.getpass("输入 Google Gemini 的 API Key：")from langchain.chat_models import init_chat_modelllm = init_chat_model("gemini-2.0-flash", model_provider="google_genai")

选择嵌入模型：

pip install -qU langchain-openai

if not os.environ.get("OPENAI_API_KEY"):os.environ["OPENAI_API_KEY"] = getpass.getpass("输入 OpenAI 的 API Key：")from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings(model="text-embedding-3-large")

选择向量数据库：

pip install -qU langchain-core

from langchain_core.vectorstores import InMemoryVectorStorevector_store = InMemoryVectorStore(embeddings)

---

其他依赖

%%capture --no-stderr
%pip install --upgrade --quiet langgraph langchain-community beautifulsoup4

---

使用 LangSmith（可选）

LangSmith 可视化工具能帮助你调试链条或代理中的多步骤调用过程。若使用，需设置以下环境变量：

os.environ["LANGSMITH_TRACING"] = "true"
if not os.environ.get("LANGSMITH_API_KEY"):os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

---

使用 Chains 实现对话式 RAG

我们回顾一下第1部分中构建的向量数据库，索引了 Lilian Weng 的博客内容：

import bs4
from langchain import hub
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing_extensions import List, TypedDictloader = WebBaseLoader(web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),bs_kwargs=dict(parse_only=bs4.SoupStrainer(class_=("post-content", "post-title", "post-header"))),
)
docs = loader.load()text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
all_splits = text_splitter.split_documents(docs)_ = vector_store.add_documents(documents=all_splits)

---

状态结构设计

在 Part 1 中，我们将用户输入、检索上下文和生成的回答分别作为状态的不同字段。而在对话体验中，更自然的做法是将状态表示为消息序列，包括：

• 用户输入 → HumanMessage

• 检索请求 → AIMessage（含工具调用）

• 检索结果 → ToolMessage

• 最终回答 → AIMessage

LangGraph 提供了一个内置的 MessagesState 结构以简化这一过程：

from langgraph.graph import MessagesState, StateGraphgraph_builder = StateGraph(MessagesState)

将检索封装为工具

from langchain_core.tools import tool@tool(response_format="content_and_artifact")
def retrieve(query: str):"""检索与查询相关的信息"""retrieved_docs = vector_store.similarity_search(query, k=2)serialized = "\n\n".join((f"Source: {doc.metadata}\n" f"Content: {doc.page_content}")for doc in retrieved_docs)return serialized, retrieved_docs

---

构建三步图节点

1. 判断是否检索，或直接回复

from langchain_core.messages import SystemMessagedef query_or_respond(state: MessagesState):llm_with_tools = llm.bind_tools([retrieve])response = llm_with_tools.invoke(state["messages"])return {"messages": [response]}

2. 执行检索工具

from langgraph.prebuilt import ToolNodetools = ToolNode([retrieve])

3. 使用检索内容生成回答

def generate(state: MessagesState):recent_tool_messages = []for message in reversed(state["messages"]):if message.type == "tool":recent_tool_messages.append(message)else:breaktool_messages = recent_tool_messages[::-1]docs_content = "\n\n".join(doc.content for doc in tool_messages)system_message_content = ("你是一个问答助手。请使用下列检索到的上下文内容回答用户问题。如果不知道答案，就说不知道。""请将回答控制在三句话以内，保持简洁。\n\n"f"{docs_content}")conversation_messages = [msg for msg in state["messages"]if msg.type in ("human", "system") or (msg.type == "ai" and not msg.tool_calls)]prompt = [SystemMessage(system_message_content)] + conversation_messagesresponse = llm.invoke(prompt)return {"messages": [response]}

---

构建图流程

from langgraph.graph import END
from langgraph.prebuilt import tools_conditiongraph_builder.add_node(query_or_respond)
graph_builder.add_node(tools)
graph_builder.add_node(generate)graph_builder.set_entry_point("query_or_respond")
graph_builder.add_conditional_edges("query_or_respond",tools_condition,{END: END, "tools": "tools"},
)
graph_builder.add_edge("tools", "generate")
graph_builder.add_edge("generate", END)graph = graph_builder.compile()

可视化流程图：

from IPython.display import Image, displaydisplay(Image(graph.get_graph().draw_mermaid_png()))

---

测试

input_message = "Hello"
for step in graph.stream({"messages": [{"role": "user", "content": input_message}]},stream_mode="values",
):step["messages"][-1].pretty_print()

---

管理聊天历史（多轮记忆）

我们可以使用 MemorySaver 作为内存检查点：

from langgraph.checkpoint.memory import MemorySavermemory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "abc123"}}

---

测试历史上下文对话

input_message = "Can you look up some common ways of doing it?"for step in graph.stream({"messages": [{"role": "user", "content": input_message}]},stream_mode="values",config=config,
):step["messages"][-1].pretty_print()

模型能正确理解上下文并生成带有历史的检索查询。

---

使用 Agent 实现更复杂的 RAG

我们使用 LangGraph 提供的 ReAct 代理构造器：

from langgraph.prebuilt import create_react_agentagent_executor = create_react_agent(llm, [retrieve], checkpointer=memory)

可视化代理图：

display(Image(agent_executor.get_graph().draw_mermaid_png()))

---

测试 Agent 多步检索

config = {"configurable": {"thread_id": "def234"}}
input_message = ("What is the standard method for Task Decomposition?\n\n""Once you get the answer, look up common extensions of that method."
)for event in agent_executor.stream({"messages": [{"role": "user", "content": input_message}]},stream_mode="values",config=config,
):event["messages"][-1].pretty_print()

---

小结与后续步骤

我们已经构建了：

• 使用 Chain 模式的单步检索 RAG；

• 使用 Agent 模式的多步迭代检索 RAG。

下一步可探索：

• 不同的检索器类型；

• 更复杂的消息历史结构；

• 高级 Agent 架构。

---