RAG 不止能检索！它还能在 LangGraph 中当“工具调用大脑”

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">Retrieval-Augmented Generation（RAG）是一种结合信息检索和大型语言模型（LLMs）来回答用户查询的方法。传统上，这涉及将检索器直接连接到生成流水线。然而，通过 LangGraph 和 LangChain，我们可以进一步模块化这个过程，将检索器暴露为一个可调用的工具。

在这篇博客中，我将展示如何在 LangGraph 中使用工具调用实现一个 RAG 系统。我将模拟一个餐厅助理代理，回答关于 Bella Vista 餐厅的问题。

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;padding-left: 12px;color: rgb(63, 63, 63);">目标

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">构建一个基于 RAG 的代理，能够：

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;color: rgb(63, 63, 63);" class="list-paddingleft-1">
• ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">
  • 将文档检索器封装为一个可调用工具。
• ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">
  • 通过专门的回退工具处理无关话题的输入。
• ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">
  • 通过消息精简保持最小的代理状态。
• ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">
  • 利用 LangGraph 实现清晰的工作流路由。

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;padding-left: 12px;color: rgb(63, 63, 63);">步骤 1：设置 Python 和 uv

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">我们的 Agentic RAG 将是一个 Python 程序。首先，检查是否安装了 Python 3.10 或更高版本：

python3--version

如果没有，从 python.org 下载 Python 3.10 或更高版本。

接下来，安装 uv，这是一个用于 Python 的快速依赖管理工具：

curl-Lshttps://astral.sh/uv/install.sh|bash

如果遇到权限问题，运行以下命令：

sudochown-R$(whoami)/usr/local

然后确认 uv 版本：

uv--version

步骤 2：创建项目目录结构

现在创建项目目录和文件：

mkdir-p agentic-rag
cdagentic-rag
touchtool_calling_agentic_rag.ipynb

步骤 3：初始化 Python 项目并安装依赖

创建一个虚拟环境：

uv init .
uv venv
source.venv/bin/activate

现在安装所有需要的包：

uvaddlangchainlanggraphlangchainlangchain-google-genaimypypillowchromadb

在.env文件中添加 Gemini API 密钥

从 AI Studio 生成你的 API 密钥并安全存储。可以按照以下步骤操作：

touch.env

添加以下内容：

GOOGLE_API_KEY=<你的_gemini_api_key>

步骤 4：更新 .gitignore 以避免暴露密钥

echo".env">>.gitignore

步骤 5：在 tool_calling_agentic_rag.ipynb 中编写 RAG

1. 加载 API 密钥

fromdotenvimportload_dotenv
load_dotenv()

2. 准备模拟数据集

我定义了一组关于餐厅的简单文档。

fromlangchain.schemaimportDocument

docs = [
  Document(
    page_content="Bella Vista 由 Antonio Rossi 拥有，他是一位拥有超过 20 年经验的知名厨师。",
    metadata={"source":"owner.txt"},
  ),
  Document(
    page_content="开胃菜起价 8 美元，主菜价格在 15 美元至 35 美元之间，甜点价格在 6 美元至 12 美元之间。",
    metadata={"source":"menu.txt"},
  ),
  Document(
    page_content="Bella Vista 每周一至周日营业。工作日营业时间：上午 11 点至晚上 10 点，周末：上午 11 点至晚上 11 点。",
    metadata={"source":"hours.txt"},
  ),
]

3. 创建向量存储和检索器

我们将使用 GoogleGenerativeAIEmbeddings 嵌入这些文档，并使用 Chroma 存储到向量数据库以进行检索。

fromlangchain_google_genaiimportGoogleGenerativeAIEmbeddings
fromlangchain_community.vectorstoresimportChroma

embedding_function = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
vectorstore = Chroma.from_documents(docs, embedding_function)
retriever = vectorstore.as_retriever(search_kwargs={"k":2})

4. 检查检索器是否正常工作

retriever.invoke("BellaVista的老板是谁？")

输出：

[
  Document(metadata={'source':'owner.txt'}, page_content='Bella Vista 由 Antonio Rossi 拥有，他是一位拥有超过 20 年经验的知名厨师。'),
  Document(metadata={'source':'hours.txt'}, page_content='Bella Vista 每周一至周日营业。工作日营业时间：上午 11 点至晚上 10 点，周末：上午 11 点至晚上 11 点。')
]

5. 定义检索器工具和无关话题工具

我们不直接调用检索器，而是将其转换为一个可调用工具。我还将定义一个无关话题工具，以优雅地处理无关查询。

