Skip to main content

​
概述

LLM 最强大的应用之一是复杂的问答（Q&A）聊天机器人。这类应用能够针对特定来源的信息回答问题，使用的技术称为检索增强生成（Retrieval Augmented Generation），即 RAG。 本教程将展示如何在非结构化文本数据源上构建一个简单的问答应用。我们将演示：
  1. 一个使用简单工具执行检索的 RAG 代理。这是一种通用的实现方式。
  2. 一个每次查询仅进行单次 LLM 调用的两步 RAG 链。这是针对简单查询的快速有效方法。

​
概念

我们将涵盖以下概念：
  • 索引：将数据从来源摄取并建立索引的流水线。这通常在独立进程中进行。
  • 检索与生成：实际的 RAG 流程，在运行时获取用户查询，从索引中检索相关数据，然后将其传递给模型。
完成数据索引后，我们将使用代理作为编排框架来实现检索与生成步骤。
本教程的索引部分大体遵循语义搜索教程。如果你的数据已可供搜索（即你已有执行搜索的函数），或者你已熟悉该教程的内容，可以直接跳到检索与生成部分。

​
预览

在本指南中，我们将构建一个回答网站内容相关问题的应用。我们使用的具体网站是 Lilian Weng 的博客文章 LLM Powered Autonomous Agents，以便提问关于该文章内容的问题。 我们可以用约 40 行代码创建一个简单的索引流水线和 RAG 链。完整代码片段如下： 
import bs4
from langchain.agents import AgentState, create_agent
from langchain_community.document_loaders import WebBaseLoader
from langchain.messages import MessageLikeRepresentation
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load and chunk contents of the blog
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)

# Index chunks
_ = vector_store.add_documents(documents=all_splits)

# Construct a tool for retrieving context
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """Retrieve information to help answer a query."""
    retrieved_docs = vector_store.similarity_search(query, k=2)
    serialized = "\n\n".join(
        (f"Source: {doc.metadata}\nContent: {doc.page_content}")
        for doc in retrieved_docs
    )
    return serialized, retrieved_docs

tools = [retrieve_context]
# If desired, specify custom instructions
prompt = (
    "You have access to a tool that retrieves context from a blog post. "
    "Use the tool to help answer user queries."
)
agent = create_agent(model, tools, system_prompt=prompt)

query = "What is task decomposition?"
for step in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()

================================ Human Message =================================

What is task decomposition?
================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_xTkJr8njRY0geNz43ZvGkX0R)
 Call ID: call_xTkJr8njRY0geNz43ZvGkX0R
  Args:
    query: task decomposition
================================= Tool Message =================================
Name: retrieve_context

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Task decomposition can be done by...

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Component One: Planning...
================================== Ai Message ==================================

Task decomposition refers to...
查看 LangSmith 追踪记录。

​
安装配置

​
安装

本教程需要以下 langchain 依赖： 
pip install langchain langchain-text-splitters langchain-community bs4
更多详情，请参见安装指南。

​
LangSmith

使用 LangChain 构建的许多应用包含多个步骤和多次 LLM 调用。随着应用变得更加复杂，能够检查链或代理内部发生的事情变得至关重要。最好的方式是使用 LangSmith。 在上方链接注册后，确保设置环境变量以开始记录追踪： 
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
或者在 Python 中设置： 
import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

​
组件

我们需要从 LangChain 的集成套件中选择三个组件。 选择聊天模型：
👉 Read the OpenAI chat model integration docs
pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.2")
选择嵌入模型： 
pip install -U "langchain-openai"

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
选择向量存储： 
pip install -U "langchain-core"

from langchain_core.vectorstores import InMemoryVectorStore

vector_store = InMemoryVectorStore(embeddings)

​
1. 索引

