文章目录

前言

通过前面两篇博客，我们已经掌握了 LangGraph 的核心机制：状态、节点、边、条件路由、检查点持久化，以及人机交互。这些能力共同构成了一个“活的”流程引擎。但真正让智能体能干实事的，是它调用外部工具的能力——搜索最新信息、执行代码、操作数据库、发送请求……以及将这些执行过程实时、流式地展现给用户。

今天这篇，我们就动手构建一个生产级的 LangGraph Agent。它会：

  • 使用真正的 LLM（如 GPT-4）进行决策
  • 调用搜索和计算两个自定义工具
  • 形成“思考 → 行动 → 观察”的 ReAct 循环
  • 以 token 级别的流式输出向前端实时推送
  • 最后，我还会给出将其部署为 API 并添加可观测性的完整思路

1. Agent 架构总览

我们要实现的 Agent 核心循环如下：

用户输入 → agent（LLM 决策）
              │
              ├─ 需要工具？─→ tools（执行工具）─→ 工具结果返回 agent
              │
              └─ 直接回答 ─→ 结束

在 LangGraph 中，这对应一张包含两个主要节点的图：agent 节点（负责调用 LLM）和 tools 节点（负责执行工具）。agent 节点后接一条条件边，检查 LLM 的响应中是否包含工具调用请求；如果有，就路由到 tools 节点，tools 节点执行完毕后再回到 agent 节点继续思考，直到模型认为可以给出最终答案。

整个过程中，所有对话历史和工具调用记录都存放在我们熟悉的状态 messages 列表中。下面我们逐步把这张图实现出来。

2. 准备工作：环境与工具定义

首先安装必要的包：

pip install langgraph langchain langchain-openai langchain-community tavily-python

我们假定你已经有 OpenAI API Key 和 Tavily Search API Key（用于网络搜索工具）。在实际项目中，你也可以换成任何其他工具。

先定义我们 Agent 将要使用的两个工具：一个网络搜索工具，一个简单计算器。

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
import math

# 搜索工具
search = TavilySearchResults(max_results=2)

# 计算器工具
@tool
def calculator(expression: str) -> str:
    """计算一个数学表达式，例如 '2+3'，返回计算结果。"""
    try:
        # 安全地计算数学表达式（仅允许数字和运算符）
        allowed = set("0123456789+-*/(). ")
        if not all(c in allowed for c in expression):
            return "表达式包含非法字符"
        return str(eval(expression))
    except Exception as e:
        return f"计算出错: {str(e)}"

# 工具列表，稍后绑定到 LLM
tools = [search, calculator]

3. 定义 Agent 的状态和 LLM 节点

状态依然以 messages 为核心，它用 add_messages 合并，能自动记录完整的对话和工具调用历史。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

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

接下来定义 agent 节点。这个节点会绑定工具，并调用 LLM。

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
import os

# 初始化 LLM，确保它可以请求调用工具
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# agent 节点函数
def agent_node(state: AgentState) -> dict:
    # 调用 LLM，传入当前消息历史
    response = llm_with_tools.invoke(state["messages"])
    # 返回 AI 消息（可能包含 tool_calls）
    return {"messages": [response]}

这里的 bind_tools 是 LangChain 提供的便捷方法，它告诉 OpenAI 模型可用哪些工具，并让模型在需要时返回 tool_calls。agent_node 接收当前状态，调用 LLM，并将返回的 AI 消息追加到消息列表。如果 LLM 决定调用工具，那么这条 AI 消息里就会包含 tool_calls 信息。

4. 构建工具调用循环

工具调用循环的奥妙在于条件边。我们需要一个路由函数来判断 LLM 的最后一条消息是否包含工具调用。

from langgraph.graph import END

# 路由函数：检查最后一条消息是否请求了工具调用
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    # 如果有 tool_calls 属性且非空，说明需要调用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

然后，我们需要一个能执行工具的节点。LangGraph 提供了一个极其方便的预置节点 ToolNode，它接收包含工具调用的 AIMessage，自动执行相应的工具，并将结果包装成 ToolMessage 追加回消息列表。

