使用 LangGraph 构建自定义 RAG Agent

概述

在本教程中，我们将使用 LangGraph 构建一个检索 (retrieval) Agent。 LangChain 提供了内置的 Agent 实现，这些实现是使用 LangGraph 原语实现的。如果需要更深度的定制，可以在 LangGraph 中直接实现 Agent。本指南展示了一个检索 Agent 的示例实现。检索 (Retrieval) Agent 在您希望 LLM 决定是从向量存储中检索上下文还是直接回复用户时非常有用。在本教程结束时，我们将完成以下工作：

获取并预处理用于检索的文档。
为语义搜索索引这些文档，并为 Agent 创建一个检索器工具。
构建一个能够决定何时使用检索器工具的 Agentic RAG 系统。

概念

我们将涵盖以下概念：

使用文档加载器、文本分割器、Embeddings 和向量存储进行检索
LangGraph Graph API，包括状态、节点、边和条件边。

设置

让我们下载所需的包并设置我们的 API 密钥：

npm install @langchain/langgraph @langchain/openai @langchain/community @langchain/textsplitters

1. 预处理文档

获取要在我们的 RAG 系统中使用的文档。我们将使用 Lilian Weng 精彩博客中的三个最新页面。我们将从使用 `CheerioWebBaseLoader` 获取页面内容开始：

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

const urls = [
  "https://lilianweng.github.io/posts/2023-06-23-agent/",
  "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
  "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
];

const docs = await Promise.all(
  urls.map((url) => new CheerioWebBaseLoader(url).load()),
);

将获取的文档分割成更小的块，以便索引到我们的向量存储中：

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

const docsList = docs.flat();

const textSplitter = new RecursiveCharacterTextSplitter({
  chunkSize: 500,
  chunkOverlap: 50,
});
const docSplits = await textSplitter.splitDocuments(docsList);

2. 创建检索器工具

现在我们有了分割后的文档，我们可以将它们索引到一个向量存储中，用于语义搜索。

使用内存向量存储和 OpenAI Embeddings：

import { MemoryVectorStore } from "@langchain/classic/vectorstores/memory";
import { OpenAIEmbeddings } from "@langchain/openai";

const vectorStore = await MemoryVectorStore.fromDocuments(
  docSplits,
  new OpenAIEmbeddings(),
);

const retriever = vectorStore.asRetriever();

使用 LangChain 预构建的 `createRetrieverTool` 创建一个检索器工具：

import { createRetrieverTool } from "@langchain/classic/tools/retriever";

const tool = createRetrieverTool(
  retriever,
  {
    name: "retrieve_blog_posts",
    description:
      "Search and return information about Lilian Weng blog posts on LLM agents, prompt engineering, and adversarial attacks on LLMs.",
  },
);
const tools = [tool];

3. 生成查询

现在我们将开始为我们的 Agentic RAG 图构建组件（节点和边）。

构建一个 `generateQueryOrRespond` 节点。它将调用 LLM 根据当前图状态（消息列表）生成响应。给定输入消息，它将决定使用检索器工具检索，还是直接回复用户。请注意，我们通过 `.bindTools` 让聊天模型访问我们之前创建的 `tools`：

import { ChatOpenAI } from "@langchain/openai";
import { GraphNode } from "@langchain/langgraph";

const generateQueryOrRespond: GraphNode<typeof State> = async (state) => {
  const model = new ChatOpenAI({
    model: "gpt-4.1",
    temperature: 0,
  }).bindTools(tools);

  const response = await model.invoke(state.messages);
  return {
    messages: [response],
  };
}

在随机输入上尝试：

import { HumanMessage } from "@langchain/core/messages";

const input = { messages: [new HumanMessage("hello!")] };
const result = await generateQueryOrRespond(input);
console.log(result.messages[0]);

输出:

AIMessage {
  content: "Hello! How can I help you today?",
  tool_calls: []
}

问一个需要语义搜索的问题：

const input = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?")
  ]
};
const result = await generateQueryOrRespond(input);
console.log(result.messages[0]);

输出:

AIMessage {
  content: "",
  tool_calls: [
    {
      name: "retrieve_blog_posts",
      args: { query: "types of reward hacking" },
      id: "call_...",
      type: "tool_call"
    }
  ]
}

4. 评分文档

添加一个节点 —— `gradeDocuments` —— 以确定检索到的文档是否与问题相关。我们将使用一个使用 Zod 进行结构化输出的模型来进行文档评分。我们还将添加一个条件边 —— `checkRelevance` —— 它检查评分结果并返回要转到的节点名称（`generate` 或 `rewrite`）：

import * as z from "zod";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";
import { GraphNode } from "@langchain/langgraph";
import { AIMessage } from "@langchain/core/messages";

