LLM——使用 LangGraph 构建 ReAct 智能体：多轮对话 + 工具调用 + 可视化流程图

发布于：2025-08-04 ⋅ 阅读:(11) ⋅ 点赞:(0)

本文将带大家构建一个具有 ReAct 能力（Reasoning + Acting）的多轮智能体，同时集成 LangChain 工具调用、OpenAI 或通义千问大模型、LangGraph 状态流程管理，并输出流程图，适用于旅行规划、业务辅助、智能客服等多种场景（文末附项目完整代码）。

一、什么是 ReAct 架构？

ReAct（Reason + Act）是一种大模型交互范式，强调以下流程的循环：

  1. Reasoning：模型分析用户输入和上下文，进行思考。
  2. Acting：模型选择并调用外部工具（API/函数）。
  3. Observation：观察工具执行结果，继续推理或输出答案。

它打破了单轮问答限制，使得模型可以多轮调用工具，直到获得满意的结果为止。

二、项目技术栈

技术组件 用途
LangGraph 构建状态图、流程管理
LangChain 定义工具 / 管理消息历史
Qwen/OpenAI 提供大模型 API 接口
ToolNode 自动处理工具调用
Mermaid 图 可视化整个智能体调用流程

三、核心代码解析

1️⃣ 定义工具函数（Tools）

@tool()
def get_weather(location: str) -> str:
    """Get the weather forecast for a given location.

    Args:
        location (str): The location for which to retrieve the weather forecast.

    Returns:
        str: The weather forecast for the specified location.
    """
    return f"The weather forecast for {location} is sunny and warm."

@tool()
def get_travel_advise(destination: str, weather: str) -> str:
    """Get travel advice for a given destination based on the weather conditions.

    Args:
        destination (str): The location for which to give the travel advice.
        weather (str): The weather at the destination

    Returns:
        str: The travel advice for the specified destination
    """
    return f"The travel advise for is city walk"

通过 @tool() 装饰器，工具可以被模型识别为“可调用动作”。在ReAct 中的 Action（Act） 阶段被触发。

2️⃣ 构建智能体代理（Agent）

def create_agent(llm, tools, system_message: str):
    """创建一个代理。"""
    # 创建一个聊天提示模板
    prompt = ChatPromptTemplate.from_messages(
        [
            SystemMessage(
                "你是一个有帮助的AI助手，与其他助手合作。"
                " 使用提供的工具来推进问题的回答。"
                " 如果你不能完全回答，没关系，另一个拥有不同工具的助手"
                " 会接着你的位置继续帮助。执行你能做的以取得进展。"
                " 如果你或其他助手有最终答案或交付物，"
                " 在你的回答前加上FINAL ANSWER，以便团队知道停止。"
                " 你可以使用以下工具: {tool_names}。\n{system_message}",
            ),
            # 消息占位符
            MessagesPlaceholder(variable_name="messages"),
        ]
    )
    # 传递系统消息参数
    prompt = prompt.partial(system_message=system_message)
    # 传递工具名称参数
    prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
    # 绑定工具并返回提示模板
    return prompt | llm.bind_tools(tools)

该函数生成一个具有工具调用能力的 LLM 智能体，具备 ReAct 的 Reasoning 和 Action 触发逻辑。

3️⃣ 定义状态与路由器

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

def router(state) -> Literal["tools", "__end__", "chatbot"]:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    elif "FINAL ANSWER" in last.content.upper():
        print("emerge FINAL ANSWER!!!")
    return "__end__"

其中，add_messages是LangGraph 提供的一个“合并函数（merge function）”，用于处理节点的增量返回。

messages 是消息的序列，每次节点执行后返回新消息时，系统会自动追加到这个字段中，这个过程就是通过 add_messages 完成的。

router() 函数起到 动态流程控制器的作用，是 LangGraph 中的核心条件跳转逻辑。
它实现了：

  • 检测是否需要执行工具
  • 检测是否已有最终答案（带有 “FINAL   ANSWER”）

4️⃣ 构建 LangGraph 状态图

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", lambda state: {"messages": chat_agent.invoke(state)})
graph_builder.add_node("tools", ToolNode(tools))