tool_node = ToolNode(tools)

现在，把节点和边拼装成图：

from langgraph.graph import StateGraph

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

# 设置入口
workflow.set_entry_point("agent")

# 添加条件边：agent 执行后，根据消息内容决定走向
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",   # 需要工具 → 去 tools 节点
        "end": END          # 直接回答 → 结束
    }
)

# tools 节点执行完毕后，总是回到 agent 继续思考
workflow.add_edge("tools", "agent")

# 编译
app = workflow.compile()

至此，一个标准的工具调用 Agent 图就构建完成了。你可以像之前一样用 invoke 运行它：

initial_state = {"messages": [HumanMessage(content="今天北京天气怎么样？然后帮我算一下 123 * 456 等于多少")]}
result = app.invoke(initial_state)
print(result["messages"][-1].content)

这个图会经历：agent（LLM 想查天气）→ tools（执行搜索）→ agent（拿到天气，又想算乘法）→ tools（执行计算器）→ agent（综合结果，生成最终答案）。一切都在状态 messages 中自动记录。

5. 流式输出：让用户看见 Agent 的“思考过程”

上面的 invoke 是阻塞式一次性返回最终结果，用户无法感知中间步骤。在生产级应用中，我们通常希望逐步推送执行进度：模型正在生成哪个 token、正在调用哪个工具、工具执行结果是什么……这能极大提升用户体验，也便于调试。

LangGraph 提供了多种流式方法，最常用的是 stream 和 astream_events。

5.1 按节点级别流式输出：stream

stream 会在每个节点执行完毕后，输出该节点产生的状态更新。你可以用它的输出来构造“当前正在执行哪个步骤”的 UI 提示。

for chunk in app.stream(initial_state, stream_mode="updates"):
    # chunk 是一个字典，键为节点名，值为该节点返回的状态更新
    node_name = list(chunk.keys())[0]
    update = chunk[node_name]
    print(f"--- 节点 {node_name} 完成 ---")
    if "messages" in update:
        for msg in update["messages"]:
            msg.pretty_print()

输出会清晰展示每一步：先是 agent 返回的 AIMessage（可能包含工具调用请求），然后是 tools 返回的 ToolMessage，接着又是 agent…… 直到最终回复。

5.2 Token 级流式输出：astream_events

如果你需要像 ChatGPT 那样逐 token 显示模型回复，就需要用到 astream_events（异步）。它可以捕获更细粒度的事件，包括 LLM 的流式生成 chunk。

import asyncio

async def stream_agent():
    config = {"configurable": {"thread_id": "demo-1"}}
    initial_state = {"messages": [HumanMessage(content="用搜索工具找到最新的奥斯卡最佳影片，并用计算器算一下今年是哪一年减去去年的年份")]}
    
    async for event in app.astream_events(initial_state, config, version="v1"):
        kind = event["event"]
        # 捕获 LLM 的 token 流
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                print(content, end="", flush=True)
        # 捕获工具调用开始和结束
        elif kind == "on_tool_start":
            print(f"\n🔧 正在调用工具: {event['name']}")
        elif kind == "on_tool_end":
            print(f"✅ 工具调用完成: {event['name']}")

# 运行异步函数
asyncio.run(stream_agent())

astream_events 的用法非常灵活：你可以监听 on_chat_model_stream 来逐字输出 LLM 的思考，监听 on_tool_start/on_tool_end 来显示工具执行状态，甚至监听自定义事件。结合 WebSocket 或 Server-Sent Events（SSE），你就能把 Agent 的整个“心路历程”实时推送到前端。

6. 完整可运行示例（汇总）

为了方便你测试，我把上述代码整合成一个完整的脚本。确保已设置环境变量 OPENAI_API_KEY 和 TAVILY_API_KEY。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# 1. 定义工具
search = TavilySearchResults(max_results=2)