本节是语义搜索教程内容的简化版。如果你的数据已完成索引并可供搜索（即你已有执行搜索的函数），或者你已熟悉文档加载器、嵌入模型和向量存储，可以直接跳到下一节检索与生成。
索引通常按如下流程进行：
  1. 加载：首先需要加载数据，使用文档加载器完成。
  2. 分割：文本分割器将大型 Document 拆分成较小的块。这对于索引数据和将其传递给模型都很有用，因为大块内容更难检索，且可能超出模型有限的上下文窗口。
  3. 存储：我们需要一个地方来存储和索引分割后的内容，以便之后进行检索。通常使用向量存储和嵌入模型来完成。
索引示意图

​
加载文档

我们首先需要加载博客文章的内容。可以使用文档加载器来完成，它们是从来源加载数据并返回 Document 对象列表的对象。 本例中我们使用 WebBaseLoader，它使用 urllib 从网页 URL 加载 HTML，并使用 BeautifulSoup 将其解析为文本。我们可以通过 bs_kwargs 向 BeautifulSoup 解析器传入参数来自定义 HTML → 文本的解析过程（参见 BeautifulSoup 文档）。本例中只保留 class 为 “post-content”、“post-title” 或 “post-header” 的 HTML 标签。 
import bs4
from langchain_community.document_loaders import WebBaseLoader

# Only keep post title, headers, and content from the full HTML.
bs4_strainer = bs4.SoupStrainer(class_=("post-title", "post-header", "post-content"))
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs={"parse_only": bs4_strainer},
)
docs = loader.load()

assert len(docs) == 1
print(f"Total characters: {len(docs[0].page_content)}")

Total characters: 43131

print(docs[0].page_content[:500])

      LLM Powered Autonomous Agents

Date: June 23, 2023  |  Estimated Reading Time: 31 min  |  Author: Lilian Weng


Building agents with LLM (large language model) as its core controller is a cool concept. Several proof-of-concepts demos, such as AutoGPT, GPT-Engineer and BabyAGI, serve as inspiring examples. The potentiality of LLM extends beyond generating well-written copies, stories, essays and programs; it can be framed as a powerful general problem solver.
Agent System Overview#
In
深入了解 DocumentLoader：从来源加载数据并返回 Document 列表的对象。
  • 集成：160+ 种集成可供选择。
  • BaseLoader：基础接口的 API 参考。

​
分割文档

我们加载的文档超过 42,000 个字符，对于许多模型的上下文窗口来说过长。即便是能够放入完整文章的模型，在非常长的输入中查找信息也会遇到困难。 为了解决这个问题，我们将 Document 分割成块以便嵌入和向量存储。这有助于我们在运行时只检索博客文章中最相关的部分。 与语义搜索教程一样，我们使用 RecursiveCharacterTextSplitter，它会使用换行符等常见分隔符递归地分割文档，直到每个块达到合适的大小。这是通用文本场景下推荐的文本分割器。 
from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # chunk size (characters)
    chunk_overlap=200,  # chunk overlap (characters)
    add_start_index=True,  # track index in original document
)
all_splits = text_splitter.split_documents(docs)

print(f"Split blog post into {len(all_splits)} sub-documents.")

Split blog post into 66 sub-documents.
深入了解 TextSplitter：将 Document 对象列表分割成较小块以供存储和检索的对象。

​
存储文档

现在我们需要对 66 个文本块建立索引，以便在运行时进行检索。遵循语义搜索教程，我们的做法是嵌入每个文档块的内容，并将这些嵌入插入向量存储。对于给定的输入查询，我们可以使用向量搜索检索相关文档。 我们可以使用教程开头选择的向量存储和嵌入模型，通过单条命令完成所有文档块的嵌入和存储。 
document_ids = vector_store.add_documents(documents=all_splits)

print(document_ids[:3])

