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

Retrieval-Augmented Generation（RAG）是一种结合信息检索和大型语言模型（LLMs）来回答用户查询的方法。传统上，这涉及将检索器直接连接到生成流水线。然而，通过 LangGraph 和 LangChain，我们可以进一步模块化这个过程，将检索器暴露为一个可调用的工具。

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

目标

构建一个基于 RAG 的代理，能够：

• • 将文档检索器封装为一个可调用工具。
• • 通过专门的回退工具处理无关话题的输入。
• • 通过消息精简保持最小的代理状态。
• • 利用 LangGraph 实现清晰的工作流路由。

步骤 1：设置 Python 和 uv

我们的 Agentic RAG 将是一个 Python 程序。首先，检查是否安装了 Python 3.10 或更高版本：

python3 --version

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

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

curl -Ls https://astral.sh/uv/install.sh | bash

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

sudo chown -R $(whoami) /usr/local

然后确认 uv 版本：

uv --version

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

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

mkdir -p agentic-rag
cd agentic-rag
touch tool_calling_agentic_rag.ipynb

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

创建一个虚拟环境：

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

现在安装所有需要的包：

uv add langchain langgraph langchain langchain-google-genai mypy pillow chromadb

在 .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 密钥

from dotenv import load_dotenv
load_dotenv()

2. 准备模拟数据集

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

from langchain.schema import Document

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 存储到向量数据库以进行检索。

from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import Chroma

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("Bella Vista 的老板是谁？")

输出：

[
    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. 定义检索器工具和无关话题工具

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

from langchain.tools.retriever import create_retriever_tool
from langchain_core.tools import tool

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

6. 定义代理状态

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

from typing import Sequence, Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

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

7. 创建代理节点

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

from langchain_google_genai import ChatGoogleGenerativeAI

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

8. 定义工作流路由器

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

from typing import Literal
from langgraph.graph import END

def should_continue(state) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

9. 构建并编译 LangGraph 工作流

from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode

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. 显示代理工作流

from IPython.display import Image, display
from langchain_core.runnables.graph import MermaidDrawMethod

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

将 RAG 用作工具调用代理

11. 测试代理

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

for state in graph.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 更简洁的替代方案。

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

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