@tool
def calculator(expression: str) -> str:
    """计算数学表达式，例如 '2+3'"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"错误: {e}"

tools = [search, calculator]

# 2. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 3. 初始化 LLM 并绑定工具
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 4. 路由逻辑
def should_continue(state: AgentState):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "end"

# 5. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
tool_node = ToolNode(tools)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
app = workflow.compile()

# 6. 测试调用
initial_state = {"messages": [HumanMessage(content="杭州今天的温度是多少摄氏度？并计算 345+678")]}
result = app.invoke(initial_state)
print(result["messages"][-1].content)

7. 部署与可观测性

有了一个能干活、能流式输出的 Agent，下一步就是把它部署为 API 服务，并监控它的每一步行为。

7.1 通过 LangServe 快速部署

LangServe 可以将你的 LangGraph 图一键转为带 /invoke、/stream 和 /stream_events 端点的 FastAPI 服务。

from fastapi import FastAPI
from langserve import add_routes

app_fastapi = FastAPI()
# app 是你上面编译好的 LangGraph 图
add_routes(app_fastapi, app, path="/agent")

# 运行服务器：uvicorn script:app_fastapi --reload

然后你就可以通过 HTTP POST 请求来与 Agent 交互：

curl -X POST http://localhost:8000/agent/stream \
  -H "Content-Type: application/json" \
  -d '{"input": {"messages": [{"type": "human", "content": "你好"}]}}'

/stream 端点返回 text/event-stream，非常适合 SSE 消费。

7.2 可观测性：LangSmith 追踪

在生产环境中，你需要清楚地知道 Agent 的每一步耗时、LLM 调用成本、工具调用成功率等。集成 LangSmith 只需设置环境变量：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=你的key
export LANGCHAIN_PROJECT=你的项目名

之后每一次 invoke 或 stream 都会自动在 LangSmith 平台上生成一条完整的执行轨迹，包括每个节点的输入输出、延迟、token 用量等。你可以像查看“分布式链路追踪”一样调试你的 Agent。

7.3 生产环境检查点

注意，前面的示例中我们没有传入检查点。在生产环境中，强烈建议为 app 传入持久化检查点（如 AsyncPostgresSaver），以便支持会话持久化和人工中断恢复。方法就是在 compile 时传入 checkpointer 参数。

from langgraph.checkpoint.postgres import AsyncPostgresSaver
# 需要先创建数据库连接池
checkpointer = AsyncPostgresSaver(conn_string)
app = workflow.compile(checkpointer=checkpointer)

此时所有执行都会持久化到 PostgreSQL，服务重启不影响进行中的会话。

8. 回顾与展望

经过三篇博客的旅程，我们从 LangGraph 的最基本概念走到了一个生产可用的工具调用 Agent。让我们回顾一下核心收获：

  • 状态 是图的记忆，add_messages 让对话历史管理极其简单。
  • 节点和边 构建了流程骨架，条件边赋予了 Agent 动态决策的能力。
  • 检查点 提供了持久化、暂停与恢复、时间旅行等高级特性。
  • 人机交互 让人类可以在关键时刻审批和修改，防止 Agent 越权。
  • 工具调用 让 Agent 与外部世界交互，ReAct 循环是“思考-行动-观察”的经典范式。
  • 流式输出 通过 stream 和 astream_events 实现了实时、透明的用户体验。
  • 部署与可观测性 借助 LangServe 和 LangSmith 可以快速将 Agent 推向生产并监控它。

LangGraph 的强大远不止于此。它的高级特性还包括：

  • 子图：将复杂流程拆分为可复用的子流程。
  • 多 Agent 协作：多个 Agent 节点互相通信，分工合作。
  • 动态图修改：在运行时动态增加或删除节点。
  • 持久化的复杂策略：如缓存、重试、并行执行等。
# 程序人生 # 人工智能
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

深入理解 AI Agent · MEMORY #02：记忆的工程机制

avatar AI Agent技术社区
cover

Windows部署OpenClaw：PowerShell安装、自定义模型与远程访问实测

avatar AI Agent技术社区
cover

DeepSeek V4 Pro正式版上线：AI Agent能力跃升背后的行业趋势

avatar AI Agent技术社区