['07c18af6-ad58-479a-bfb1-d508033f9c64', '9000bf8e-1993-446f-8d4d-f4e507ba4b8f', 'ba3b5d14-bed9-4f5f-88be-44c88aedc2e6']
深入了解 Embeddings：文本嵌入模型的封装，用于将文本转换为嵌入向量。
  • 集成：30+ 种集成可供选择。
  • 接口：基础接口的 API 参考。
VectorStore：向量数据库的封装，用于存储和查询嵌入向量。
  • 集成：40+ 种集成可供选择。
  • 接口：基础接口的 API 参考。
至此，索引部分的流水线已完成。现在我们拥有一个可查询的向量存储，其中包含博客文章分块后的内容。对于用户的问题，我们应该能够返回博客文章中能够回答该问题的片段。

​
2. 检索与生成

RAG 应用通常按如下流程工作：
  1. 检索：根据用户输入，使用检索器从存储中检索相关的分块内容。
  2. 生成：模型使用包含问题和检索到的数据的提示词生成答案。
检索示意图 现在让我们编写实际的应用逻辑。我们希望创建一个简单的应用：接收用户问题，检索与该问题相关的文档，将检索到的文档和原始问题传递给模型，并返回答案。 我们将演示：
  1. 一个使用简单工具执行检索的 RAG 代理。这是一种通用的实现方式。
  2. 一个每次查询仅进行单次 LLM 调用的两步 RAG 链。这是针对简单查询的快速有效方法。

​
RAG 代理

RAG 应用的一种实现方式是构建一个带有检索信息工具的简单代理。我们可以通过实现一个封装向量存储的工具来组建一个最简 RAG 代理： 
from langchain.tools import tool

@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """Retrieve information to help answer a query."""
    retrieved_docs = vector_store.similarity_search(query, k=2)
    serialized = "\n\n".join(
        (f"Source: {doc.metadata}\nContent: {doc.page_content}")
        for doc in retrieved_docs
    )
    return serialized, retrieved_docs
这里我们使用工具装饰器将工具配置为把原始文档作为制品附加到每条 ToolMessage 上。这让我们能够在应用中访问文档元数据，与发送给模型的字符串表示相互独立。
检索工具并不局限于像上面示例中那样只有一个字符串 query 参数。你可以通过添加参数来强制 LLM 指定额外的搜索参数——例如，一个分类字段：
from typing import Literal

def retrieve_context(query: str, section: Literal["beginning", "middle", "end"]):
有了工具，我们可以构建代理： 
from langchain.agents import create_agent


tools = [retrieve_context]
# If desired, specify custom instructions
prompt = (
    "You have access to a tool that retrieves context from a blog post. "
    "Use the tool to help answer user queries."
)
agent = create_agent(model, tools, system_prompt=prompt)
让我们来测试一下。我们构造一个通常需要多轮迭代检索才能回答的问题： 
query = (
    "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.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    event["messages"][-1].pretty_print()

================================ Human Message =================================

What is the standard method for Task Decomposition?

Once you get the answer, look up common extensions of that method.
================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_d6AVxICMPQYwAKj9lgH4E337)
 Call ID: call_d6AVxICMPQYwAKj9lgH4E337
  Args:
    query: standard method for Task Decomposition
================================= Tool Message =================================
Name: retrieve_context

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Task decomposition can be done...

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Component One: Planning...
================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_0dbMOw7266jvETbXWn4JqWpR)
 Call ID: call_0dbMOw7266jvETbXWn4JqWpR
  Args:
    query: common extensions of the standard method for Task Decomposition
================================= Tool Message =================================
Name: retrieve_context

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Task decomposition can be done...