const prompt = ChatPromptTemplate.fromTemplate(
  \`You are a grader assessing relevance of retrieved docs to a user question.
  Here are the retrieved docs:
  \n ------- \n
  {context}
  \n ------- \n
  Here is the user question: {question}
  If the content of the docs are relevant to the users question, score them as relevant.
  Give a binary score 'yes' or 'no' score to indicate whether the docs are relevant to the question.
  Yes: The docs are relevant to the question.
  No: The docs are not relevant to the question.\`,
);

const gradeDocumentsSchema = z.object({
  binaryScore: z.string().describe("Relevance score 'yes' or 'no'"),  // [!code highlight]
})

const gradeDocuments: GraphNode<typeof State> = async (state) => {
  const model = new ChatOpenAI({
    model: "gpt-4.1",
    temperature: 0,
  }).withStructuredOutput(gradeDocumentsSchema);

  const score = await prompt.pipe(model).invoke({
    question: state.messages.at(0)?.content,
    context: state.messages.at(-1)?.content,
  });

  if (score.binaryScore === "yes") {
    return "generate";
  }
  return "rewrite";
}

在工具响应中使用不相关的文档运行此操作：

import { ToolMessage } from "@langchain/core/messages";

const input = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?"),
    new AIMessage({
      tool_calls: [
        {
          type: "tool_call",
          name: "retrieve_blog_posts",
          args: { query: "types of reward hacking" },
          id: "1",
        }
      ]
    }),
    new ToolMessage({
      content: "meow",
      tool_call_id: "1",
    })
  ]
}
const result = await gradeDocuments(input);

确认相关文档被分类为相关：

const input = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?"),
    new AIMessage({
      tool_calls: [
        {
          type: "tool_call",
          name: "retrieve_blog_posts",
          args: { query: "types of reward hacking" },
          id: "1",
        }
      ]
    }),
    new ToolMessage({
      content: "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
      tool_call_id: "1",
    })
  ]
}
const result = await gradeDocuments(input);

5. 重写问题

构建 `rewrite` 节点。检索器工具可能会返回可能不相关的文档，这表明需要改进原始用户问题。为此，我们将调用 `rewrite` 节点：

import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";
import { GraphNode } from "@langchain/langgraph";

const rewritePrompt = ChatPromptTemplate.fromTemplate(
  \`Look at the input and try to reason about the underlying semantic intent / meaning. \n
  Here is the initial question:
  \n ------- \n
  {question}
  \n ------- \n
  Formulate an improved question:\`,
);

const rewrite: GraphNode<typeof State> = async (state) => {
  const question = state.messages.at(0)?.content;

  const model = new ChatOpenAI({
    model: "gpt-4.1",
    temperature: 0,
  });

  const response = await rewritePrompt.pipe(model).invoke({ question });
  return {
    messages: [response],
  };
}

尝试一下：

import { HumanMessage, AIMessage, ToolMessage } from "@langchain/core/messages";

const input = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?"),
    new AIMessage({
      content: "",
      tool_calls: [
        {
          id: "1",
          name: "retrieve_blog_posts",
          args: { query: "types of reward hacking" },
          type: "tool_call"
        }
      ]
    }),
    new ToolMessage({ content: "meow", tool_call_id: "1" })
  ]
};

const response = await rewrite(input);
console.log(response.messages[0].content);

输出:

What are the different types of reward hacking described by Lilian Weng, and how does she explain them?

6. 生成答案

构建 `generate` 节点：如果我们通过了评分员检查，我们可以根据原始问题和检索到的上下文生成最终答案：

import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";
import { GraphNode } from "@langchain/langgraph";

const generate: GraphNode<typeof State> = async (state) => {
  const question = state.messages.at(0)?.content;
  const context = state.messages.at(-1)?.content;

  const prompt = ChatPromptTemplate.fromTemplate(
  \`You are an assistant for question-answering tasks.
      Use the following pieces of retrieved context to answer the question.
      If you don't know the answer, just say that you don't know.
      Use three sentences maximum and keep the answer concise.
      Question: {question}
      Context: {context}\`
  );

  const llm = new ChatOpenAI({
    model: "gpt-4.1",
    temperature: 0,
  });

  const ragChain = prompt.pipe(llm);

  const response = await ragChain.invoke({
    context,
    question,
  });

  return {
    messages: [response],
  };
}

尝试一下：

import { HumanMessage, AIMessage, ToolMessage } from "@langchain/core/messages";

const input = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?"),
    new AIMessage({
      content: "",
      tool_calls: [
        {
          id: "1",
          name: "retrieve_blog_posts",
          args: { query: "types of reward hacking" },
          type: "tool_call"
        }
      ]
    }),
    new ToolMessage({
      content: "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
      tool_call_id: "1"
    })
  ]
};

const response = await generate(input);
console.log(response.messages[0].content);

输出:

Lilian Weng categorizes reward hacking into two types: environment or goal misspecification, and reward tampering. She considers reward hacking as a broad concept that includes both of these categories. Reward hacking occurs when an agent exploits flaws or ambiguities in the reward function to achieve high rewards without performing the intended behaviors.

7. 组装图

现在我们将所有节点和边组装成一个完整的图：

从 `generateQueryOrRespond` 开始，并确定是否需要调用检索器工具
使用条件边路由到下一步：
- 如果 `generateQueryOrRespond` 返回 `tool_calls`，则调用检索器工具以检索上下文
- 否则，直接回复用户
对检索到的文档内容与问题的相关性进行评分 (`gradeDocuments`) 并路由到下一步：
- 如果不相关，则使用 `rewrite` 重写问题，然后再次调用 `generateQueryOrRespond`
- 如果相关，则继续进行 `generate`，并使用带有检索到的文档上下文的 `ToolMessage` 生成最终响应

import { StateGraph, START, END, ConditionalEdgeRouter } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { AIMessage } from "langchain";

// Create a ToolNode for the retriever
const toolNode = new ToolNode(tools);

// Helper function to determine if we should retrieve
const shouldRetrieve: ConditionalEdgeRouter<typeof State, "retrieve"> = (state) => {
  const lastMessage = state.messages.at(-1);
  if (AIMessage.isInstance(lastMessage) && lastMessage.tool_calls.length) {
    return "retrieve";
  }
  return END;
}

// Define the graph
const builder = new StateGraph(GraphState)
  .addNode("generateQueryOrRespond", generateQueryOrRespond)
  .addNode("retrieve", toolNode)
  .addNode("gradeDocuments", gradeDocuments)
  .addNode("rewrite", rewrite)
  .addNode("generate", generate)
  // Add edges
  .addEdge(START, "generateQueryOrRespond")
  // Decide whether to retrieve
  .addConditionalEdges("generateQueryOrRespond", shouldRetrieve)
  .addEdge("retrieve", "gradeDocuments")
  // Edges taken after grading documents
  .addConditionalEdges(
    "gradeDocuments",
    // Route based on grading decision
    (state) => {
      // The gradeDocuments function returns either "generate" or "rewrite"
      const lastMessage = state.messages.at(-1);
      return lastMessage.content === "generate" ? "generate" : "rewrite";
    }
  )
  .addEdge("generate", END)
  .addEdge("rewrite", "generateQueryOrRespond");

// Compile
const graph = builder.compile();

8. 运行 Agentic RAG

现在让我们通过运行一个问题来测试完整的图：

import { HumanMessage } from "@langchain/core/messages";

const inputs = {
  messages: [
    new HumanMessage("What does Lilian Weng say about types of reward hacking?")
  ]
};

for await (const output of await graph.stream(inputs)) {
  for (const [key, value] of Object.entries(output)) {
    const lastMsg = output[key].messages[output[key].messages.length - 1];
    console.log(\`Output from node: '\${key}'\`);
    console.log({
      type: lastMsg._getType(),
      content: lastMsg.content,
      tool_calls: lastMsg.tool_calls,
    });
    console.log("---\n");
  }
}

输出:

Output from node: 'generateQueryOrRespond'
{
  type: 'ai',
  content: '',
  tool_calls: [
    {
      name: 'retrieve_blog_posts',
      args: { query: 'types of reward hacking' },
      id: 'call_...',
      type: 'tool_call'
    }
  ]
}
---

Output from node: 'retrieve'
{
  type: 'tool',
  content: '(Note: Some work defines reward tampering as a distinct category...\n' +
    'At a high level, reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering.\n' +
    '...',
  tool_calls: undefined
}
---

Output from node: 'generate'
{
  type: 'ai',
  content: 'Lilian Weng categorizes reward hacking into two types: environment or goal misspecification, and reward tampering. She considers reward hacking as a broad concept that includes both of these categories. Reward hacking occurs when an agent exploits flaws or ambiguities in the reward function to achieve high rewards without performing the intended behaviors.',
  tool_calls: []
}
---

在 GitHub 上编辑此页面或提交 issue。

将本文档连接到 Claude、VSCode 等，通过 MCP 获取实时解答。

Tutorials

Conceptual overviews

LangChain Academy

Additional resources

概述

概念

设置

1. 预处理文档

2. 创建检索器工具

3. 生成查询

4. 评分文档

5. 重写问题

6. 生成答案

7. 组装图

8. 运行 Agentic RAG

Tutorials

Conceptual overviews

LangChain Academy

Additional resources

​概述

​概念

​设置

​1. 预处理文档

​2. 创建检索器工具

​3. 生成查询

​4. 评分文档

​5. 重写问题

​6. 生成答案

​7. 组装图

​8. 运行 Agentic RAG

概述

概念

设置

1. 预处理文档

2. 创建检索器工具

3. 生成查询

4. 评分文档

5. 重写问题

6. 生成答案

7. 组装图

8. 运行 Agentic RAG