使用 LangGraph 构建 ReAct Agent

发表于 分类于 ， 本文字数： 1.2k 阅读时长 ≈ 4 分钟

上一篇文章中我们已经实现了一个简单的对话 Agent，那么接下来让我们为该 Agent 添加工具调用（Tool Calling）功能，从而让它成为一个 ReAct Agent。

基本概念

在上一篇文章中，我们已经描述了 LangGraph 的基本概念，这里再补充几个：

  1. Tools 是专门的函数或实用工具，节点可以用它们来执行特定任务，比如从 API 获取数据这类操作。它们的作用是增强节点的能力，给节点额外的功能支持。
  2. ToolNode 是一种特殊的节点，它的主要工作就是运行工具。它会把工具运行产生的输出连接回状态里，这样其他节点就可以使用这些信息。
  3. Conditional Edges 是一种特殊的连接关系，它的作用是依据应用当前状态里的特定条件或逻辑，来决定接下来要执行哪个节点。

创建工具

让我们来创建一个搜索工具方法，该方法会利用搜索引擎获取信息。

Tavily

使用 Tavily 首先要去官网注册账号从而获取 API Key，免费账户一个月可以调用 1000 次搜索，完全够用。此外还需要安装 langchain-tavily 包。

1
2
3
4
5
from langchain_tavily import TavilySearch

# 记得在环境中导入 tavily 的 api key
# os.environ["TAVILY_API_KEY"] = tavily_api_key
search_tool = TavilySearch(max_results=3)

DDGS

还有一种方式是使用 DDGS 框架，这里我们需要自行封装工具方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langchain_core.tools import tool
from ddgs import DDGS

@tool
def ddgs_text_search(
    query: str,
    region: str = "us-en",
    safesearch: str = "moderate",
    timelimit: str | None = None,
    max_results: int | None = 10,
    page: int = 1,
    backend: str = "auto",
):
    """
    Use this tool to search the web for text information.
    """
    return DDGS().text(
        query,
        region=region,
        safesearch=safesearch,
        timelimit=timelimit,
        max_results=max_results,
        page=page,
        backend=backend,
    )

根据 PyPI 上的描述，duckduckgo-search 包已经被重命名为 ddgs，所以不要参考 LangChain 的官方文档 进行操作。截至发布博客时，LangChain 的 DuckDuckGo Search 工具并不支持 ddgs。

绑定工具

然后我们需要为创建的 LLM 对象绑定工具：

1
2
tools = [search_tool] # [ddgs_text_search]
llm_with_tools = llm.bind_tools(tools=tools)

创建流程

和之前简单的顺序流程不同，我们希望模型能够自行选择是否需要调用工具，因此这里需要使用条件选择。我们先来创建一个条件选择方法：

1
2
3
4
5
6
7
def should_continue(state: MessagesState):
    messages = state["messages"]
    last_message = messages[-1]
    if not last_message.tool_calls:
        return "end" # edge name
    else:
        return "continue" # edge name

上述代码中，tool_calls 是一个列表，其意思是当模型觉得不需要使用工具时，就返回 "end"，代表后续走名为 "end" 的路径；否则返回 "continue"，代表后续走名为 "continue" 的路径。因此我们会这样添加一个条件判断路径：

1
2
3
4
5
6
7
8
workflow.add_conditional_edges(
    "chatbot", # start node name
    should_continue, # condition method
    {
        "continue": "tools", # edge name: node name
        "end": END,
    },
)

对于上述这样较为简单的条件判断，LangGraph 中提供了 tools_condition 用于简化代码。所以在基于之前 ChatAgent 的流程基础上，ToolCallAgent 的流程实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langgraph.graph import START
from langgraph.prebuilt import ToolNode, tools_condition

workflow.add_node("chatbot", call_model)

# 工具节点
tool_node = ToolNode(tools)
workflow.add_node("tools", tool_node)

# 条件判断，用 tools_condition 简化代码
workflow.add_conditional_edges("chatbot", tools_condition)

workflow.add_edge(START, "chatbot")
workflow.add_edge("tools", "chatbot")

graph = workflow.compile(checkpointer=memory)

我们可以将上述流程转换为图像查看。

Tool Call Agent

对话测试

让我们修改 chat 方法，来查看 Agent 是否真的调用了搜索工具。

1
2
3
4
5
6
7
8
9
from langchain_core.messages import AIMessage, HumanMessage

def chat(query: str):
    input_message: list[HumanMessage] = [HumanMessage(content=query)]
    for event in graph.stream({"messages": input_message}, checkpoint_config):
        for value in event.values():
            print(value["messages"][-1])

# ...

接着我们运行程序，向模型询问 “What is Anthropic?”，然后在模型的输出中我们应该能看到工具调用的搜索结果。

封装

最后，让我们重新把上述零散的代码封装成 ToolCallAgent 类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.prebuilt import ToolNode, tools_condition
import uuid
from functools import cached_property
from langchain_tavily import TavilySearch


class ToolCallAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="gemini-2.5-flash",
            base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
            api_key = api_key,
        )
        self.workflow = StateGraph(state_schema=MessagesState)
        self.memory = InMemorySaver()
        self.checkpoint_config = {"configurable": {"thread_id": uuid.uuid4().hex}}

        self.tools = [TavilySearch(max_results=3)]
        self.llm_with_tools = self.llm.bind_tools(tools=self.tools)

    def call_model(self, state: MessagesState):
        response = self.llm_with_tools.invoke(state["messages"])
        return {"messages": [response]}

    @cached_property
    def graph(self):
        self.workflow.add_node("chatbot", self.call_model)

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

        self.workflow.add_conditional_edges("chatbot", tools_condition)

        self.workflow.add_edge(START, "chatbot")
        self.workflow.add_edge("tools", "chatbot")

        return self.workflow.compile(checkpointer=self.memory)

    def chat(self, query: str):
        input_message: list[HumanMessage] = [HumanMessage(content=query)]
        for event in self.graph.stream({"messages": input_message}, self.checkpoint_config):
            for value in event.values():
                if isinstance(value["messages"][-1], AIMessage):
                    print(value["messages"][-1].content)


if __name__ == "__main__":
    agent = ToolCallAgent()

    while True:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            break
        agent.chat(user_input)

更简单的实现

LangGraph 提供了 create_react_agent 方法，可以更简单的创建 ReAct Agent。

1
2
3
4
from langgraph.prebuilt import create_react_agent

prompt = "You are a helpful assistant."
react_agent = create_react_agent(llm, tools, prompt=prompt, checkpointer=memory)

然后调用方式就和上面实现的 chat 方法一样：react_agent.stream({"messages": input_message}, checkpoint_config)。

参考

Add tools