使用 LangChain 构建 RAG Agent

如果你的数据已经可用于搜索（即，你有一个执行搜索的函数），或者你对该教程的内容感到满意，请随时跳至 [检索和生成](#2-retrieval-and-generation) 部分。

如果你的数据已经索引并可用于搜索（即，你有一个执行搜索的函数），或者如果你对 Document Loaders、Embeddings 和 Vector Stores 感到满意，请随时跳至下一节 检索和生成。

索引通常的工作流程如下：

1. 加载（Load）：首先我们需要加载数据。这是通过 Document Loaders 完成的。
2. 分割（Split）：Text Splitters 将大型 Documents 分割成较小的块。这对于索引数据和将其传递给模型都很有用，因为大块更难搜索并且不适合模型的有限上下文窗口。
3. 存储（Store）：我们需要 somewhere 来存储和索引我们的分割，以便以后可以搜索它们。这通常使用 VectorStore 和 Embeddings 模型完成。

加载文档

我们首先需要加载博客文章内容。我们可以使用 DocumentLoaders 来完成，这是从源加载数据并返回 @[Document] 对象列表的对象。

:::python 在这种情况下，我们将使用 WebBaseLoader，它使用 urllib 从 Web URL 加载 HTML，并使用 BeautifulSoup 将其解析为文本。我们可以通过 bs_kwargs 将参数传递给 BeautifulSoup 解析器来自定义 HTML -> 文本解析（请参阅 BeautifulSoup 文档）。在这种情况下，只有类为”post-content”、“post-title”或”post-header”的 HTML 标签是相关的，所以我们将删除所有其他标签。

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 仅保留完整 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"总字符数：{len(docs[0].page_content)}")

总字符数：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

:::

:::js

import "cheerio";
import { CheerioWebBaseLoader } from "@langchain/community/document_loaders/web/cheerio";

const pTagSelector = "p";
const cheerioLoader = new CheerioWebBaseLoader(
  "https://lilianweng.github.io/posts/2023-06-23-agent/",
  {
    selector: pTagSelector,
  }
);

const docs = await cheerioLoader.load();

console.assert(docs.length === 1);
console.log(`总字符数：${docs[0].pageContent.length}`);

总字符数：22360

console.log(docs[0].pageContent.slice(0, 500));

Building agents with LLM (large language model) as its core controller is...

:::

深入了解

DocumentLoader：从源加载数据作为 Documents 列表的对象。

• 集成：160+ 集成可供选择。
• @[BaseLoader]：基础接口的 API 参考。

分割文档

我们加载的文档超过 4.2 万个字符，太长而无法适应许多模型的上下文窗口。即使对于那些可以将完整帖子放入其上下文窗口的模型，模型也很难在非常长的输入中找到信息。

为了解决这个问题，我们将把 @[Document] 分割成块以便嵌入和向量存储。这应该有助于我们在运行时只检索博客文章中最相关的部分。

与 语义搜索教程 中一样，我们使用 RecursiveCharacterTextSplitter，它将递归地使用常见分隔符（如换行符）分割文档，直到每个块达到适当的大小。这是通用文本用例的推荐文本分割器。

:::python

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 块大小（字符）
    chunk_overlap=200,  # 块重叠（字符）
    add_start_index=True,  # 跟踪原始文档中的索引
)
all_splits = text_splitter.split_documents(docs)

print(f"将博客文章分割成 {len(all_splits)} 个子文档。")

将博客文章分割成 66 个子文档。

深入了解

TextSplitter：将 @[Document] 对象列表分割成较小块以进行存储和检索的对象。

• 集成
• 接口：基础接口的 API 参考。 :::

:::js

import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters";

const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 1000,
  chunkOverlap: 200,
});
const allSplits = await splitter.splitDocuments(docs);
console.log(`将博客文章分割成 ${allSplits.length} 个子文档。`);

将博客文章分割成 29 个子文档。

:::

存储文档

现在我们需要索引 66 个文本块，以便在运行时搜索它们。遵循 语义搜索教程，我们的方法是 嵌入 每个文档分割的内容，并将这些嵌入插入到 Vector Store 中。给定输入查询，我们可以使用向量搜索来检索相关文档。

我们可以使用在 教程开始 选择的 Vector Store 和 Embeddings 模型在单个命令中嵌入和存储所有文档分割。

:::python

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']

:::

:::js

await vectorStore.addDocuments(allSplits);

:::

深入了解