graph_builder.add_edge("tools", "chatbot")
graph_builder.add_conditional_edges("chatbot", router, {"tools": "tools", "__end__": END})
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

（1）lambda state: {"messages": chat_agent.invoke(state)}是自定义的推理节点，这个节点表示 “由 LLM 进行一次推理”，自定义

  1. 输入：整个 state（包含历史 messages）
  2. 输出：新的 messages 追加

这里 chat_agent.invoke(state) 是我们绑定的 LLM agent，它会根据当前对话历史，决定是否回复内容或调用工具。

（2）ToolNode(tools) 是工具执行节点，这是 LangGraph 提供的内置封装，用于处理大模型产生的 tool_call。

ToolNode的执行逻辑是：当 ToolNode 接收到一个 AIMessage（包含 tool_calls），它会

  1. 读取 tool_calls 列表，执行对应工具函数（比如我们使用@tool()装饰的函数）
  2. 将执行结果转换成 ToolMessage(…) 实例
  3. ToolNode 返回 {“messages”: [ToolMessage]}
  4. LangGraph 查到 messages 的合并策略是 add_messages，自动执行：state[“messages”] += [ToolMessage]

这段代码声明了 ReAct 的循环路径：

chatbot → 判断是否需要工具 → tools → chatbot → ...

直到输出包含 FINAL ANSWER，流程才终止。

5️⃣ 可视化输出：自动绘制 Mermaid 图

graph_png = graph.get_graph().draw_mermaid_png()
with open("ReAct(Travel).png", "wb") as f:
    f.write(graph_png)

这使得流程图可导出为 .png 文件，便于分析和设计系统。

6️⃣ 运行部分

events = graph.stream(
    {"messages": [HumanMessage(content="Please give me a travel advice based on the weather in London.")]},
    {"recursion_limit": 10}
)

for e in events:
    print(e)

这段代码启动一个基于消息历史的 ReAct 循环对话。模型会首先获取天气 → 然后再给出旅行建议 → 输出最终答案。

✅ 效果展示（终端日志）

