LangGraph入门：构建有状态的AI Agent工作流

LangGraph 入门用状态图构建 Agent手写 ReAct 循环容易写出 bug。LangGraph 用「状态图」的方式定义 Agent把每一步定义为一个节点跳转逻辑定义为边——清晰、可测试、可扩展。一、为什么需要 LangGraph手写 Agent 循环的痛点# 手写版容易出 bug难扩展whilestepmax_steps:responsellm.chat(messages)ifFinal Answerinresponse:breakeliftool_call:resultexecute(tool_call)messages.append(result)elifneed_retry:# 嗯...重试逻辑写哪# 越写越乱LangGraph 换了个思路Agent 就是一个状态图State Graph。每个「做什么」是一个节点每个「下一步去哪」是一条边。二、核心概念┌──────────────────────────────────────┐ │ LangGraph 四要素 │ ├──────────────────────────────────────┤ │ State状态 Agent 当前记住的所有信息 │ │ Node节点 做什么调用 LLM、执行工具│ │ Edge边 做完之后去哪 │ │ Graph图 节点 边的集合 │ └──────────────────────────────────────┘三、第一个 LangGraph Agent# pip install langgraph langchain-openaifromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAI# ── 1. 定义状态 ──classAgentState(TypedDict):messages:list[dict]# 对话历史next_step:str# 下一步去哪# ── 2. 初始化 LLM ──llmChatOpenAI(modeldeepseek-chat,api_keyyour-key,base_urlhttps://api.deepseek.com/v1)# ── 3. 定义节点函数 ──defchatbot(state:AgentState)-AgentState:LLM 节点调用大模型responsellm.invoke(state[messages])state[messages].append(response)state[next_step]check_tools# 下一步去检查是否需要工具returnstatedefcheck_tools(state:AgentState)-AgentState:工具路由判断是否需要调工具last_msgstate[messages][-1]ifhasattr(last_msg,tool_calls)andlast_msg.tool_calls:state[next_step]execute_toolselse:state[next_step]endreturnstatedefexecute_tools(state:AgentState)-AgentState:工具执行节点# 简化的工具执行逻辑last_msgstate[messages][-1]fortool_callinlast_msg.tool_calls:resultf工具{tool_call[name]}执行结果state[messages].append({role:tool,content:result})state[next_step]chatbot# 回到 LLM 继续returnstate# ── 4. 构建图 ──defbuild_graph():graphStateGraph(AgentState)# 添加节点graph.add_node(chatbot,chatbot)graph.add_node(check_tools,check_tools)graph.add_node(execute_tools,execute_tools)# 添加边graph.add_edge(chatbot,check_tools)# 条件边根据 next_step 决定去向graph.add_conditional_edges(check_tools,lambdastate:state[next_step],{execute_tools:execute_tools,end:END})graph.add_edge(execute_tools,chatbot)# 返回 LLMgraph.set_entry_point(chatbot)# 入口returngraph.compile()# ── 5. 运行 ──agentbuild_graph()resultagent.invoke({messages:[{role:user,content:你好}],next_step:chatbot})print(result[messages][-1].content)执行流程可视化┌──────────┐ │ chatbot │ ← 入口调用 LLM └────┬─────┘ │ ┌────▼─────┐ │check_tools│ ← 判断需要工具吗 └────┬─────┘ ┌───┴───┐ ▼ ▼ execute END _tools │ │ │ └───→───┘ 回到 chatbot四、完整实战代码审查 Agentfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverclassCodeReviewState(TypedDict):code:strreview_result:strfixed_code:strtest_result:strstep:strdefreview_code(state:CodeReviewState)-CodeReviewState:节点 1审查代码promptf审查以下代码指出问题和改进建议\n\n{state[code]}\nresponsellm.invoke([{role:user,content:prompt}])state[review_result]response.content state[step]fixif问题inresponse.contentelsepassreturnstatedeffix_code(state:CodeReviewState)-CodeReviewState:节点 2修复问题promptf 原始代码 {state[code]} 审查意见{state[review_result]}请修复所有问题只输出修复后的代码。 responsellm.invoke([{role:user,content:prompt}])state[fixed_code]response.content state[step]testreturnstatedefrun_tests(state:CodeReviewState)-CodeReviewState:节点 3运行测试验证# 实际项目中会真的执行测试promptf检查以下代码是否有明显的语法或逻辑错误\n\n{state[fixed_code]}\nresponsellm.invoke([{role:user,content:prompt}])state[test_result]response.content state[step]donereturnstate# 构建审查工作流defbuild_review_graph():graphStateGraph(CodeReviewState)graph.add_node(review,review_code)graph.add_node(fix,fix_code)graph.add_node(test,run_tests)graph.set_entry_point(review)# 条件路由graph.add_conditional_edges(review,lambdas:s[step],{fix:fix,pass:END})graph.add_edge(fix,test)graph.add_edge(test,END)returngraph.compile()# 运行review_agentbuild_review_graph()resultreview_agent.invoke({code:def add(a,b):\n return ab\n\nprint(add(1,2)),review_result:,fixed_code:,test_result:,step:review})print(f审查结果{result[review_result]})print(f修复代码\n{result[fixed_code]})五、LangGraph vs 手写循环维度手写 while 循环LangGraph逻辑清晰度❌ 嵌套深了容易乱✅ 状态图一目了然可测试性❌ 整个循环是一个函数✅ 每个节点独立测试可扩展性❌ 加新逻辑要改主循环✅ 加新节点 新边调试❌ print 大法✅ 每个节点状态可见持久化❌ 自己实现✅ MemorySaver 开箱即用学习成本✅ 不需要学框架❌ 需要学新概念六、总结LangGraph 是构建 Agent 的工业级方案——告别手写循环核心是「状态图」——节点 做什么边 去哪条件边实现路由——根据不同状态走不同分支每个节点独立可测试——符合软件工程最佳实践七、生产实战LangGraph 跑到生产才知道的事7.1 持久化和断点续跑Agent 跑了 10 步后崩溃了从头再跑一遍生产环境不允许fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.checkpoint.sqliteimportSqliteSaver# 开发环境内存存储memoryMemorySaver()# 生产环境SQLite 持久化重启不丢checkpointerSqliteSaver.from_conn_string(checkpoints.db)graphgraph.compile(checkpointercheckpointer)# 运行时可指定 thread_id 来区分会话config{configurable:{thread_id:user-session-123}}resultgraph.invoke(initial_state,config)# 如果崩溃了同一个 thread_id 可以从上次的 checkpoint 继续resultgraph.invoke(None,config)# None 从上次 checkpoint 恢复7.2 流式输出用户不想等Agent 执行期间用户盯着空白界面体验很差# 每个节点完成后推送状态更新asyncforeventingraph.astream(initial_state):node_namelist(event.keys())[0]yield{type:node_update,node:node_name,status:completed}# 如果是 chatbot 节点流式输出 Tokenifnode_namechatbot:forchunkinevent[node_name][messages][-1].content:yield{type:chunk,text:chunk}7.3 节点级错误处理fromlanggraph.graphimportStateGraphdefsafe_review(state:State)-State:try:returnreview_code(state)exceptExceptionase:state[error]str(e)state[step]error_handlerreturnstate# 注册错误处理节点graph.add_node(review,safe_review)graph.add_node(error_handler,handle_error)graph.add_conditional_edges(review,lambdas:s[step],{fix:fix,pass:END,error_handler:error_handler})7.4 并行节点速度翻倍LangGraph 的 Send API 支持并行fromlanggraph.typesimportSend# 继续条件为每个 sub-task 创建并行执行asyncdefcontinue_to_workers(state):return[Send(worker,{task:t})fortinstate[tasks]]# 3 个 worker 同时执行总时间 最慢的那个下一篇《MCP 协议实战给 AI 接上外部世界》——写一个 MCP Server让 Claude Desktop、Cursor 都能调用你的工具。系列文章00-总纲 → ①-LLM 原理 → ②-Prompt 工程 → ③-Function Calling → ④-RAG → ⑤-Agent 模式 → ⑥-LangGraph → ⑦-MCP → ⑧-Multi-Agent