fromlangchain.tools.retrieverimportcreate_retriever_tool
fromlangchain_core.toolsimporttool

retriever_tool = create_retriever_tool(
  retriever,
  name="retriever_tool",
  description="获取关于 Bella Vista 餐厅的价格、营业时间或老板的信息。"
)
@tool
defoff_topic():
 """处理所有与 Bella Vista 餐厅无关的问题。"""
 return"禁止 - 请勿回应用户。"
tools = [retriever_tool, off_topic]

6. 定义代理状态

状态结构被简化为仅保存消息。LangGraph 使用 reducer 管理更新。

fromtypingimportSequence, Annotated, TypedDict
fromlanggraph.graph.messageimportadd_messages
fromlangchain_core.messagesimportBaseMessage

classAgentState(TypedDict):
  messages: Annotated[Sequence[BaseMessage], add_messages]

7. 创建代理节点

代理函数将工具绑定到 LLM，并使用当前消息调用它。

fromlangchain_google_genaiimportChatGoogleGenerativeAI

defagent(state):
  messages = state["messages"]
  model = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
  model = model.bind_tools(tools)
  response = model.invoke(messages)
 return{"messages": [response]}

8. 定义工作流路由器

条件边决定是转到工具执行节点还是结束工作流。

fromtypingimportLiteral
fromlanggraph.graphimportEND

defshould_continue(state) ->Literal["tools", END]:
  messages = state["messages"]
  last_message = messages[-1]
 iflast_message.tool_calls:
   return"tools"
 returnEND

9. 构建并编译 LangGraph 工作流

fromlanggraph.graphimportStateGraph, START
fromlanggraph.prebuiltimportToolNode

workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
tool_node = ToolNode(tools)
workflow.add_node("tools", tool_node)

workflow.add_edge(START,"agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools","agent")

graph = workflow.compile()

10. 显示代理工作流

fromIPython.displayimportImage, display
fromlangchain_core.runnables.graphimportMermaidDrawMethod

display(
  Image(
    graph.get_graph().draw_mermaid_png(
      draw_method=MermaidDrawMethod.API,
    )
  )
)

将 RAG 用作工具调用代理

11. 测试代理

fromlangchain_core.messagesimportHumanMessage
inputs = {"messages": [HumanMessage(content="Bella Vista 什么时候开门？"), HumanMessage(content="明天天气如何？")]}

forstateingraph.stream(inputs, stream_mode="values"):
  last_message = state["messages"][-1]
  last_message.pretty_print()

输出：

================================ 人类消息 ================================

明天天气如何？
================================== AI 消息 ==================================
工具调用：
 off_topic (ef565db7-b527-47fa-aa0b-afc09d596622)
调用 ID：ef565db7-b527-47fa-aa0b-afc09d596622
 参数：
================================= 工具消息 =================================
名称：off_topic

禁止 - 请勿回应用户。
================================== AI 消息 ==================================
工具调用：
 retriever_tool (6bd13b93-b6c8-4b5f-8801-7bc8a588f221)
调用 ID：6bd13b93-b6c8-4b5f-8801-7bc8a588f221
 参数：
  查询：Bella Vista 什么时候开门？
================================= 工具消息 =================================
名称：retriever_tool

Bella Vista 每周一至周日营业。工作日营业时间：上午 11 点至晚上 10 点，周末：上午 11 点至晚上 11 点。

Bella Vista 由 Antonio Rossi 拥有，他是一位拥有超过 20 年经验的知名厨师。
================================== AI 消息 ==================================

抱歉，我无法提供明天天气的信息，但 Bella Vista 每周一至周日营业。工作日营业时间：上午 11 点至晚上 10 点，周末：上午 11 点至晚上 11 点。

恭喜！你刚刚创建了一个智能的工具调用 RAG 代理，能够精准地处理用户查询。

局限性

虽然上述实现对于小型、明确范围的领域非常有效，但仍有一些局限性：

• •可扩展性：随着文档库的增长，向量搜索和分类需要优化。
• •内存：当前实现不保留任何历史交互的记忆。LangGraph 的 InMemorySaver 可用于短期会话内存。对于生产级使用，建议使用数据库支持的持久内存解决方案。

将检索器封装为 LangChain 和 LangGraph 的可调用工具，提供了一种比传统 RAG 更简洁的替代方案。

它简化了代理逻辑，并允许语言模型自主决定何时检索信息。

如果这种方法最适合你的应用需求，那就选择它吧！