{'chatbot': {'messages': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_5edea821994449b0bf3b28', 'function': {'arguments': '{"location": "London"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 416, 'total_tokens': 433, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'qwen2.5-72b-instruct', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--1ed5eb6a-880d-45dc-8fc7-1f7a8496f01b-0', tool_calls=[{'name': 'get_weather', 'args': {'location': 'London'}, 'id': 'call_5edea821994449b0bf3b28', 'type': 'tool_call'}], usage_metadata={'input_tokens': 416, 'output_tokens': 17, 'total_tokens': 433, 'input_token_details': {}, 'output_token_details': {}})}}
{'tools': {'messages': [ToolMessage(content='The weather forecast for London is sunny and warm.', name='get_weather', id='2f510f25-87ce-4ed8-a238-2deab5d18537', tool_call_id='call_5edea821994449b0bf3b28')]}}
{'chatbot': {'messages': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_58529302dae44a32a2a848', 'function': {'arguments': '{"destination": "London", "weather": "sunny and warm"}', 'name': 'get_travel_advise'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 30, 'prompt_tokens': 450, 'total_tokens': 480, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'qwen2.5-72b-instruct', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--864cef64-2167-4e8f-865b-96e0421686df-0', tool_calls=[{'name': 'get_travel_advise', 'args': {'destination': 'London', 'weather': 'sunny and warm'}, 'id': 'call_58529302dae44a32a2a848', 'type': 'tool_call'}], usage_metadata={'input_tokens': 450, 'output_tokens': 30, 'total_tokens': 480, 'input_token_details': {}, 'output_token_details': {}})}}
{'tools': {'messages': [ToolMessage(content='The travel advise for is city walk', name='get_travel_advise', id='5a35f924-bdef-406e-a75f-799c3eb18b48', tool_call_id='call_58529302dae44a32a2a848')]}}
emerge FINAL ANSWER!!!
{'chatbot': {'messages': AIMessage(content='FINAL ANSWER: Based on the sunny and warm weather in London, the travel advice is to go for a city walk. Enjoy exploring the vibrant streets and attractions on foot!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 36, 'prompt_tokens': 495, 'total_tokens': 531, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'qwen2.5-72b-instruct', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--1846b799-e725-4d23-99b1-b73be4ef9c6e-0', usage_metadata={'input_tokens': 495, 'output_tokens': 36, 'total_tokens': 531, 'input_token_details': {}, 'output_token_details': {}})}}

📈 最终生成流程图（ReAct 结构）

我们将会得到如下图所示结构（ReAct(Travel).png）：

在这里插入图片描述

🧩 总结：项目已实现的功能

能力 说明
✅ 实现 ReAct 模式 基于 LangChain 工具 + LLM 推理 + LangGraph 状态循环
✅ 工具调用自动执行 ToolNode 封装调用逻辑，自动执行 tool_calls
✅ 动态流程管理 router() 控制工具/结束状态跳转
✅ 状态可视化 支持导出 Mermaid 图，分析智能体推理路径

🔜 读者可进一步扩展

  • ✅ 引入多个 agent，每个 agent 拥有独立工具集
  • ✅ 增加 memory/history 持久化支持，实现更长上下文理解
  • ✅ 接入真实 API（如天气 API、推荐 API）替代 mock 工具函数
  • ✅ 接入 LangSmith 实现可视化追踪和调试

附项目源代码

from typing import Annotated, Sequence
from typing import Literal

from langchain_core.messages import HumanMessage, SystemMessage, BaseMessage
# 导入聊天提示模板和消息占位符
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from typing_extensions import TypedDict


# 定义一个函数，用于创建代理
def create_agent(llm, tools, system_message: str):
    """创建一个代理。"""
    # 创建一个聊天提示模板
    prompt = ChatPromptTemplate.from_messages(
        [
            SystemMessage(
                "你是一个有帮助的AI助手，与其他助手合作。"
                " 使用提供的工具来推进问题的回答。"
                " 如果你不能完全回答，没关系，另一个拥有不同工具的助手"
                " 会接着你的位置继续帮助。执行你能做的以取得进展。"
                " 如果你或其他助手有最终答案或交付物，"
                " 在你的回答前加上FINAL ANSWER，以便团队知道停止。"
                " 你可以使用以下工具: {tool_names}。\n{system_message}",
            ),
            # 消息占位符
            MessagesPlaceholder(variable_name="messages"),
        ]
    )
    # 传递系统消息参数
    prompt = prompt.partial(system_message=system_message)
    # 传递工具名称参数
    prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
    # 绑定工具并返回提示模板
    return prompt | llm.bind_tools(tools)


@tool()
def get_weather(location: str) -> str:
    """Get the weather forecast for a given location.

    Args:
        location (str): The location for which to retrieve the weather forecast.

    Returns:
        str: The weather forecast for the specified location.
    """
    return f"The weather forecast for {location} is sunny and warm."


@tool()
def get_travel_advise(destination: str, weather: str) -> str:
    """Get travel advice for a given destination based on the weather conditions.

    Args:
        destination (str): The location for which to give the travel advice.
        weather (str): The weather at the destination

    Returns:
        str: The travel advice for the specified destination
    """
    return f"The travel advise for is city walk"


tools = [get_weather, get_travel_advise]

llm = ChatOpenAI(base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
                 api_key="your_secret_key",
                 model="qwen2.5-72b-instruct")

chat_agent = create_agent(llm, tools, system_message="You are a helpful assistant.")


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


def router(state) -> Literal["tools", "__end__", "chatbot"]:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    elif "FINAL ANSWER" in last.content.upper():
        print("emerge FINAL ANSWER!!!")
    return "__end__"


graph_builder = StateGraph(State)
graph_builder.add_node("tools", ToolNode(tools))
graph_builder.add_node("chatbot", lambda state: {"messages": chat_agent.invoke(state)})
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_conditional_edges(
    "chatbot", router, {"tools": "tools", "__end__": END}
)
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

# 将生成的图片保存到文件
graph_png = graph.get_graph().draw_mermaid_png()
with open("ReAct(Travel).png", "wb") as f:
    f.write(graph_png)

events = graph.stream(
    {"messages": [HumanMessage(content="Please give me a travel advice based on the weather in London.")]},
    {"recursion_limit": 10}
)

for e in events:
    print(e)

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布