一、先说结论为什么需要 LangGraph用大模型做 Agent很多人第一步都是手写while循环whileTrue:responsemodel.invoke(messages)ifnotresponse.tool_calls:breakforcallinresponse.tool_calls:messages.append(run_tool(call))messages.append(response)跑起来没问题但只要需求稍微复杂一点就崩了想加个「先检索、再回答、不通过就重试」的分支 → 循环里开始堆if/else想在写数据库前让人工确认一下 → 得自己实现「暂停 / 恢复」想把中间状态存下来重启后接着跑 → 得自己序列化整个messages想让多个子任务并行再汇总 → 手写调度器越写越像在造轮子。LangGraph 就是把这个「手写循环」抽象成一张有向图Graph节点是干活的函数边是流转规则状态是一份所有节点共享的「白板」。官方在 v1 里把图 API 定为稳定接口核心就三件套State状态、Node节点、Edge边学一次能用来搭绝大多数 Agent 工作流。本文基于LangGraph v1当前稳定版Python ≥ 3.10。如果你在旧教程里看到create_react_agent它在 v1 已被标记为弃用官方推荐用create_agent文中两种写法都会给到。二、环境准备# 一次性装好pipinstall-Ulanggraph langchain langchain-openai# 可选本地持久化用 SQLite第二篇会用到pipinstall-Ulanggraph-checkpoint-sqlite设置一下 Key这里以 OpenAI 为例换成任意兼容模型都行# Linux / macOSexportOPENAI_API_KEYsk-xxx# Windows PowerShell$env:OPENAI_API_KEYsk-xxx三、三个核心概念一分钟讲透概念是什么一句话记忆State一张所有节点共享的数据白板通常用TypedDict定义大家读写同一份数据Node一个普通 Python 函数入参是 State返回「要更新的字段」收到状态返回补丁Edge节点之间的连线分普通边和条件边决定下一步去哪有个细节新手最容易踩节点函数不是「返回新状态」而是「返回一个更新补丁」。LangGraph 会把这个补丁和旧状态按规则合并merge并规则由Annotated上挂的reducer决定。默认是「覆盖」你可以指定成「追加」等。┌─────────────┐ │ State │ ← 所有节点共享的白板 └──────┬──────┘ add_edge │ add_conditional_edges ┌───────────┼───────────┐ ┌──▼──┐ ┌──▼──┐ ┌──▼──┐ │Node │────▶│Node │────▶│Node │ └─────┘ └─────┘ └─────┘四、第一个例子手写一个最小状态图先不请出大模型用最简单的方式把「状态 reducer 边」跑通。importoperatorfromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,START,ENDclassCounterState(TypedDict):# 没有 Annotated默认「覆盖」后写的值直接盖掉旧值total:int# 带 Annotated[list, operator.add]reducer 是「追加」每次返回会拼接到旧列表后面logs:Annotated[list[str],operator.add]defadd_two(state:CounterState)-dict:return{total:state[total]2,logs:[add_two: 2]}deftimes_three(state:CounterState)-dict:return{total:state[total]*3,logs:[times_three: *3]}# 1. 用状态类型创建图builderStateGraph(CounterState)# 2. 添加节点builder.add_node(add_two,add_two)builder.add_node(times_three,times_three)# 3. 连边START - add_two - times_three - ENDbuilder.add_edge(START,add_two)builder.add_edge(add_two,times_three)builder.add_edge(times_three,END)# 4. 编译成可执行对象graphbuilder.compile()resultgraph.invoke({total:0,logs:[]})print(result)# {total: 6, logs: [add_two: 2, times_three: *3]}执行轨迹total从 0 → 2 → 6logs因为是追加型 reducer两条日志都被保留。划重点logs如果没有Annotated[..., operator.add]第二次返回会把第一次的日志整个覆盖掉最后只剩一条。这是 LangGraph 里排名第一的高频坑记住一句话要「累积」的字段必须挂 reducer。五、进阶搭一个能调用工具的 ReAct AgentReAct 的本质就是一个带条件边的循环模型决定要不要调工具 → 要调就去tools节点 → 调完回到模型 → 直到模型不再要求调工具。用图来表达简直严丝合缝。fromtypingimportAnnotated,Literal,TypedDictfromlangchain_core.messagesimportBaseMessage,HumanMessagefromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.prebuiltimportToolNode# 1) 状态一个 messages 列表用官方 add_messages 做 reducer它会智能地追加/合并消息classAgentState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# 2) 定义工具tooldefget_weather(city:str)-str:查询指定城市的天气情况。fake_db{北京:晴26℃,上海:多云24℃,深圳:小雨28℃}returnfake_db.get(city,f暂时查不到{city}的天气)tools[get_weather]# 3) 把工具绑到模型上modelChatOpenAI(modelgpt-4o-mini,temperature0).bind_tools(tools)# 4) 模型节点接收全部消息产出新消息defagent_node(state:AgentState)-dict:return{messages:[model.invoke(state[messages])]}# 5) 路由函数模型还想调工具就回 tools否则结束defshould_continue(state:AgentState)-Literal[tools,__end__]:laststate[messages][-1]ifgetattr(last,tool_calls,None):returntoolsreturnEND# 6) 装配图builderStateGraph(AgentState)builder.add_node(agent,agent_node)builder.add_node(tools,ToolNode(tools))# 官方的工具执行节点省去手写builder.add_edge(START,agent)builder.add_conditional_edges(agent,should_continue,{tools:tools,END:END},# 路由返回值 - 目标节点)builder.add_edge(tools,agent)# 关键调完工具回到模型形成循环graphbuilder.compile()# 7) 跑一下resultgraph.invoke({messages:[HumanMessage(content北京今天天气怎么样)]})print(result[messages][-1].content)# 北京今天天气是晴天气温约 26℃。这张图长这样┌──────────────────────┐ │ │ (还有 tool_calls) ┌────▼────┐ 条件边 ┌────┴────┐ │ agent ├────────────▶│ tools │ └────┬────┘ └─────────┘ │ (没有 tool_calls) ▼ END三条add_edge/add_conditional_edges就表达出了「模型 ↔ 工具」的循环比手写while清晰太多。六、更省事的写法官方封装的create_agent上面那张图其实就是业界标准的 ReAct 结构官方直接帮你装好了。v1 里推荐写法是langchain.agents.create_agentfromlangchain.agentsimportcreate_agentfromlangchain_core.messagesimportHumanMessage agentcreate_agent(modelopenai:gpt-4o-mini,# 也支持 anthropic:claude-... 等tools[get_weather],system_prompt你是一个乐于助人的中文助手。,)resultagent.invoke({messages:[HumanMessage(content上海和深圳哪里天气好)]})print(result[messages][-1].content)什么时候用手写图什么时候用create_agent只想快速跑通一个标准 Agent → 用create_agent需要自定义分支、并行、人工审批、多智能体协作 → 回到StateGraph手写图自由度最高。⚠️ 版本提醒老教程里的from langgraph.prebuilt import create_react_agent在 v1 已弃用函数名换成了create_agentprompt参数改叫system_prompt。如果你照着旧文报 ImportError / 弃用警告就是这个原因。七、调试神器流式输出看每一步搭好图一定要会看执行过程stream是最好的调试工具。forchunkingraph.stream({messages:[HumanMessage(content北京和上海天气对比一下)]},stream_modeupdates,# 每个节点执行完就吐一块):print(chunk)会依次打印{agent: {messages: [AIMessage(..., tool_calls[...])]}} {tools: {messages: [ToolMessage(晴26℃), ToolMessage(多云24℃)]}} {agent: {messages: [AIMessage(北京晴 26℃上海多云 24℃……)]}}把stream_mode换成values则会输出每一步的完整状态快照两种模式按需切换。八、小结与预告这一篇我们把 LangGraph 的地基打好了图三件套State 白板、Node 打补丁、Edge 定流向reducer 是灵魂要累积的字段必须挂Annotated[..., add]ReAct 带条件边的循环三条边就能表达也可以用create_agent一键生成stream是调试入口能看清图里每一步。但仔细想想上面这个 Agent 有个致命缺陷只要程序重启、或者换个会话它就把之前聊过的事全忘了——因为它没有任何「记忆」。会话状态到底存在哪怎么做到断点续跑又怎么在关键操作前插入「人工确认」下一篇《LangGraph 实战二持久化、检查点与人机协同》我们就来解决这三个问题Checkpointer、thread_id、以及interruptCommand这一对 HITL 神器。代码可以直接接着本篇跑。如果这篇对你有帮助欢迎点赞收藏第二篇已同步发出搜索「LangGraph 实战二」即可找到。