文章目录前言1. Agent 架构总览2. 准备工作环境与工具定义3. 定义 Agent 的状态和 LLM 节点4. 构建工具调用循环5. 流式输出让用户看见 Agent 的“思考过程”5.1 按节点级别流式输出stream5.2 Token 级流式输出astream_events6. 完整可运行示例汇总7. 部署与可观测性7.1 通过 LangServe 快速部署7.2 可观测性LangSmith 追踪7.3 生产环境检查点8. 回顾与展望前言通过前面两篇博客我们已经掌握了 LangGraph 的核心机制状态、节点、边、条件路由、检查点持久化以及人机交互。这些能力共同构成了一个“活的”流程引擎。但真正让智能体能干实事的是它调用外部工具的能力——搜索最新信息、执行代码、操作数据库、发送请求……以及将这些执行过程实时、流式地展现给用户。今天这篇我们就动手构建一个生产级的 LangGraph Agent。它会使用真正的 LLM如 GPT-4进行决策调用搜索和计算两个自定义工具形成“思考 → 行动 → 观察”的 ReAct 循环以 token 级别的流式输出向前端实时推送最后我还会给出将其部署为 API 并添加可观测性的完整思路1. Agent 架构总览我们要实现的 Agent 核心循环如下用户输入 → agentLLM 决策 │ ├─ 需要工具─→ 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# 搜索工具searchTavilySearchResults(max_results2)# 计算器工具tooldefcalculator(expression:str)-str:计算一个数学表达式例如 23返回计算结果。try:# 安全地计算数学表达式仅允许数字和运算符allowedset(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确保它可以请求调用工具llmChatOpenAI(modelgpt-4o,temperature0)llm_with_toolsllm.bind_tools(tools)# agent 节点函数defagent_node(state:AgentState)-dict:# 调用 LLM传入当前消息历史responsellm_with_tools.invoke(state[messages])# 返回 AI 消息可能包含 tool_callsreturn{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_messagestate[messages][-1]# 如果有 tool_calls 属性且非空说明需要调用工具ifhasattr(last_message,tool_calls)andlast_message.tool_calls:returntoolsreturnend然后我们需要一个能执行工具的节点。LangGraph 提供了一个极其方便的预置节点ToolNode它接收包含工具调用的 AIMessage自动执行相应的工具并将结果包装成ToolMessage追加回消息列表。tool_nodeToolNode(tools)现在把节点和边拼装成图fromlanggraph.graphimportStateGraph workflowStateGraph(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)# 编译appworkflow.compile()至此一个标准的工具调用 Agent 图就构建完成了。你可以像之前一样用invoke运行它initial_state{messages:[HumanMessage(content今天北京天气怎么样然后帮我算一下 123 * 456 等于多少)]}resultapp.invoke(initial_state)print(result[messages][-1].content)这个图会经历agentLLM 想查天气→tools执行搜索→agent拿到天气又想算乘法→tools执行计算器→agent综合结果生成最终答案。一切都在状态messages中自动记录。5. 流式输出让用户看见 Agent 的“思考过程”上面的invoke是阻塞式一次性返回最终结果用户无法感知中间步骤。在生产级应用中我们通常希望逐步推送执行进度模型正在生成哪个 token、正在调用哪个工具、工具执行结果是什么……这能极大提升用户体验也便于调试。LangGraph 提供了多种流式方法最常用的是stream和astream_events。5.1 按节点级别流式输出streamstream会在每个节点执行完毕后输出该节点产生的状态更新。你可以用它的输出来构造“当前正在执行哪个步骤”的 UI 提示。forchunkinapp.stream(initial_state,stream_modeupdates):# chunk 是一个字典键为节点名值为该节点返回的状态更新node_namelist(chunk.keys())[0]updatechunk[node_name]print(f--- 节点{node_name}完成 ---)ifmessagesinupdate: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,versionv1):kindevent[event]# 捕获 LLM 的 token 流ifkindon_chat_model_stream:contentevent[data][chunk].contentifcontent:print(content,end,flushTrue)# 捕获工具调用开始和结束elifkindon_tool_start:print(f\n 正在调用工具:{event[name]})elifkindon_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 EventsSSE你就能把 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. 定义工具searchTavilySearchResults(max_results2)tooldefcalculator(expression:str)-str:计算数学表达式例如 23try:returnstr(eval(expression))exceptExceptionase:returnf错误:{e}tools[search,calculator]# 2. 定义状态classAgentState(TypedDict):messages:Annotated[list,add_messages]# 3. 初始化 LLM 并绑定工具llmChatOpenAI(modelgpt-4o,temperature0)llm_with_toolsllm.bind_tools(tools)defagent_node(state:AgentState)-dict:responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 4. 路由逻辑defshould_continue(state:AgentState):last_msgstate[messages][-1]ifhasattr(last_msg,tool_calls)andlast_msg.tool_calls:returntoolsreturnend# 5. 构建图workflowStateGraph(AgentState)workflow.add_node(agent,agent_node)tool_nodeToolNode(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)appworkflow.compile()# 6. 测试调用initial_state{messages:[HumanMessage(content杭州今天的温度是多少摄氏度并计算 345678)]}resultapp.invoke(initial_state)print(result[messages][-1].content)7. 部署与可观测性有了一个能干活、能流式输出的 Agent下一步就是把它部署为 API 服务并监控它的每一步行为。7.1 通过 LangServe 快速部署LangServe 可以将你的 LangGraph 图一键转为带/invoke、/stream和/stream_events端点的 FastAPI 服务。fromfastapiimportFastAPIfromlangserveimportadd_routes app_fastapiFastAPI()# app 是你上面编译好的 LangGraph 图add_routes(app_fastapi,app,path/agent)# 运行服务器uvicorn script:app_fastapi --reload然后你就可以通过 HTTP POST 请求来与 Agent 交互curl-XPOST http://localhost:8000/agent/stream\-HContent-Type: application/json\-d{input: {messages: [{type: human, content: 你好}]}}/stream端点返回text/event-stream非常适合 SSE 消费。7.2 可观测性LangSmith 追踪在生产环境中你需要清楚地知道 Agent 的每一步耗时、LLM 调用成本、工具调用成功率等。集成 LangSmith 只需设置环境变量exportLANGCHAIN_TRACING_V2trueexportLANGCHAIN_API_KEY你的keyexportLANGCHAIN_PROJECT你的项目名之后每一次invoke或stream都会自动在 LangSmith 平台上生成一条完整的执行轨迹包括每个节点的输入输出、延迟、token 用量等。你可以像查看“分布式链路追踪”一样调试你的 Agent。7.3 生产环境检查点注意前面的示例中我们没有传入检查点。在生产环境中强烈建议为app传入持久化检查点如AsyncPostgresSaver以便支持会话持久化和人工中断恢复。方法就是在compile时传入checkpointer参数。fromlanggraph.checkpoint.postgresimportAsyncPostgresSaver# 需要先创建数据库连接池checkpointerAsyncPostgresSaver(conn_string)appworkflow.compile(checkpointercheckpointer)此时所有执行都会持久化到 PostgreSQL服务重启不影响进行中的会话。8. 回顾与展望经过三篇博客的旅程我们从 LangGraph 的最基本概念走到了一个生产可用的工具调用 Agent。让我们回顾一下核心收获状态是图的记忆add_messages让对话历史管理极其简单。节点和边构建了流程骨架条件边赋予了 Agent 动态决策的能力。检查点提供了持久化、暂停与恢复、时间旅行等高级特性。人机交互让人类可以在关键时刻审批和修改防止 Agent 越权。工具调用让 Agent 与外部世界交互ReAct 循环是“思考-行动-观察”的经典范式。流式输出通过stream和astream_events实现了实时、透明的用户体验。部署与可观测性借助 LangServe 和 LangSmith 可以快速将 Agent 推向生产并监控它。LangGraph 的强大远不止于此。它的高级特性还包括子图将复杂流程拆分为可复用的子流程。多 Agent 协作多个 Agent 节点互相通信分工合作。动态图修改在运行时动态增加或删除节点。持久化的复杂策略如缓存、重试、并行执行等。