LangGraph[3] ---- 实战 LangGraph Agent:从工具调用到流式交付

前言

通过前面两篇博客，我们已经掌握了 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. 准备工作：环境与工具定义

首先安装必要的包：

pipinstalllanggraph langchain langchain-openai langchain-community tavily-python

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

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

fromlangchain_community.tools.tavily_searchimportTavilySearchResultsfromlangchain_core.toolsimporttoolimportmath# 搜索工具search=TavilySearchResults(max_results=2)# 计算器工具@tooldefcalculator(expression:str)->str:"""计算一个数学表达式，例如 '2+3'，返回计算结果。"""try:# 安全地计算数学表达式（仅允许数字和运算符）allowed=set("0123456789+-*/(). ")ifnotall(cinallowedforcinexpression):return"表达式包含非法字符"returnstr(eval(expression))exceptExceptionase:returnf"计算出错:{str(e)}"# 工具列表，稍后绑定到 LLMtools=[search,calculator]

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

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

fromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassAgentState(TypedDict):messages:Annotated[list,add_messages]

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

fromlangchain_openaiimportChatOpenAIfromlanggraph.prebuiltimportToolNodeimportos# 初始化 LLM，确保它可以请求调用工具llm=ChatOpenAI(model="gpt-4o",temperature=0)llm_with_tools=llm.bind_tools(tools)# agent 节点函数defagent_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 的最后一条消息是否包含工具调用。

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

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

tool_node=ToolNode(tools)

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

fromlanggraph.graphimportStateGraph 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 提示。

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

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

5.2 Token 级流式输出：astream_events

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

importasyncioasyncdefstream_agent():config={"configurable":{"thread_id":"demo-1"}}initial_state={"messages":[HumanMessage(content="用搜索工具找到最新的奥斯卡最佳影片，并用计算器算一下今年是哪一年减去去年的年份")]}asyncforeventinapp.astream_events(initial_state,config,version="v1"):kind=event["event"]# 捕获 LLM 的 token 流ifkind=="on_chat_model_stream":content=event["data"]["chunk"].contentifcontent:print(content,end="",flush=True)# 捕获工具调用开始和结束elifkind=="on_tool_start":print(f"\n🔧 正在调用工具:{event['name']}")elifkind=="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。

fromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesfromlanggraph.graphimportStateGraph,ENDfromlanggraph.prebuiltimportToolNodefromlangchain_openaiimportChatOpenAIfromlangchain_community.tools.tavily_searchimportTavilySearchResultsfromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportHumanMessage# 1. 定义工具search=TavilySearchResults(max_results=2)@tooldefcalculator(expression:str)->str:"""计算数学表达式，例如 '2+3'"""try:returnstr(eval(expression))exceptExceptionase:returnf"错误:{e}"tools=[search,calculator]# 2. 定义状态classAgentState(TypedDict):messages:Annotated[list,add_messages]# 3. 初始化 LLM 并绑定工具llm=ChatOpenAI(model="gpt-4o",temperature=0)llm_with_tools=llm.bind_tools(tools)defagent_node(state:AgentState)->dict:response=llm_with_tools.invoke(state["messages"])return{"messages":[response]}# 4. 路由逻辑defshould_continue(state:AgentState):last_msg=state["messages"][-1]ifhasattr(last_msg,"tool_calls")andlast_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 服务。

fromfastapiimportFastAPIfromlangserveimportadd_routes app_fastapi=FastAPI()# app 是你上面编译好的 LangGraph 图add_routes(app_fastapi,app,path="/agent")# 运行服务器：uvicorn script:app_fastapi --reload

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

curl-XPOST 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 只需设置环境变量：

exportLANGCHAIN_TRACING_V2=trueexportLANGCHAIN_API_KEY=你的keyexportLANGCHAIN_PROJECT=你的项目名

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

7.3 生产环境检查点

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

fromlanggraph.checkpoint.postgresimportAsyncPostgresSaver# 需要先创建数据库连接池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 节点互相通信，分工合作。
• 动态图修改：在运行时动态增加或删除节点。
• 持久化的复杂策略：如缓存、重试、并行执行等。