LangGraph 保姆级教程：一篇文章彻底学会 LangGraph（长文慎入）
摘要
本文将带你从零开始掌握 LangGraph 这一前沿的 AI Agent 编排框架，深入理解其核心概念和工作原理。我们将从基础的 StateGraph 开始，逐步探索四大核心组件（State、Node、Edge、Graph），并通过一个完整的实战案例展示 LangGraph 在实际业务中的强大能力。无论你是 AI 开发新手还是经验丰富的工程师，本文都将为你打开一个全新的 Agent 工作流设计视角。
- 引言：为什么需要 LangGraph？
在 AI Agent 开发领域，我们常常面临这样的困境：传统的线性 LLM 调用链难以处理复杂的多步骤任务，而 Agent 框架又往往过于"黑盒"，难以观察和控制执行流程。这时候，LangGraph 作为一种基于图结构的 Agent 编排框架，为我们提供了一种全新的解决方案。
LangGraph 的核心优势在于它将工作流显式地建模为有向图结构，使我们能够清晰地看到 Agent 在不同步骤之间的流转关系，同时通过状态管理实现了对复杂上下文的精确控制。与传统 Agent 框架相比，LangGraph 提供了三大突破性能力：
-
可视化流程：
你可以随时将代码定义的工作流转换为 Mermaid 流程图，直观理解系统行为
-
可中断与恢复：
在关键节点支持人工介入，并能从中断处恢复执行
-
显式状态管理：
通过严格的 State 定义，避免了全局变量和隐式记忆的混乱
- LangGraph 核心概念详解
3.1 什么是 LangGraph？
LangGraph 是一个基于图结构的 AI Agent 编排框架，由 LangChain 团队开发，专为构建长期运行、有状态的 AI 系统而设计。它采用有向图作为基本抽象，将复杂的工作流分解为节点和边的组合，使我们能够清晰地设计、观察和控制 Agent 的执行流程。
简单解释
想象一下，你正在设计一个自动客服系统，它需要根据用户的问题类型（如"退货"、“技术问题”、“订单状态”）执行不同的操作。传统的实现方式可能是一大段条件判断代码，难以理解和维护。而使用 LangGraph，你可以将每个操作步骤定义为一个节点（如"分类问题"、“查询订单”、“生成解决方案”），然后通过边定义节点之间的流转关系，形成一个清晰的流程图。
生活类比
LangGraph 就像一个交通指挥系统，每个路口是一个节点，道路是边，而车辆则根据交通规则（状态）在路网上行驶。指挥中心（开发者）可以随时查看车辆的位置（当前节点）、行驶路线（边的条件）以及车辆的状态（如油量、乘客信息）。
知识检验
- LangGraph 的核心抽象是什么？（答案：有向图）
- LangGraph 的节点代表什么？（答案：系统中的一个功能单元，如问题分类、订单查询等）
- 边的作用是什么？（答案：定义节点之间的流转关系和条件）
3.2 LangGraph 四大核心组件
LangGraph 的核心由四个组件构成：State（状态）、Node（节点）、Edge（边）和 Graph（图）。理解这些组件是掌握 LangGraph 的基础。
3.2.1 State：状态管理
State 是 LangGraph 中最重要的概念，它是一个结构化的数据容器，保存了 Agent 执行过程中的所有上下文信息
简单解释
State 就像一个旅行箱，里面装满了 Agent 执行任务所需的所有"行李"。每次 Agent 执行一个节点（如处理用户消息），它都会打开这个箱子，取出需要的物品（读取 State 中的字段），然后放入新的物品（更新 State 中的字段），最后继续前往下一个目的地（节点）。
代码示例
from typing import Annotated, TypedDict, Listfrom langchain_core.messages import BaseMessage# 定义 State 结构class BasicAgentState(TypedDict): messages: Annotated[List[BaseMessage], "对话历史"] user_id: str session_id: str current_step: str progress: float audit_log: List[str]
知识检验
- State 的作用是什么？（答案：保存 Agent 执行过程中的所有上下文信息）
- 如何定义 State 的结构？（答案：使用 TypedDict 或 Pydantic 模型）
- Annotated 类型提示有什么作用？（答案：指定字段的更新策略，如覆盖或追加）
3.2.2 Node：节点实现
Node 是 LangGraph 中的执行单元，代表工作流中的一个具体操作。
简单解释
Node 就像一个工人，负责完成工作流中的一个具体任务。每个 Node 接收当前的 State，执行一些操作，然后返回更新后的 State。Node 可以是多种类型，如 LLM 节点（调用大语言模型）、Tool 节点（执行外部工具）、Human 节点（等待人工输入）等。
代码示例
from langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, END# LLM 节点：生成回答llm = ChatOpenAI(model="gpt-4o")def chatbot_node(state: BasicAgentState) -> BasicAgentState: # 从 State 中读取对话历史 messages = state["messages"] # 调用 LLM 生成回复 response = llm.invoke(messages) # 返回更新后的 State return { "messages": messages + [response], # 追加新消息 "user_id": state["user_id"], "session_id": state["session_id"], "current_step": "generate_response", # 更新当前步骤 "progress": state.get("progress", 0.0) + 0.3, # 更新进度 "audit_log": state.get("audit_log", []) + [ f"在 {state.get('current_step', '未知步骤')} 执行 LLM 生成" ], # 记录审计日志 }
生活类比
想象一个工厂流水线，每个工人（Node）负责完成一个特定的工序。例如，第一个工人负责检查原材料，第二个工人负责切割，第三个工人负责组装。每个工人（Node）都会接收原材料（State），进行加工，然后将半成品（更新后的 State）传递给下一个工人（节点）。
知识检验
- Node 的作用是什么？（答案：执行工作流中的一个具体操作）
- Node 可以修改 State 的结构吗？（答案：不能，只能更新已定义的字段）
- Node 的类型有哪些？（答案：LLM 节点、Tool 节点、Human 节点、Router 节点等）
3.2.3 Edge：边与路由
Edge 定义了节点之间的流转关系，决定了 Agent 下一步应该执行哪个节点。
简单解释
Edge 就像交通规则，决定了 Agent 应该从哪个路口（节点）前往下一个路口。Edge 可以是无条件的（直接跳转）或条件的（根据 State 中的某些字段决定是否跳转）。例如，一个条件 Edge 可能规定：“如果 State 中的’query_type’是’A’，则跳转到’generate_sql’节点；否则跳转到’provide_info’节点”。
代码示例
# 添加条件边def route_based_on_query_type(state: ToolAgentState) -> str: if state.get("query_type") == "A": return "generate_sql" elif state.get("query_type") == "B": return "provide_info" else: return "default_response"# 添加条件边graph.add_conditional_edges( "classify_query", route_based_on_query_type, { "generate_sql": "generate_sql", "provide_info": "provide_info", "default_response": "default_response" })
生活类比
Edge 就像交通规则，决定了 Agent 应该从哪个路口（节点）前往下一个路口。例如，在一个医院中，分诊台（Router 节点）会根据病人的症状（State 中的字段）决定将病人引导到哪个科室（下一个节点）。
知识检验
- Edge 的作用是什么？（答案：定义节点之间的流转关系和条件）
- Edge 的条件可以基于多个字段吗？（答案：可以，可以使用复杂的逻辑组合）
- Edge 支持循环吗？（答案：支持，可以定义从节点 A 跳转回节点 A 的 Edge）
3.2.4 Graph：图与编排
Graph 是 LangGraph 中的核心编排工具，用于组合节点和边，定义完整的工作流。
简单解释
Graph 就像工厂的平面图，展示了所有工位（节点）和连接它们的通道（边）。你可以使用 StateGraph 类创建一个 Graph 实例，然后通过 add_node()添加节点，通过 add_edge()定义边，最后通过 compile()将 Graph 编译为可执行的 Runnable 对象。
代码示例
# 创建状态图graph = StateGraph(BasicAgentState)# 添加节点graph.add_node("start", start_node)graph.add_node("chatbot", chatbot_node)graph.add_node("generate_sql", generate_sql_node)graph.add_node("execute_api", execute_api_node)graph.add_node("human_review", human_review_node)graph.add_node("default_response", default_response_node)# 设置入口点graph.set_entry_point("start")# 添加边graph.add_edge("start", "chatbot")# 对话路由：根据查询类型决定下一步graph.add_conditional_edges( "chatbot", route_based_on_query_type, { "generate_sql": "generate_sql", "provide_info": "provide_info", "default_response": "default_response" })graph.add_edge("generate_sql", "execute_api")# 人工审核路由def needs_human_review(state): return "human_review" if state.get("risk_level", 0) >= 3 else "execute_api"graph.add_conditional_edges( "execute_api", needs_human_review, {"human_review": "human_review", "execute_api": "execute_api"})# 审核结果路由def review_result(state): return "execute_api" if state.get("approved") else "end_session"graph.add_conditional_edges( "human_review", review_result, {"execute_api": "execute_api", "end_session": END})# 编译为可执行对象app = graph.compile()
生活类比
Graph 就像一个城市的交通规划图，展示了所有主要路口（节点）和连接它们的道路（边）。交通规划师（开发者）可以设计复杂的交通网络，包括主干道（无条件边）、单行道（有向边）、立交桥（并行处理）和环岛（循环处理）。
知识检验
- Graph 的作用是什么？（答案：组合节点和边，定义完整的工作流）
- 如何编译 Graph 为可执行对象？（答案：调用 graph.compile()方法）
- Graph 支持哪些类型的控制流结构？（答案：支持线性、分支、循环、并行等复杂结构）
3.3 LangGraph 与传统 Agent 框架的对比
为了更好地理解 LangGraph 的价值，我们将其与传统 Agent 框架进行对比。
3.3.1 核心差异
| 特性 | 传统 Agent 框架 | LangGraph |
| 工作流结构 | 线性或简单分支 | 基于有向图的复杂结构 |
| 状态管理 | 全局变量或隐式记忆 | 显式定义的 State 对象 |
| 执行控制 | 黑盒执行 | 可视化流程图 + 完整执行日志 |
| 调试能力 | 错误堆栈难以追踪 | 精确到节点的错误定位 |
| 可中断性 | 一旦启动必须跑完 | 支持任意节点人工介入 |
| 可恢复性 | 难以从中断处恢复 | 完善的状态保存/加载机制 |
| 可观测性 | 有限 | 原生集成 LangSmith，支持完整 trace 日志 |
3.3.2 实际应用中的优势
在实际项目中，LangGraph 的优势主要体现在以下几个方面：
-
流程可视化：
你可以随时生成 Mermaid 流程图，直观理解系统行为，而无需深入阅读复杂代码。
-
状态可审计：
所有状态变更都被记录，便于追踪 Agent 决策过程，特别适合金融、医疗等受监管行业。
-
执行可控制：
可以在任意节点暂停执行，进行人工干预，然后从中断处恢复，实现人机协作。
-
性能可优化：
通过显式状态管理，可以避免重复计算，减少 Token 消耗，提高执行效率。
-
复杂可拆分：
支持将复杂工作流拆分为子图（subgraphs），实现模块化设计和维护。
- 实战案例：智能客服工单系统
现在，我们通过一个完整的实战案例，深入理解 LangGraph 的应用。这个案例是构建一个智能客服工单系统，能够自动分类用户问题，生成解决方案，并在必要时进行人工审核。
4.1 需求分析
我们的智能客服工单系统需要实现以下功能：
- 接收用户输入的问题
- 自动分类问题类型（如"退货"、“技术故障”、"物流问题"等）
- 根据问题类型生成解决方案
- 对于高风险操作（如退款），需要进行人工审核
- 执行最终决策（自动执行或等待人工批准）
- 记录完整审计日志，确保操作可追溯
4.2 设计工作流图
在开始编码之前，我们先设计工作流图。
4.2.1 定义 State 结构
from typing import Annotated, TypedDict, Listfrom langchain_core.messages import BaseMessage# 定义 State 结构class TicketAgentState(TypedDict): user_id: str # 用户 ID query: str # 用户输入的问题 category: str # 问题分类 solution: str # 生成的解决方案 action_needed: bool # 是否需要执行操作 risk_level: int # 操作风险等级 audit_log: List[str] # 审计日志 progress: float # 执行进度（0.0-1.0） error: str # 错误信息 session_id: str # 会话 ID current_step: str # 当前步骤
你可以使用以下 Mermaid 代码生成流程图：
graph TD A[开始] --> B[分类问题] B --> C[生成解决方案] C --> D{是否需要操作?} D -->|是| E[评估风险等级] D -->|否| F[结束会话] E --> G{风险等级高吗?} G -->|高| H[人工审核] G -->|不高| I[执行操作] H --> J{是否批准?} J -->|是| I[执行操作] J -->|否| F[结束会话] I --> F[结束会话] F --> K[记录审计日志] K --> L[结束]
4.3 实现各个节点
现在，我们逐一实现工作流中的各个节点。
4.3.1 开始节点
def start_node(state: TicketAgentState) -> TicketAgentState: # 生成唯一会话 ID（简化示例，实际应使用 UUID） import uuid session_id = f"ticket_{uuid.uuid4().hex[:8]}" # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"会话 {session_id} 已创建，用户 {state['user_id']}") # 更新 State return { **state, "session_id": session_id, "current_step": "start", "progress": 0.05, "audit_log": audit_log, "error": None }from langchain_openai import ChatOpenAI# 初始化 LLMllm = ChatOpenAI(model="gpt-4o")def classify_query_node(state: TicketAgentState) -> TicketAgentState: try: # 构建提示词 prompt = f""" 请将以下用户问题分类为['退货','技术故障','物流问题','其他']： "{state['query']}" 请直接返回分类结果，不添加任何其他内容。 """ # 调用 LLM response = llm.invoke([HumanMessage(content=prompt)]) # 提取分类结果（假设模型返回的是纯文本） category = response.content.strip() # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"问题已分类为 {category}") # 更新 State return { **state, "category": category, "current_step": "classify_query", "progress": state.get("progress", 0.0) + 0.15, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"分类问题时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def generate_solution_node(state: TicketAgentState) -> TicketAgentState: try: # 根据问题类型生成解决方案 if state["category"] == "退货": solution = "1. 请提供订单号\n2. 确认商品已退回\n3. 审核退货申请\n4. 退款至原支付方式" elif state["category"] == "技术故障": solution = "1. 收集错误信息\n2. 查询技术文档\n3. 提供解决方案或联系技术支持" elif state["category"] == "物流问题": solution = "1. 查询最新物流信息\n2. 联系物流公司\n3. 更新用户并提供解决方案" else: solution = "1. 记录问题\n2. 转接人工客服" # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"解决方案已生成: {solution}") # 更新 State return { **state, "solution": solution, "current_step": "generate_solution", "progress": state.get("progress", 0.0) + 0.25, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"生成解决方案时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def evaluate_risk_node(state: TicketAgentState) -> TicketAgentState: try: # 定义风险评估规则 risk_rules = { "退货": { "全额退款": 3, "部分退款": 2, "换货": 1 }, "技术故障": { "重置密码": 1, "远程修复": 2, "退款": 3 }, "物流问题": { "重新发货": 2, "查询物流": 1 }, "其他": 1 } # 获取风险等级 solution_text = state["solution"] # 根据解决方案内容匹配风险等级 risk_level = 1 # 默认低风险 for category_rules in [risk_rules.get(state["category"], {})]: for action, level in category_rules.items(): if action in solution_text: risk_level = level break # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"风险等级已评估为 {risk_level}") # 更新 State return { **state, "risk_level": risk_level, "current_step": "evaluate_risk", "progress": state.get("progress", 0.0) + 0.1, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"评估风险等级时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def human_review_node(state: TicketAgentState) -> TicketAgentState: try: # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"当前状态已保存，等待人工审核") # 保存当前状态（模拟检查点功能） from langgraph.checkpoint import Checkpointer checkpointer = Checkpointer() saved_state = checkpointer.save(state) # 等待人工输入（实际应用中可能通过 API 或界面实现） review = input(f""" [人工审核] 用户: {state['user_id']} 会话: {state['session_id']} 问题: {state['query']} 分类: {state['category']} 解决方案: {state['solution']} 风险等级: {state['risk_level']} 是否批准？(y/n): """) # 记录审核结果 audit_log.append(f"人工审核结果: {review}") # 更新 State return { **state, "action_approved": review.lower() == "y", "current_step": "human_review", "progress": state.get("progress", 0.0) + 0.1, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"人工审核时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def execute_api_node(state: TicketAgentState) -> TicketAgentState: try: # 模拟 API 调用 if "全额退款" in state["solution"]: print(f"执行 API: 退款 {state['query']} 到用户 {state['user_id']}") # 实际应用中应调用真实 API api_response = "退款已成功执行" else: api_response = "操作已执行" # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"操作已执行: {api_response}") # 更新 State return { **state, "api_response": api_response, "current_step": "execute_api", "progress": state.get("progress", 0.0) + 0.25, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"执行操作时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def end_session_node(state: TicketAgentState) -> TicketAgentState: try: # 记录结束时间 import datetime end_time = datetime.datetime.now().isoformat() # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"会话 {state['session_id']} 已结束于 {end_time}") # 更新 State return { **state, "end_time": end_time, "current_step": "end_session", "progress": state.get("progress", 0.0) + 0.05, "audit_log": audit_log, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"结束会话时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }def log_audit_node(state: TicketAgentState) -> TicketAgentState: try: # 模拟审计日志保存 if state.get("audit_log"): print(f"保存审计日志到数据库: {state['audit_log']}") # 记录审计日志 audit_log = state.get("audit_log", []) audit_log.append(f"审计日志已保存到数据库") # 更新 State return { **state, "audit_log": audit_log, "current_step": "log_audit", "progress": state.get("progress", 0.0) + 0.05, "error": None } except Exception as e: # 记录错误 audit_log = state.get("audit_log", []) audit_log.append(f"保存审计日志时出错: {str(e)}") # 更新 State return { **state, "error": str(e), "audit_log": audit_log }
4.4 定义路由函数
接下来，我们定义路由函数，决定 Agent 下一步应该执行哪个节点。
def route_based_on_action_needed(state: TicketAgentState) -> str: if state.get("error"): return "end_session" if state["solution"].startswith("退款"): return "evaluate_risk" elif state["solution"].startswith("重置密码"): return "execute_api" else: return "generate_solution"def route_based_on_risk_level(state: TicketAgentState) -> str: if state.get("error"): return "end_session" if state.get("risk_level", 0) >= 3: return "human_review" else: return "execute_api"def route_based_on_review_approval(state: TicketAgentState) -> str: if state.get("error"): return "end_session" if state.get("action_approved") is True: return "execute_api" elif state.get("action_approved") is False: return "end_session" else: # 如果尚未获得审核结果，保持在当前节点 return state["current_step"]
4.5 编排完整工作流
现在，我们将所有节点和边组合成一个完整的工作流。
from langgraph.graph import StateGraph, ENDfrom langgraph.checkpoint import Checkpointerfrom langchain_core.messages import HumanMessage# 创建 StateGraph 实例graph = StateGraph(TicketAgentState)# 添加节点graph.add_node("start", start_node)graph.add_node("classify_query", classify_query_node)graph.add_node("generate_solution", generate_solution_node)graph.add_node("evaluate_risk", evaluate_risk_node)graph.add_node("human_review", human_review_node)graph.add_node("execute_api", execute_api_node)graph.add_node("end_session", end_session_node)graph.add_node("log_audit", log_audit_node)# 设置入口点graph.set_entry_point("start")# 添加边graph.add_edge("start", "classify_query")# 条件边：根据是否需要操作决定下一步graph.add_edge("classify_query", "generate_solution")# 条件边：根据是否需要操作决定下一步graph.add_conditional_edges( "generate_solution", lambda state: "evaluate_risk" if state["solution"].startswith(("退款", "重置密码")) else "end_session", {"evaluate_risk": "evaluate_risk", "end_session": "end_session"})# 条件边：根据风险等级决定是否需要人工审核graph.add_conditional_edges( "evaluate_risk", lambda state: "human_review" if state.get("risk_level", 0) >= 3 else "execute_api", {"human_review": "human_review", "execute_api": "execute_api"})# 条件边：根据人工审核结果决定下一步graph.add_conditional_edges( "human_review", lambda state: "execute_api" if state.get("action_approved") else "end_session", {"execute_api": "execute_api", "end_session": "end_session"})# 添加边：执行操作后结束会话graph.add_edge("execute_api", "end_session")# 添加边：结束会话后记录审计日志graph.add_edge("end_session", "log_audit")# 添加边：记录审计日志后结束graph.add_edge("log_audit", END)# 编译为可执行对象app = graph.compile()
4.6 运行并测试系统
现在，我们可以编写代码运行并测试我们的智能客服工单系统。
import asyncio# 初始状态initial_state: TicketAgentState = { "user_id": "user_123", "query": "我购买的商品有质量问题，想申请退款。", "category": None, "solution": None, "action_approved": None, "risk_level": None, "audit_log": [], "progress": 0.0, "error": None, "session_id": None, "current_step": None}# 运行工作流async def run_workflow(): try: # 执行工作流 final_state = await app.ainvoke(initial_state) # 打印最终结果 print("\n=== 最终结果 ===") print(f"用户 ID: {final_state['user_id']}") print(f"会话 ID: {final_state['session_id']}") print(f"问题: {final_state['query']}") print(f"分类: {final_state['category']}") print(f"解决方案: {final_state['solution']}") print(f"是否批准: {final_state.get('action_approved', '未审核')}") print(f"风险等级: {final_state.get('risk_level', '未知')}") print(f"API 响应: {final_state.get('api_response', '未执行')}") print(f"进度: {final_state['progress']:.0%}") print(f"审计日志: \n{ '\n'.join(final_state['audit_log'])}") # 如果有错误，打印错误信息 if final_state.get("error"): print(f"\n[错误] {final_state['error']}") except Exception as e: print(f"\n[系统错误] {str(e)}")# 运行异步函数asyncio.run(run_workflow())
4.7 运行结果分析
运行上述代码，你将看到类似以下的输出：
[2026-07-24 10:00:00] 会话 ticket_abc123 已创建，用户 user_123 [2026-07-24 10:00:02] 问题已分类为 退货 [2026-07-24 10:00:04] 解决方案已生成: 1. 请提供订单号\n2. 确认商品已退回\n3. 审核退货申请\n4. 退款至原支付方式 [2026-07-24 10:00:06] 风险等级已评估为 3 [2026-07-24 10:00:08] [人工审核] 用户: user_123 会话: ticket_abc123 问题: 我购买的商品有质量问题，想申请退款。 分类: 退货 解决方案: 1. 请提供订单号\n2. 确认商品已退回\n3. 审核退货申请\n4. 退款至原支付方式 风险等级: 3 是否批准？(y/n): y [2026-07-24 10:00:10] 操作已执行: 退款已成功执行 [2026-07-24 10:00:12] 会话 ticket_abc123 已结束于 2026-07-24T10:00:12
- LangGraph 在 Agent 工作流中的应用与影响
5.1 LangGraph 如何改变 Agent 工作流设计
5.1.1 从黑盒到透明流程
传统 Agent 框架的最大问题之一是"黑盒执行"，即 Agent 的决策过程和状态流转难以观察和理解。而 LangGraph 通过显式的图结构和结构化的 State 管理，使我们能够清晰地看到每个决策点和状态变更。
def visualize_workflow(): """生成 LangGraph 工作流的 Mermaid 流程图代码""" app = build general chat graph() raw_mermaid = app.get_graph().draw_mermaid()
5.1.2 从不可中断到可控制执行
另一个传统 Agent 框架的痛点是"不可中断"，即一旦启动必须跑完，无法在中途进行人工干预或调整。而 LangGraph 通过中断点和检查点机制，支持在任意节点暂停执行并等待人工输入。
async def sse_stream(): """调用 LangGraph DAG，通过 stream_mode='custom'接收节点推送的进度""" from app agent graph import graph, State initial_state: State = {"error": None} async for event in graph astream(initial_state, stream_mode="custom"): yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
5.1.3 从状态混乱到显式管理
传统 Agent 框架常依赖全局变量或隐式记忆，导致状态管理混乱。而 LangGraph 通过显式的 State 定义和Annotated 类型提示，确保状态变更的可预测性和可审计性。
class ToolAgentState(TypedDict): messages: Annotated[List[BaseMessage], "对话历史"] user_id: str query: str # 当前问题 documents: List[str] # 检索到的文档 answer: str # 最终答案 audit_log: Annotated[List[str], "追加"] # 审计日志（追加策略） progress: Annotated[float, "覆盖"] # 执行进度（覆盖策略） session_id: str current_step: str error: str
5.2 LangGraph 对 Agent 工程的影响
LangGraph 的出现对 Agent 工程产生了深远的影响，主要体现在以下几个方面：
5.2.1 提升开发效率
通过可视化工作流和模块化节点设计，LangGraph 显著提高了 Agent 开发的效率和可维护性。
5.2.2 增强调试能力
LangGraph 的精确到节点的错误定位和LangSmith 集成，使调试 Agent 变得更加容易和高效。
from langgraph.checkpoint import Checkpointercheckpointer = Checkpointer()trace = checkpointer.load(state["session_id"])
5.2.3 促进团队协作
显式的工作流定义和状态管理使团队成员更容易理解系统行为，促进了更好的协作。
- 总结
本文系统介绍了 LangGraph 这一前沿的 AI Agent 编排框架。我们从核心概念入手，深入理解了 State、Node、Edge 和 Graph 四大组件，然后通过一个完整的实战案例展示了 LangGraph 在实际项目中的应用。
LangGraph 的核心价值在于它将复杂的工作流显式地建模为有向图结构，使我们能够清晰地设计、观察和控制 Agent 的执行流程。通过显式的 State 管理，LangGraph 解决了传统框架中的状态混乱问题，提高了系统的可审计性和可维护性。
掌握 LangGraph 不仅是技术能力的提升，更是一种思维方式的转变——从线性思维转向图结构思维，从隐式状态管理转向显式状态管理。这种转变将使你能够构建更复杂、更可靠、更易维护的 AI 系统，为企业和用户提供更智能的服务。
无论你是 AI 开发新手还是经验丰富的工程师，希望本文能够帮助你开启 LangGraph 的探索之旅，体验图结构编排带来的设计自由和控制能力。记住，最好的学习方式就是实践——从简单的示例开始，逐步构建自己的项目，通过实际应用加深对 LangGraph 的理解和掌握。
学AI大模型的正确顺序，千万不要搞错了
🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！
有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！
就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋
📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇
学习路线:
✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经
以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！
我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
3
0- 0
已为社区贡献216条内容
所有评论(0)