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

发布于：2025-06-06 ⋅ 阅读:(16) ⋅ 点赞:(0)

来源：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, os

if 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_model

llm = 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 OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
选择向量数据库：
pip install -qU langchain-core

from langchain_core.vectorstores import InMemoryVectorStore

vector_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, TypedDict

loader = 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, StateGraph

graph_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 SystemMessage

def 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 ToolNode

tools = 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:
            break
    tool_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_messages

    response = llm.invoke(prompt)
    return {"messages": [response]}

构建图流程

from langgraph.graph import END
from langgraph.prebuilt import tools_condition

graph_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, display

display(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 MemorySaver

memory = 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_agent

agent_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 架构。

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布