Embeddings：围绕文本嵌入模型的包装器，用于将文本转换为嵌入。

• 集成：30+ 集成可供选择。
• @[接口][Embeddings]：基础接口的 API 参考。

VectorStore：围绕向量数据库的包装器，用于存储和查询嵌入。

• 集成：40+ 集成可供选择。
• 接口：基础接口的 API 参考。

这完成了管道的 索引 部分。此时我们有一个可查询的 Vector Store，包含博客文章的分块内容。给定用户问题，我们应该能够返回回答该问题的博客文章片段。

2. 检索和生成（Retrieval and Generation）

RAG 应用通常的工作流程如下：

1. 检索（Retrieve）：给定用户输入，使用 Retriever 从存储中检索相关分割。
2. 生成（Generate）：模型 使用包含问题和检索数据的提示生成答案。

现在让我们编写实际应用逻辑。我们希望创建一个简单的应用，接受用户问题，搜索与该问题相关的文档，将检索到的文档和初始问题传递给模型，并返回答案。

我们将演示：

1. 一个 RAG Agent，使用简单工具执行搜索。这是一个很好的通用实现。
2. 一个两步 RAG Chain，每个查询只使用一次 LLM 调用。这是简单查询的快速有效方法。

RAG Agents

RAG 应用的一种形式是作为具有检索信息工具的简单 Agent。我们可以通过实现包装 Vector Store 的 工具 来组装一个最小的 RAG Agent：

:::python

from langchain.tools import tool

@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """检索信息以帮助回答问题。"""
    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

:::js

import * as z from "zod";
import { tool } from "@langchain/core/tools";

const retrieveSchema = z.object({ query: z.string() });

const retrieve = tool(
  async ({ query }) => {
    const retrievedDocs = await vectorStore.similaritySearch(query, 2);
    const serialized = retrievedDocs
      .map(
        (doc) => `Source: ${doc.metadata.source}\nContent: ${doc.pageContent}`
      )
      .join("\n");
    return [serialized, retrievedDocs];
  },
  {
    name: "retrieve",
    description: "检索与查询相关的信息。",
    schema: retrieveSchema,
    responseFormat: "content_and_artifact",
  }
);

:::python 检索工具不限于单个字符串 query 参数，如上面的示例。你可以通过添加参数强制 LLM 指定额外的搜索参数——例如，一个类别：

```python
from typing import Literal

def retrieve_context(query: str, section: Literal["beginning", "middle", "end"]):
```

给定我们的工具，我们可以构建 Agent：

:::python

from langchain.agents import create_agent


tools = [retrieve_context]
# 如果需要，指定自定义指令
prompt = (
    "你可以使用一个从博客文章中检索上下文的工具。"
    "使用该工具帮助回答用户查询。"
)
agent = create_agent(model, tools, system_prompt=prompt)

:::

:::js

import { createAgent } from "langchain";

const tools = [retrieve];
const systemPrompt = new SystemMessage(
    "你可以使用一个从博客文章中检索上下文的工具。" +
    "使用该工具帮助回答用户查询。"
)

const agent = createAgent({ model: "gpt-5", tools, systemPrompt });

:::

让我们测试一下。我们构建一个通常需要迭代检索步骤序列才能回答的问题：

:::python

query = (
    "任务分解的标准方法是什么？\n\n"
    "得到答案后，查找该方法的常见扩展。"
)

for event in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    event["messages"][-1].pretty_print()

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

任务分解的标准方法是什么？

得到答案后，查找该方法的常见扩展。
================================== 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)...

:::

:::js

let inputMessage = `任务分解的标准方法是什么？
得到答案后，查找该方法的常见扩展。`;

let agentInputs = { messages: [{ role: "user", content: inputMessage }] };

const stream = await agent.stream(agentInputs, {
  streamMode: "values",
});
for await (const step of stream) {
  const lastMessage = step.messages[step.messages.length - 1];
  console.log(`[${lastMessage.role}]: ${lastMessage.content}`);
  console.log("-----\n");
}

[human]: 任务分解的标准方法是什么？
得到答案后，查找该方法的常见扩展。
-----

[ai]:
Tools:
- retrieve({"query":"standard method for Task Decomposition"})
-----

[tool]: Source: https://lilianweng.github.io/posts/2023-06-23-agent/
Content: hard tasks into smaller and simpler steps...
Source: https://lilianweng.github.io/posts/2023-06-23-agent/
Content: System message:Think step by step and reason yourself...
-----

