【大模型实战】LangGraph入门教程：从零搭建可控AI工作流，旅游规划助手实例解析6大核心概念！
本文通过旅游规划助手实例，系统介绍LangGraph六大核心知识点：State作为全局数据源、Node作为功能模块、流程控制(顺序/分支/循环/并行)、工具调用、状态持久化及人工干预。采用极简完整示例，帮助开发者快速建立对LangGraph整体认知，理解AI工作流的设计与实现机制，为构建复杂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
}
)
这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
📌一句话总结：通过条件分支，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大模型？
如果你对AI大模型入门感兴趣，那么你需要的话可以点击这里大模型重磅福利：入门进阶全套104G学习资源包免费分享！
这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
这是一份大模型从零基础到进阶的学习路线大纲全览，小伙伴们记得点个收藏！
第一阶段： 从大模型系统设计入手，讲解大模型的主要方法；
第二阶段： 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；
第三阶段： 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；
第四阶段： 大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；
第五阶段： 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；
第六阶段： 以SD多模态大模型为主，搭建了文生图小程序案例；
第七阶段： 以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。
100套AI大模型商业化落地方案
大模型全套视频教程
200本大模型PDF书籍
👉学会后的收获：👈
• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；
• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；
• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；
• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。
LLM面试题合集
大模型产品经理资源合集
大模型项目实战合集
👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
14
0- 0
已为社区贡献367条内容
所有评论(0)