本文通过构建智能旅游规划系统，详细介绍了LangGraph的六大核心概念：State作为全局数据源、Node作为功能模块、流程控制（顺序/分支/循环/并行）、工具调用实现AI执行能力、状态持久化支持流程暂停恢复，以及人工干预引入用户决策。文章以极简示例直观展示各组件如何协作，帮助开发者快速掌握LangGraph整体框架和运行机制，为构建复杂AI工作流提供实践参考。
越来越多的人开始使用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可以在自动执行过程中 引入用户决策点。总结
本文基于LangGraph 1.0构建了一个完整的智能旅游规划系统，主要展示了State、Node、流程控制、工具调用、State持久化与人工干预等工作流的具体实现，让初学LangGraph的同学能够系统性地了解 LangGraph的核心概念与运行机制，理解复杂AI工作流业务是如何被拆解为清晰、可控的图结构流程，为自己的AI应用与智能系统设计提供实践参考。
AI时代，未来的就业机会在哪里？
答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具，到自然语言处理、计算机视觉、多模态等核心领域，技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。
掌握大模型技能，就是把握高薪未来。
那么，普通人如何抓住大模型风口？
AI技术的普及对个人能力提出了新的要求，在AI时代，持续学习和适应新技术变得尤为重要。无论是企业还是个人，都需要不断更新知识体系，提升与AI协作的能力，以适应不断变化的工作环境。
因此，这里给大家整理了一份《2025最新大模型全套学习资源》，包括2025最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题等，带你从零基础入门到精通，快速掌握大模型技术！
由于篇幅有限，有需要的小伙伴可以扫码获取！
1. 成长路线图&学习规划
要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。
2. 大模型经典PDF书籍
书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）
3. 大模型视频教程
对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。
4. 大模型项目实战
学以致用，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。
5. 大模型行业报告
行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。
6. 大模型面试题
面试不仅是技术的较量，更需要充分的准备。
在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。
为什么大家都在学AI大模型？
随着AI技术的发展，企业对人才的需求从“单一技术”转向 “AI+行业”双背景。企业对人才的需求从“单一技术”转向 “AI+行业”双背景。金融+AI、制造+AI、医疗+AI等跨界岗位薪资涨幅达30%-50%。
同时很多人面临优化裁员，近期科技巨头英特尔裁员2万人，传统岗位不断缩减，因此转行AI势在必行！
这些资料有用吗？
这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。
大模型全套学习资料已整理打包，有需要的小伙伴可以
微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】