[ai]:
Tools:
- retrieve({"query":"common extensions of Task Decomposition method"})
-----

[tool]: Source: https://lilianweng.github.io/posts/2023-06-23-agent/
Content: hard tasks into smaller and simpler steps...
Source: https://lilianweng.github.io/posts/2023-06-23-agent/
Content: be provided by other developers (as in Plugins) or self-defined...
-----

[ai]: ### 任务分解的标准方法

任务分解的标准方法涉及...
-----

:::

注意 Agent：

1. 生成查询以搜索任务分解的标准方法；
2. 收到答案后，生成第二个查询以搜索其常见扩展；
3. 收到所有必要的上下文后，回答问题。

我们可以在 LangSmith 跟踪 中看到完整的步骤序列，以及延迟和其他元数据。

RAG Chains

在上面的 Agentic RAG 表述中，我们允许 LLM 自行决定生成 工具调用 来帮助回答用户查询。这是一个很好的通用解决方案，但有一些权衡：

另一种常见的方法是两步 Chain，我们始终运行搜索（可能使用原始用户查询）并将结果作为单个 LLM 查询的上下文纳入。这导致每个查询一次推理调用，以灵活性为代价换取降低的延迟。

在这种方法中，我们不再循环调用模型，而是进行一次传递。

我们可以通过从 Agent 中移除工具并将检索步骤纳入自定义提示来实现这个 Chain：

:::python

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def prompt_with_context(request: ModelRequest) -> str:
    """将上下文注入状态消息。"""
    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 = (
        "你是一个有用的助手。在你的响应中使用以下上下文："
        f"\n\n{docs_content}"
    )

    return system_message


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

:::

:::js

import { createAgent, dynamicSystemPromptMiddleware } from "langchain";
import { SystemMessage } from "@langchain/core/messages";

const agent = createAgent({
  model,
  tools: [],
  middleware: [
    dynamicSystemPromptMiddleware(async (state) => {
        const lastQuery = state.messages[state.messages.length - 1].content;

        const retrievedDocs = await vectorStore.similaritySearch(lastQuery, 2);

        const docsContent = retrievedDocs
        .map((doc) => doc.pageContent)
        .join("\n\n");

        // 构建系统消息
        const systemMessage = new SystemMessage(
        `你是一个有用的助手。在你的响应中使用以下上下文：\n\n${docsContent}`
        );

        // 返回系统 + 现有消息
        return [systemMessage, ...state.messages];
    })
  ]
});

:::

让我们试试这个：

:::python

query = "什么是任务分解？"
for step in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()

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

什么是任务分解？
================================== Ai Message ==================================

Task decomposition is...

:::

:::js

let inputMessage = `什么是任务分解？`;

let chainInputs = { messages: [{ role: "user", content: inputMessage }] };

const stream = await agent.stream(chainInputs, {
  streamMode: "values",
})
for await (const step of stream) {
  const lastMessage = step.messages[step.messages.length - 1];
  prettyPrint(lastMessage);
  console.log("-----\n");
}

:::

在 LangSmith 跟踪 中，我们可以看到检索到的上下文被纳入模型提示中。

这是一种快速有效的方法，适用于受限环境中的简单查询，当我们通常确实想通过语义搜索运行用户查询以获取额外上下文时。

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"
            "使用以下上下文回答问题：\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()],
)

import { createMiddleware, Document, createAgent } from "langchain";
import { StateSchema, MessagesValue } from "@langchain/langgraph";
import { z } from "zod";

const CustomState = new StateSchema({
  messages: MessagesValue,
  context: z.array(z.custom<Document>()),
});

const retrieveDocumentsMiddleware = createMiddleware({
  stateSchema: CustomState,
  beforeModel: async (state) => {
    const lastMessage = state.messages[state.messages.length - 1].content;
    const retrievedDocs = await vectorStore.similaritySearch(lastMessage, 2);

    const docsContent = retrievedDocs
      .map((doc) => doc.pageContent)
      .join("\n\n");

    const augmentedMessageContent = [
        ...lastMessage.content,
        { type: "text", text: `使用以下上下文回答问题：\n\n${docsContent}` }
    ]

    // 下面我们用上下文增强每个输入消息，但我们也可能
    // 只修改系统消息，如前所述。
    return {
      messages: [{
        ...lastMessage,
        content: augmentedMessageContent,
      }]
      context: retrievedDocs,
    }
  },
});

const agent = createAgent({
  model,
  tools: [],
  middleware: [retrieveDocumentsMiddleware],
});