越来越多的人开始使用LangGraph搭建可控、可观测、可插拔的AI工作流。无论是智能客服、多步骤智能体，还是自动化运维机器人、AI建站系统，它都能游刃有余地实现。
然而，许多开发者在初次接触LangGraph时，往往会感到无从下手，想快速入门，却又缺乏对其整体框架和运行机制的系统性认知。
因此，本文的目标是：用最短的时间，帮助大家建立对LangGraph的“整体认知”。
全文不绕弯、不堆概念，而是通过一个极简但完整的示例，直观展示LangGraph中以下六个核心知识点的具体实现：State、Node、流程控制、工具调用、State 持久化，以及人工干预。
通过本文，让大家轻松了解掌握：
- Graph是怎么搭建和运作的
- 节点、状态、工具、流程之间是什么关系
- 如何编排顺序、分支、循环、并行业务流
本文以一个旅游规划助手为例，整体框架如下：
用户：我想去云南旅游5天，预算8000
流程：
开始 → 解析意图 → 并行查询 → 汇总结果 → 预算评估
↓
预算优化循环（最多3次）←→ 人工干预（条件分支）
↓
生成最终行程 → 文件写入 → 结束
↓
中断恢复支持（任意节点可恢复）
说明：
解析意图：调用大模型解析用户的输入，分析想去的地点、天数、预算，如果缺少关键信息需要返回让用户补充
顺序验证：预算、目的、时间、个人信息
并行查询：
查询机票信息
查询酒店信息
查询景点信息
汇总查询：汇总打印并行执行返回的内容
循环节点：预算优化、行程优化
调用大模型生成行程表
调用工具保存
接下来我们逐个拆解这些流程在LangGraph中的具体实现。
1、State（状态）：流程的“唯一数据源”
State就是整个业务流程的“数据背包”，所有Node、工具等都从这里读数据、写结果。
class TravelState(TypedDict): """旅游规划状态""" messages: Annotated[Sequence[BaseMessage], add_messages] input: Optional[str] # 用户输入查询 travel_info: Optional[Dict[str, Any]] # 旅行基本信息：destination, days, budget, travel_date, travelers, requirements query_results: Optional[Dict[str, Any]] # 合并所有查询结果 cost_analysis: Optional[Dict[str, Any]] # 合并成本分析 itinerary: Optional[str] # 生成的行程 status: str # 当前状态 _control: Optional[Dict[str, Any]] # 流程控制字段 # 并行查询结果键 flight_info: Optional[Dict[str, Any]] # 航班查询结果 hotel_info: Optional[Dict[str, Any]] # 酒店查询结果 attractions_info: Optional[Dict[str, Any]] # 景点查询结果无论是查机票、查酒店、查景点、评估预算、生成行程表，一切数据都在State中流动。
📌一句话总结：State是流程的“唯一上下文”，所有节点靠它协作，其本质就是一个全局可共享的上下文字典。
2、Node（节点）：每一Node都是一个“功能模块”
Node是Graph的基本执行单元，可以把它理解为“流程步骤函数”。
"""创建旅游规划Graph""" workflow = StateGraph(TravelState) # 添加核心处理节点 workflow.add_node("parse_intent", node_parse_intent) # 解析用户意图，提取旅游需求信息 # 🔄 顺序执行节点链 - 旅行前置验证流程 workflow.add_node("validate_budget", node_validate_budget) # 1️⃣ 预算验证 workflow.add_node("check_destination", node_check_destination) # 2️⃣ 目的地可行性检查 workflow.add_node("verify_travel_time", node_verify_travel_time) # 3️⃣ 时间可行性检查 workflow.add_node("check_documents", node_check_documents) # 4️⃣ 个人信息验证 # 并行执行节点 workflow.add_node("query_flights", node_query_flights) # 并行查询航班信息 workflow.add_node("query_hotels", node_query_hotels) # 并行查询酒店信息 workflow.add_node("query_attractions", node_query_attractions) # 并行查询景点信息 workflow.add_node("aggregate_results", node_aggregate_parallel_results) # 汇总并行查询结果 # 🔄 循环执行节点 - 预算优化循环 workflow.add_node("budget_optimization", node_budget_optimization) # 预算优化处理 workflow.add_node("check_budget_satisfaction", node_check_budget_satisfaction) # 检查预算满意度 # 🔄 循环执行节点 - 行程优化循环 workflow.add_node("itinerary_optimization", node_itinerary_optimization) # 行程优化处理 workflow.add_node("check_itinerary_satisfaction", node_check_itinerary_satisfaction) # 检查行程满意度 workflow.add_node("human_intervention", node_human_intervention) # 人工干预处理 workflow.add_node("generate_itinerary", node_generate_itinerary) # 生成最终旅游行程一个Node专注一件事：意图解析、预算验证、航班查询、酒店查询、景点查询……，模块越拆越细，后期编排流程越灵活。
📌一句话总结：Node = 单功能步骤函数。
3、Control Flow（流程控制）：让智能体能决定下一步
LangGraph支持多种流程控制方式，也就是Node的“组织方式”。
①顺序执行：A → B → C（最常见）
让 Node A → Node B → Node C 按顺序执行：
最简单的流水线：
# 🔄 顺序执行链 - 旅行前置验证流程 print("📚 配置顺序执行链：预算验证 → 目的地检查 → 时间验证 → 文件检查") workflow.add_edge("validate_budget", "check_destination") # 1️⃣ → 2️⃣ workflow.add_edge("check_destination", "verify_travel_time") # 2️⃣ → 3️⃣ workflow.add_edge("verify_travel_time", "check_documents") # 3️⃣ → 4️⃣ workflow.add_edge("check_documents", "start_parallel") # 4️ → 下一个节点📌一句话总结：顺序流程适用于有明确固定业务流程的业务。
②条件分支：根据状态决定下一步
# 解析意图后的路由 def after_parse_router(state: TravelState) -> str: control = state.get("_control", {}) status = state.get("status", "") print(f"🔍 路由检查: status={status}, control={control}") if status == "collecting_info": return END # 需要用户输入，暂停流程 elif status == "planning" and not control.get("validation_completed"): return "validate_budget" # 🔄 开始顺序执行的前置验证流程 elif status == "planning" and control.get("validation_completed"): print("✅ 验证已完成，跳过重复验证") return "start_parallel" # 直接进入并行查询 elif status == "processing" and not control.get("parsed_attempted"): return "parse_intent" # 首次解析 elif status == "processing" and control.get("user_confirmed"): return "generate_itinerary" # 用户已确认，生成行程 elif status == "continuing": # 人工干预后继续流程 if control.get("human_intervention_completed"): print("✅ 人工干预完成，继续生成行程") return "generate_itinerary" else: print("⚠️ 人工干预状态异常") return END else: # 避免无限循环，如果状态不明确就结束 print(f"⚠️ 未知状态，结束流程: status={status}") return END workflow.add_conditional_edges( "parse_intent", after_parse_router, { "parse_intent": "parse_intent", "validate_budget": "validate_budget", # 🔄 开始顺序验证流程 "start_parallel": "start_parallel", # 🔄 跳过验证，直接并行查询 "generate_itinerary": "generate_itinerary", END: END } )📌一句话总结：通过条件分支，Graph可以根据运行时状态，决定流程的走向。
③循环：自动反复直到通过# 🔄 预算优化循环逻辑 def budget_satisfaction_router(state: TravelState) -> str: """预算优化循环的条件路由 - 示例：条件分支判断""" control = state.get("_control", {}) budget_attempts = control.get("budget_optimization_attempts", 0) budget_satisfied = control.get("budget_satisfied", False) needs_human_intervention = control.get("needs_human_intervention", False) cost_analysis = state.get("cost_analysis", {}) is_over_budget = cost_analysis.get("is_over_budget", False) print(f"💰 [条件分支判断] 预算循环路由决策:") print(f" 🔄 尝试次数: {budget_attempts}/3") print(f" ✅ 预算满意: {budget_satisfied}") print(f" ⚠️ 超出预算: {is_over_budget}") print(f" 👤 需要人工干预: {needs_human_intervention}") # 条件分支的优先级判断 if budget_satisfied: print(" ➡️ 路由决策: 预算满意 → 进入行程优化") return "itinerary_optimization" # 预算满意，进入行程优化 elif needs_human_intervention or (budget_attempts >= 3 and is_over_budget): print(" ➡️ 路由决策: 需要人工干预或达到最大尝试次数且仍超预算 → 人工干预") return "human_intervention" # 需要人工干预 elif budget_attempts >= 3: print(" ➡️ 路由决策: 达到最大尝试次数但预算可接受 → 进入行程优化") return "itinerary_optimization" # 强制进入下一阶段 else: print(" ➡️ 路由决策: 继续预算优化") return "budget_optimization" # 继续优化 workflow.add_edge("budget_optimization", "check_budget_satisfaction") workflow.add_conditional_edges( "check_budget_satisfaction", budget_satisfaction_router, { "budget_optimization": "budget_optimization", # 继续优化 "itinerary_optimization": "itinerary_optimization", # 进入行程优化 "human_intervention": "human_intervention" # 人工干预 } ) 这就是一个天然的循环，不需要while。 📌一句话总结：循环本质是节点再次指向自己或前序节点。④并行：同一条输入，多路同时计算
例如：根据用户的意图同时查询机票、酒店、景点
# LangGraph原生并行：从start_parallel同时启动3个查询节点 workflow.add_edge("start_parallel", "query_flights") workflow.add_edge("start_parallel", "query_hotels") workflow.add_edge("start_parallel", "query_attractions") # 所有并行查询完成后汇总结果 workflow.add_edge("query_flights", "aggregate_results") workflow.add_edge("query_hotels", "aggregate_results") workflow.add_edge("query_attractions", "aggregate_results")📌一句话总结：并行流程允许多个节点基于同一份 State 同时执行，并在汇总节点统一合并各自的执行结果。
4、工具调用（Tool / MCP）：智能体真正“能做事”
将生成的旅游规划内容保存到文件中：
# 创建统一的工具节点，包含所有工具 all_tools = [write_itinerary_to_file] tool_node = ToolNode(all_tools) # 行程生成后写入文件 workflow.add_edge("generate_itinerary", "write_itinerary_file") # 写入文件节点生成工具调用后，交给ToolNode执行 workflow.add_edge("write_itinerary_file", "tool_node") # ToolNode执行完成后结束 workflow.add_edge("tool_node", END)扩展：如果想让LLM自动决定是否调用工具，请使用：
response=llm.invoke( {"messages": state["messages"]}, tools=[write_itinerary_to_file] )注意：模型只返回调用的工具名和参数内容，实际需要ToolNode执行具体的工具
📌一句话总结：工具让AI从“说话”变成真正“执行”。****
5、状态持久化：流程可暂停、可恢复、可追踪
一句配置让Graph变成“有记忆系统”的工作流引擎：async with AsyncSqliteSaver.from_conn_string("travel_planning.db") as checkpointer: app = workflow.compile(checkpointer=checkpointer)📌一句话总结：状态持久化让Graph能够跨执行周期保存状态，从而支持暂停、恢复和审计。
6、人工干预
遇到需要用户确认的信息，智能体可以暂停等人工确认：
def node_human_intervention(state: TravelState) -> TravelState: """👤 人工干预节点 - 预算超支时的用户决策点""" print("\n" + "="*60) print("📚 [条件分支] 👤 人工干预处理") print("="*60) messages = state.get("messages", []) cost_analysis = state.get("cost_analysis", {}) total_cost = cost_analysis.get("total_cost", 0) budget = cost_analysis.get("budget", 0) control = state.get("_control", {}) or {} print(f"💰 当前总费用: {total_cost:,}元") print(f"🎯 用户预算: {budget:,}元") print(f"📊 超支金额: {total_cost - budget:,}元") # 检查是否为非交互模式 interactive_mode = control.get("interactive_mode", True) if not interactive_mode: print("🤖 非交互模式：自动应用优化建议") # 在非交互模式下，自动接受优化建议 control["user_confirmed"] = True control["user_choice"] = "accept" # 检查是否已经处理过用户确认 if control.get("user_confirmed"): # 用户已确认，应用相应方案 user_choice = control.get("user_choice", "accept") if user_choice == "accept": # 应用优化建议 overspend = total_cost - budget overspend_ratio = overspend / budget # 根据超支比例确定优化幅度 if overspend_ratio > 0.3: reduction_rate = 0.25 # 大幅优化 adjustments = [ "调整目的地为性价比更高的城市", "缩短行程天数（减少2天）", "选择经济型住宿" ] elif overspend_ratio > 0.2: reduction_rate = 0.20 # 中等优化 adjustments = [ "选择经济型酒店（节省15%）", "调整航班时间（非节假日出行）", "减少部分自费项目" ] else: reduction_rate = 0.15 # 轻度优化 adjustments = [ "减少购物预算", "选择部分免费景点", "优化餐饮预算" ] adjusted_total = total_cost * (1 - reduction_rate) human_adjustment = { "suggestions": control.get("suggestions", []), "adjusted_total": adjusted_total, "reduction_rate": reduction_rate, "adjustments": adjustments, "advisor_note": f"已根据预算优化方案，减少{reduction_rate*100:.0f}%费用，确保核心体验不受影响" } updated_cost_analysis = { **cost_analysis, "total_cost": adjusted_total, "is_over_budget": adjusted_total > budget, "human_adjustment": human_adjustment } print("✅ 用户选择：接受优化建议") print(f"🛠️ 应用优化方案，费用从 {total_cost:,}元 降至 {adjusted_total:,}元") return { **state, "cost_analysis": updated_cost_analysis, "status": "planning", "_control": {**control, "optimization_applied": True, "human_intervention_completed": True}, "messages": messages + [ AIMessage(content=f""" ✅ 已应用优化方案： 🛠️ 具体调整： {chr(10).join([f"• {adj}" for adj in adjustments])} 💰 调整后总花费：{adjusted_total:,}元 📉 节省金额：{total_cost - adjusted_total:,}元 📝 备注：{human_adjustment.get('advisor_note', '')} ✅ 优化完成，继续生成行程表... """) ] } elif user_choice == "reject": # 用户拒绝优化，终止规划 print("❌ 用户选择：拒绝继续，终止规划") return { **state, "status": "terminated", "_control": {**control, "human_intervention_completed": True, "planning_terminated": True}, "messages": messages + [ AIMessage(content=""" ❌ 已终止旅游规划： 📝 由于预算限制，用户选择不继续当前规划。 💡 建议：可以考虑调整预算或旅游需求后重新规划。 感谢使用智能旅游规划系统！ """) ] } else: # 用户选择保持原方案，继续规划 print("📝 用户选择：保持原方案，继续规划") return { **state, "status": "planning", "_control": {**control, "optimization_applied": True, "human_intervention_completed": True}, "messages": messages + [ AIMessage(content=f""" 📝 已保持原方案，继续规划： 💰 总花费：{total_cost:,}元 🎯 预算：{budget:,}元 ⚠️ 超支：{total_cost - budget:,}元 📋 将按原方案继续生成详细行程表... """) ] } # 首次进入，生成优化建议并等待用户确认 overspend = total_cost - budget overspend_ratio = overspend / budget suggestions = [f"当前超支 {overspend:,}元"] # 根据超支比例给出建议 if overspend_ratio > 0.3: suggestions.extend([ "建议：调整目的地或缩短行程天数", "预计可节省：25%费用" ]) elif overspend_ratio > 0.2: suggestions.extend([ "建议：选择经济型酒店，节省约800-1500元", "建议：调整航班时间（非节假日出行）", "预计可节省：20%费用" ]) else: suggestions.extend([ "建议：减少购物预算或选择部分免费景点", "预计可节省：15%费用" ]) print(f"🤖 生成优化建议: {suggestions}") print("⏳ 等待用户决策...") # 等待用户确认 return { **state, "status": "waiting_confirmation", "_control": { **control, "waiting_confirmation": True, "suggestions": suggestions, "overspend": overspend, "overspend_ratio": overspend_ratio }, "messages": messages + [ AIMessage(content=f""" ⚠️ 预算超支提醒： 📊 当前情况： • 总花费：{total_cost:,}元 • 预算：{budget:,}元 • 超支：{overspend:,}元 ({overspend_ratio*100:.1f}%) 💡 优化建议： {chr(10).join([f"• {s}" for s in suggestions])} 🤔 请选择您的决策： 1. 接受优化建议（输入"接受"或"1"） 2. 保持原方案继续（输入"保持"或"2"） 3. 终止规划（输入"终止"或"3"） 请输入您的选择： """) ] } 📌一句话总结：通过人工干预，Graph可以在自动执行过程中 引入用户决策点。那么，如何系统的去学习大模型LLM？
作为一名深耕行业的资深大模型算法工程师，我经常会收到一些评论和私信，我是小白，学习大模型该从哪里入手呢？我自学没有方向怎么办？这个地方我不会啊。如果你也有类似的经历，一定要继续看下去！这些问题啊，也不是三言两语啊就能讲明白的。
所以我综合了大模型的所有知识点，给大家带来一套全网最全最细的大模型零基础教程。在做这套教程之前呢，我就曾放空大脑，以一个大模型小白的角度去重新解析它，采用基础知识和实战项目相结合的教学方式，历时3个月，终于完成了这样的课程，让你真正体会到什么是每一秒都在疯狂输出知识点。
由于篇幅有限，⚡️ 朋友们如果有需要全套 《2025全新制作的大模型全套资料》，扫码获取~
👉大模型学习指南+路线汇总👈
我们这套大模型资料呢，会从基础篇、进阶篇和项目实战篇等三大方面来讲解。
👉①.基础篇👈
基础篇里面包括了Python快速入门、AI开发环境搭建及提示词工程，带你学习大模型核心原理、prompt使用技巧、Transformer架构和预训练、SFT、RLHF等一些基础概念，用最易懂的方式带你入门大模型。
👉②.进阶篇👈
接下来是进阶篇，你将掌握RAG、Agent、Langchain、大模型微调和私有化部署，学习如何构建外挂知识库并和自己的企业相结合，学习如何使用langchain框架提高开发效率和代码质量、学习如何选择合适的基座模型并进行数据集的收集预处理以及具体的模型微调等等。
👉③.实战篇👈
实战篇会手把手带着大家练习企业级的落地项目（已脱敏），比如RAG医疗问答系统、Agent智能电商客服系统、数字人项目实战、教育行业智能助教等等，从而帮助大家更好的应对大模型时代的挑战。
👉④.福利篇👈
最后呢，会给大家一个小福利，课程视频中的所有素材，有搭建AI开发环境资料包，还有学习计划表，几十上百G素材、电子书和课件等等，只要你能想到的素材，我这里几乎都有。我已经全部上传到CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
相信我，这套大模型系统教程将会是全网最齐全 最易懂的小白专用课！！
