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

LangGraph 入门：用状态图构建 Agent

手写 ReAct 循环容易写出 bug。LangGraph 用「状态图」的方式定义 Agent，把每一步定义为一个节点，跳转逻辑定义为边——清晰、可测试、可扩展。

一、为什么需要 LangGraph

手写 Agent 循环的痛点：

# 手写版：容易出 bug，难扩展whilestep<max_steps:response=llm.chat(messages)if"Final Answer"inresponse:breakeliftool_call:result=execute(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 ──llm=ChatOpenAI(model="deepseek-chat",api_key="your-key",base_url="https://api.deepseek.com/v1")# ── 3. 定义节点函数 ──defchatbot(state:AgentState)->AgentState:"""LLM 节点：调用大模型"""response=llm.invoke(state["messages"])state["messages"].append(response)state["next_step"]="check_tools"# 下一步去检查是否需要工具returnstatedefcheck_tools(state:AgentState)->AgentState:"""工具路由：判断是否需要调工具"""last_msg=state["messages"][-1]ifhasattr(last_msg,"tool_calls")andlast_msg.tool_calls:state["next_step"]="execute_tools"else:state["next_step"]="end"returnstatedefexecute_tools(state:AgentState)->AgentState:"""工具执行节点"""# 简化的工具执行逻辑last_msg=state["messages"][-1]fortool_callinlast_msg.tool_calls:result=f"工具{tool_call['name']}执行结果"state["messages"].append({"role":"tool","content":result})state["next_step"]="chatbot"# 回到 LLM 继续returnstate# ── 4. 构建图 ──defbuild_graph():graph=StateGraph(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. 运行 ──agent=build_graph()result=agent.invoke({"messages":[{"role":"user","content":"你好"}],"next_step":"chatbot"})print(result["messages"][-1].content)

执行流程可视化：

┌──────────┐ │ chatbot │ ← 入口：调用 LLM └────┬─────┘ │ ┌────▼─────┐ │check_tools│ ← 判断：需要工具吗？ └────┬─────┘ ┌───┴───┐ ▼ ▼ execute END _tools │ │ │ └───→───┘ （回到 chatbot）

四、完整实战：代码审查 Agent

fromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverclassCodeReviewState(TypedDict):code:strreview_result:strfixed_code:strtest_result:strstep:strdefreview_code(state:CodeReviewState)->CodeReviewState:"""节点 1：审查代码"""prompt=f"审查以下代码，指出问题和改进建议：\n```\n{state['code']}\n```"response=llm.invoke([{"role":"user","content":prompt}])state["review_result"]=response.content state["step"]="fix"if"问题"inresponse.contentelse"pass"returnstatedeffix_code(state:CodeReviewState)->CodeReviewState:"""节点 2：修复问题"""prompt=f""" 原始代码： ```{state['code']}``` 审查意见：{state['review_result']}请修复所有问题，只输出修复后的代码。 """response=llm.invoke([{"role":"user","content":prompt}])state["fixed_code"]=response.content state["step"]="test"returnstatedefrun_tests(state:CodeReviewState)->CodeReviewState:"""节点 3：运行测试验证"""# 实际项目中会真的执行测试prompt=f"检查以下代码是否有明显的语法或逻辑错误：\n```\n{state['fixed_code']}\n```"response=llm.invoke([{"role":"user","content":prompt}])state["test_result"]=response.content state["step"]="done"returnstate# 构建审查工作流defbuild_review_graph():graph=StateGraph(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_agent=build_review_graph()result=review_agent.invoke({"code":"def add(a,b):\n return a+b\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 手写循环

六、总结

1. LangGraph 是构建 Agent 的工业级方案——告别手写循环
2. 核心是「状态图」——节点 = 做什么，边 = 去哪
3. 条件边实现路由——根据不同状态走不同分支
4. 每个节点独立可测试——符合软件工程最佳实践

七、生产实战：LangGraph 跑到生产才知道的事

7.1 持久化和断点续跑

Agent 跑了 10 步后崩溃了，从头再跑一遍？生产环境不允许：

fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.checkpoint.sqliteimportSqliteSaver# 开发环境：内存存储memory=MemorySaver()# 生产环境：SQLite 持久化（重启不丢）checkpointer=SqliteSaver.from_conn_string("checkpoints.db")graph=graph.compile(checkpointer=checkpointer)# 运行时可指定 thread_id 来区分会话config={"configurable":{"thread_id":"user-session-123"}}result=graph.invoke(initial_state,config)# 如果崩溃了，同一个 thread_id 可以从上次的 checkpoint 继续result=graph.invoke(None,config)# None = 从上次 checkpoint 恢复

7.2 流式输出：用户不想等

Agent 执行期间用户盯着空白界面，体验很差：

# 每个节点完成后推送状态更新asyncforeventingraph.astream(initial_state):node_name=list(event.keys())[0]yield{"type":"node_update","node":node_name,"status":"completed"}# 如果是 chatbot 节点，流式输出 Tokenifnode_name=="chatbot":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_handler"returnstate# 注册错误处理节点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