Source: {'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}
Content: Component One: Planning...
================================== Ai Message ==================================

The standard method for Task Decomposition often used is the Chain of Thought (CoT)...
注意代理的执行流程：
  1. 生成查询以搜索任务分解的标准方法；
  2. 收到答案后，生成第二个查询以搜索该方法的常见扩展；
  3. 获得所有必要上下文后，回答问题。
我们可以在 LangSmith 追踪记录中看到完整的步骤序列，以及延迟和其他元数据。
你可以直接使用 LangGraph 框架添加更深层次的控制和自定义——例如，添加步骤来评估文档相关性并重写搜索查询。查看 LangGraph 的 Agentic RAG 教程了解更高级的实现方式。

​
RAG 链

在上述代理式 RAG 实现中，我们允许 LLM 自主决定是否发起工具调用来帮助回答用户查询。这是一种通用解决方案，但也存在一些权衡：
✅ 优点⚠️ 缺点
按需检索 – LLM 可以处理问候、追问和简单查询，无需触发不必要的检索。两次推理调用 – 当执行检索时，需要一次调用生成查询，另一次生成最终响应。
上下文感知搜索查询 – 将检索作为带有 query 输入的工具，LLM 可以结合对话上下文构造自己的查询。控制力降低 – LLM 可能在实际需要检索时跳过，或在不必要时发起额外检索。
支持多次检索 – LLM 可以为单个用户查询执行多次检索。
另一种常见方式是两步链：始终执行一次检索（可能直接使用原始用户查询），并将结果作为单次 LLM 查询的上下文。这样每次查询只需一次推理调用，以灵活性换取更低的延迟。 这种方式不再循环调用模型，而是进行单次处理。 我们可以通过从代理中移除工具，并将检索步骤嵌入自定义提示词来实现这个链： 
from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def prompt_with_context(request: ModelRequest) -> str:
    """Inject context into state messages."""
    last_query = request.state["messages"][-1].text
    retrieved_docs = vector_store.similarity_search(last_query)

    docs_content = "\n\n".join(doc.page_content for doc in retrieved_docs)

    system_message = (
        "You are a helpful assistant. Use the following context in your response:"
        f"\n\n{docs_content}"
    )

    return system_message


agent = create_agent(model, tools=[], middleware=[prompt_with_context])
让我们来测试一下： 
query = "What is task decomposition?"
for step in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()

================================ Human Message =================================

What is task decomposition?
================================== Ai Message ==================================

Task decomposition is...
在 LangSmith 追踪记录中，我们可以看到检索到的上下文被注入到模型提示词中。 这是在受限场景下处理简单查询的快速有效方法，适用于通常需要通过语义搜索拉取额外上下文的情况。
上述 RAG 链 将检索到的上下文注入该次运行的单条系统消息中。与代理式 RAG 类似，有时我们希望在应用状态中保留原始来源文档以访问文档元数据。对于两步链的情况，我们可以通过以下方式实现：
  1. 在状态中添加一个键用于存储检索到的文档；
  2. 通过模型前钩子添加一个新节点来填充该键（同时注入上下文）。
from typing import Any
from langchain_core.documents import Document
from langchain.agents.middleware import AgentMiddleware, AgentState


class State(AgentState):
    context: list[Document]


class RetrieveDocumentsMiddleware(AgentMiddleware[State]):
    state_schema = State

    def before_model(self, state: AgentState) -> dict[str, Any] | None:
        last_message = state["messages"][-1]
        retrieved_docs = vector_store.similarity_search(last_message.text)

        docs_content = "\n\n".join(doc.page_content for doc in retrieved_docs)

        augmented_message_content = (
            f"{last_message.text}\n\n"
            "Use the following context to answer the query:\n"
            f"{docs_content}"
        )
        return {
            "messages": [last_message.model_copy(update={"content": augmented_message_content})],
            "context": retrieved_docs,
        }


agent = create_agent(
    model,
    tools=[],
    middleware=[RetrieveDocumentsMiddleware()],
)

​
后续步骤

现在我们已经通过 create_agent 实现了一个简单的 RAG 应用，可以轻松地融入新功能并深入探索：
在 GitHub 上编辑此页面 或提交问题。
将这些文档连接到 Claude、VSCode 等工具，通过 MCP 获取实时答案。