1. 为什么说 LangGraph 是 Agent 的最终形态传统的大模型应用大多采用「单次调用」或「简单链式调用」的方式用户输入一句话模型直接返回结果。这种方式在处理简单问答时足够但面对需要多步推理、工具调用、状态记忆和条件分支的真实业务场景时就显得力不从心。Agent 的核心能力在于「自主决策」它需要根据当前状态决定下一步调用哪个工具、读取哪份数据、是否需要追问用户甚至要能自我纠错。这种能力本质上是一个有状态、可循环、可分支的图结构而不是一条直线。LangGraph 正是为此而生。它由 LangChain 团队推出把 Agent 的每一次决策建模为图Graph中的节点Node把节点之间的流转建模为边Edge并用一个全局的状态State对象在节点之间传递数据。这种设计让 Agent 具备了三个关键能力循环执行Agent 可以反复调用工具、观察结果、再次决策直到任务完成。条件分支根据当前状态动态决定下一步走向而不是固定顺序。持久化状态整个执行过程的状态可以被保存、恢复和回放便于调试和人工介入。因此LangGraph 被广泛认为是构建生产级 Agent 的「最终形态」——它把 Agent 从「提示词工程」提升到了「系统工程」的层面。2. LangGraph 核心概念在动手写代码之前先理解 LangGraph 的五个核心概念它们是后续所有实战的基础。2.1 State状态State 是贯穿整个图的数据结构相当于 Agent 的「工作记忆」。它可以是 TypedDict、Pydantic 模型或 dataclass。每个节点都可以读取 State 并返回对 State 的更新更新会合并回全局 State。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # add_messages 是 LangGraph 内置的 reducer用于把新消息追加到消息列表 messages: Annotated[list, add_messages] # 普通字段直接覆盖 current_step: str final_answer: str2.2 Node节点节点是图中的一个执行单元通常是一个 Python 函数。函数接收 State 作为参数返回一个字典作为对 State 的更新。def call_model(state: AgentState): 调用大模型的节点 messages state[messages] response llm.invoke(messages) return {messages: [response], current_step: model_called}2.3 Edge边边定义了节点之间的流转关系。LangGraph 支持两种边普通边固定流转和条件边根据状态动态选择下一个节点。from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(model, call_model) graph.add_node(tools, call_tools) 普通边model 执行完后固定进入 tools graph.add_edge(model, tools) 条件边根据状态决定是继续循环还是结束 graph.add_conditional_edges( tools, should_continue, # 判断函数 {continue: model, end: END} )2.4 编译与执行把节点和边组装成图之后调用compile()得到可执行对象然后像调用函数一样传入初始状态即可运行。app graph.compile() result app.invoke({messages: [{role: user, content: 帮我查一下今天的天气}]})2.5 检查点CheckpointLangGraph 内置了检查点机制可以在每个节点执行后保存状态快照。这让 Agent 具备断点续跑、人工介入和完整回放的能力是生产环境不可或缺的特性。3. 环境准备开始实战前先安装必要的依赖包。pip install langgraph langchain langchain-openai如果你需要使用 Tavily 搜索工具还需要安装并配置 API Keypip install tavily-python export TAVILY_API_KEYyour_tavily_api_key同时配置大模型 API Key这里以 OpenAI 为例export OPENAI_API_KEYyour_openai_api_key4. 实战一构建第一个 LangGraph Agent我们从最简单的「模型调用 工具调用 条件循环」开始完整实现一个能自主决定是否调用搜索工具的 Agent。4.1 定义状态from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]4.2 定义工具from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气情况 # 这里用模拟数据演示实际可接入天气 API return f{city} 今天晴气温 25 摄氏度适合出行。 tool def get_time() - str: 获取当前时间 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S)4.3 定义模型与节点from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode 初始化模型并绑定工具 llm ChatOpenAI(modelgpt-4o, temperature0) tools [get_weather, get_time] llm_with_tools llm.bind_tools(tools) def call_model(state: AgentState): 调用模型模型可能返回工具调用请求 response llm_with_tools.invoke(state[messages]) return {messages: [response]} ToolNode 是 LangGraph 预置的工具执行节点 tool_node ToolNode(tools)4.4 定义条件路由def should_continue(state: AgentState): 判断模型是否请求调用工具 last_message state[messages][-1] # 如果模型返回了 tool_calls则进入工具节点 if last_message.tool_calls: return tools # 否则结束 return end4.5 组装图并编译from langgraph.graph import StateGraph, START, END 1. 创建图 graph StateGraph(AgentState) 2. 添加节点 graph.add_node(model, call_model) graph.add_node(tools, tool_node) 3. 添加边 graph.add_edge(START, model) graph.add_conditional_edges( model, should_continue, {tools: tools, end: END} ) graph.add_edge(tools, model) # 工具执行完回到模型形成循环 4. 编译 app graph.compile()4.6 运行 Agentresult app.invoke({ messages: [{role: user, content: 北京今天天气怎么样顺便告诉我现在几点了。}] }) for msg in result[messages]: if msg.type ai: print(fAI: {msg.content}) elif msg.type tool: print(fTool[{msg.name}]: {msg.content})运行后你会看到 Agent 自动完成了「调用天气工具 → 调用时间工具 → 汇总回答」的完整流程这就是一个最基础的 ReAct Agent。5. 实战二带记忆的多轮对话 Agent真实业务中Agent 需要记住多轮对话的上下文。LangGraph 的检查点机制让这件事变得非常简单。5.1 使用内存检查点from langgraph.checkpoint.memory import MemorySaver 创建内存检查点 checkpointer MemorySaver() 编译时传入检查点 app graph.compile(checkpointercheckpointer) 第一次对话传入 thread_id 作为会话标识 config {configurable: {thread_id: session-001}} result1 app.invoke( {messages: [{role: user, content: 我叫小明记住我的名字。}]}, configconfig ) 第二次对话同一个 thread_idAgent 记得之前的对话 result2 app.invoke( {messages: [{role: user, content: 我叫什么名字}]}, configconfig ) print(result2[messages][-1].content) # 输出你叫小明。5.2 使用持久化检查点生产环境通常使用数据库作为检查点存储LangGraph 提供了多种后端支持。# 使用 SQLite 持久化需要安装 langgraph-checkpoint-sqlite from langgraph.checkpoint.sqlite import SqliteSaver 使用 Postgres 持久化需要安装 langgraph-checkpoint-postgres from langgraph.checkpoint.postgres import PostgresSaver 以 SQLite 为例 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-123}} result app.invoke( {messages: [{role: user, content: 帮我记住这个订单号A10086}]}, configconfig )有了持久化检查点Agent 的状态可以在服务重启后恢复这是生产级应用的关键能力。6. 实战三多 Agent 协作系统复杂任务往往需要多个专业 Agent 分工协作。LangGraph 支持在一个图中编排多个 Agent让它们像团队一样工作。6.1 定义三个专业 Agentfrom langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class TeamState(TypedDict): messages: Annotated[list, add_messages] task: str research_result: str code_result: str final_report: str 研究员 Agent def researcher_agent(state: TeamState): 负责收集资料和研究 task state[task] # 实际场景中这里会调用搜索工具或知识库 research f关于「{task}」的研究结果已收集到 5 篇相关论文和 3 个实践案例。 return {research_result: research, messages: [{role: assistant, content: research}]} 程序员 Agent def coder_agent(state: TeamState): 负责编写代码 task state[task] code f针对「{task}」生成的示例代码\nprint(Hello from {task}) return {code_result: code, messages: [{role: assistant, content: code}]} 主编 Agent def editor_agent(state: TeamState): 负责汇总成最终报告 report ( f最终报告\n f1. 研究部分{state[research_result]}\n f2. 代码部分{state[code_result]}\n f3. 结论该任务已完成建议下一步进行实际验证。 ) return {final_report: report, messages: [{role: assistant, content: report}]}6.2 组装多 Agent 图team_graph StateGraph(TeamState) 添加节点 team_graph.add_node(researcher, researcher_agent) team_graph.add_node(coder, coder_agent) team_graph.add_node(editor, editor_agent) 添加边研究员和程序员并行执行然后交给主编汇总 team_graph.add_edge(START, researcher) team_graph.add_edge(START, coder) team_graph.add_edge(researcher, editor) team_graph.add_edge(coder, editor) team_graph.add_edge(editor, END) team_app team_graph.compile() 运行多 Agent 系统 result team_app.invoke({ task: 用 Python 实现一个简单的推荐系统, messages: [] }) print(result[final_report])这个例子展示了 LangGraph 最强大的能力之一并行执行与汇聚。研究员和程序员可以同时工作最后统一交给主编汇总大幅提升复杂任务的执行效率。7. 实战四带人工审核的 Agent在金融、医疗等高风险场景中Agent 的关键操作需要人工确认。LangGraph 的interrupt机制可以暂停图执行等待人工介入。7.1 定义带中断的图from langgraph.types import interrupt, Command class ApprovalState(TypedDict): messages: Annotated[list, add_messages] action: str approved: bool def propose_action(state: ApprovalState): Agent 提出一个需要审批的操作 action 向用户账户转账 10000 元 # 触发中断等待人工审批 decision interrupt({action: action, reason: 大额转账需要人工确认}) return {action: action, approved: decision} def execute_action(state: ApprovalState): 执行审批通过的操作 if state[approved]: return {messages: [{role: assistant, content: f已执行操作{state[action]}}]} else: return {messages: [{role: assistant, content: 操作已被人工拒绝流程终止。}]} 组装图 approval_graph StateGraph(ApprovalState) approval_graph.add_node(propose, propose_action) approval_graph.add_node(execute, execute_action) approval_graph.add_edge(START, propose) approval_graph.add_edge(propose, execute) approval_graph.add_edge(execute, END) approval_app approval_graph.compile()7.2 运行并处理人工审批config {configurable: {thread_id: approval-001}} 第一次调用Agent 提出操作并中断 result approval_app.invoke( {messages: [{role: user, content: 请帮我转账 10000 元}], action: , approved: False}, configconfig ) 此时图已暂停我们可以获取中断信息 实际应用中这里会把审批请求推送给人工审核员 人工审核通过后继续执行 result approval_app.invoke( Command(resumeTrue), # 传入 True 表示审批通过 configconfig ) print(result[messages][-1].content) # 输出已执行操作向用户账户转账 10000 元这个机制让 Agent 在关键节点「停下来等人」既保留了自动化效率又满足了合规要求。8. 实战五流式输出与实时监控生产环境中用户希望看到 Agent 的实时思考过程而不是等待最终结果。LangGraph 提供了完善的流式输出支持。8.1 流式输出节点结果# 使用 stream 方法逐节点输出 for event in app.stream( {messages: [{role: user, content: 北京天气怎么样}]}, config{configurable: {thread_id: stream-001}} ): for key, value in event.items(): print(f节点 [{key}] 执行完成) if messages in value: last_msg value[messages][-1] if hasattr(last_msg, content) and last_msg.content: print(f 输出: {last_msg.content})8.2 流式输出 Token# 使用 stream_modemessages 逐 token 输出 for msg_chunk in app.stream( {messages: [{role: user, content: 用一句话介绍 LangGraph}]}, config{configurable: {thread_id: token-001}}, stream_modemessages ): chunk msg_chunk[0] if chunk.content: print(chunk.content, end, flushTrue)流式输出让 Agent 的「思考过程」变得透明用户体验大幅提升也方便前端做打字机效果。9. 生产化部署建议把 LangGraph Agent 部署到生产环境需要关注以下几个关键点。9.1 使用 LangGraph PlatformLangChain 官方提供了 LangGraph Platform支持一键部署、自动伸缩、监控告警和版本管理。如果你的团队预算充足这是最省心的方案。9.2 自建服务如果选择自建推荐使用 FastAPI 封装 Agent 服务并配合 Redis 或 Postgres 做状态持久化。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str app.post(/chat) async def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result app.invoke( {messages: [{role: user, content: req.message}]}, configconfig ) return {reply: result[messages][-1].content}9.3 监控与可观测性生产环境必须对 Agent 的执行过程进行监控。LangGraph 的检查点机制天然支持执行回放配合 LangSmith 可以记录每一次运行的完整轨迹包括每个节点的输入输出、耗时和 Token 消耗方便定位问题和优化成本。10. 总结LangGraph 之所以被称为 Agent 的最终形态是因为它把 Agent 从「提示词工程」提升到了「系统工程」的层面。通过本文的五个实战案例你已经掌握了基础 ReAct Agent模型与工具的循环协作。多轮记忆通过检查点机制实现会话级记忆。多 Agent 协作并行执行与结果汇聚。人工审核通过中断机制实现人机协同。流式输出实时展示 Agent 的思考过程。下一步建议你从自己的业务场景出发先画一张 Agent 的流程图明确节点、边和状态再用 LangGraph 把它变成可运行的代码。当你真正把 Agent 的「决策权」交给图结构时你会发现它带来的不仅是代码上的优雅更是产品能力上的质变。
