LangGraph 核心概念

LangGraph是LangChain 生态的 “进阶编排框架”是 AgentExecutor升级版基于图结构解决复杂工作流 / 多智能体问题兼容 LangChain 所有组件。AgentExecutor 是「单智能体固定循环执行器」适合简单线性任务LangGraph 是「多步骤 / 多智能体图状编排框架」适合复杂分支 / 协作任务LangGraph的特性图结构每个任务步骤是「节点」步骤间的跳转规则是「边」支持 if/else 分支、循环、并行执行状态管理全局维护一个 “状态State”所有节点可读写状态解决 AgentExecutor 上下文传递不灵活的问题多智能体协作轻松实现 “分工协作”如 “分析师智能体→审核智能体→总结智能体”持久化与中断恢复支持暂停 / 恢复工作流甚至重启后继续执行AgentExecutor 中断即从头再来兼容 LangChain 生态可直接复用 LangChain 的 LLM、Tool、Prompt 等组件。LangChain AgentExecutor和LangGraph 核心区别维度LangChain AgentExecutorLangGraph核心定位单智能体「思考 - 行动」循环的执行引擎复杂工作流 / 多智能体的编排框架执行逻辑固定循环思考→行动→反馈→思考单一路径图结构节点步骤 边跳转规则支持分支 / 并行 / 多智能体状态管理简单上下文依赖 Memory无全局状态全局可读写的 State所有节点共享 / 修改状态复杂度支持适合单工具 / 少步骤的简单任务如查天气 推荐景点适合多分支 / 多智能体 / 长流程如财报分析→风险审核→报告生成灵活性低循环逻辑固定仅能通过参数如 max_iterations微调高自定义节点跳转规则支持动态分支、条件判断多智能体支持弱仅能单智能体执行强原生支持多智能体分工、通信、协作持久化 / 中断恢复无中断后需重新执行有支持 checkpoint可暂停 / 恢复工作流学习 / 使用成本低几行代码即可跑通中高需理解图结构、状态定义、节点跳转典型场景单智能体 少工具的简单任务如查快递、计算器多步骤分支任务如 “用户提问→判断语种→对应翻译→审核”、多智能体协作如 “研究员 分析师 总结员”代码示例智能旅游助手-LangGraph 实现版填写api_key可直接运行# 安装依赖新增langgraph # pip install langchain langchain_community jwt python-dotenv langgraph import os # 第一步添加密钥配置 import warnings # 1. 屏蔽残留的警告双重保障 warnings.filterwarnings(ignore, categoryUserWarning, modulejwt) # 2. 设置LangGraph的安全密钥32字节符合SHA256要求 os.environ[LANGGRAPH_SECRET_KEY] your_secure_32bytes_key_here12345678 # 必须32位以上 from dotenv import load_dotenv from typing import TypedDict, Annotated import operator # 1. 导入核心模块LangGraph LangChain 生态复用 from langchain_community.chat_models import ChatZhipuAI from langchain_openai import ChatOpenAI from langchain_core.tools import Tool from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode # llm解析 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_core.exceptions import OutputParserException # 2. 加载环境变量 初始化LLM load_dotenv() llm ChatZhipuAI( modelglm-4, temperature0.3, api_key自己的api_key ) # 3. 定义工具复用原AgentExecutor的工具逻辑 # 工具1天气查询 def get_weather(city: str, date: str) - str: weather_data { 北京-2026-03-18: 晴10-22℃微风, 上海-2026-03-18: 多云12-20℃东风3级 } key f{city}-{date} return weather_data.get(key, f未查询到{city}{date}的天气信息) # 工具2景点推荐 def recommend_scenic_spots(city: str, weather: str) - str: if 晴 in weather: return f{city}晴天推荐故宫、颐和园、天安门广场 elif 雨 in weather: return f{city}雨天推荐国家博物馆、中国科技馆 else: return f{city}多云推荐什刹海、南锣鼓巷 # 工具3机票查询 def check_flight(from_city: str, to_city: str, date: str) - str: flight_data { 上海-北京-2026-03-18: 上海→北京2026-03-18\n1. 东航MU510108:00-10:30800元\n2. 国航CA150214:00-16:40750元 } key f{from_city}-{to_city}-{date} return flight_data.get(key, f未查询到{from_city}→{to_city}{date}的机票信息) # 封装工具列表LangGraph可直接复用LangChain Tool tools [ Tool( nameWeatherQuery, funcget_weather, description查询指定城市指定日期的天气参数city城市、dateYYYY-MM-DD ), Tool( nameScenicSpotRecommendation, funcrecommend_scenic_spots, description根据城市天气推荐景点参数city城市、weather天气描述 ), Tool( nameFlightQuery, funccheck_flight, description查询机票参数from_city出发地、to_city目的地、dateYYYY-MM-DD ) ] tool_node ToolNode(tools) # LangGraph工具执行节点 # 4. 定义全局状态核心替代AgentExecutor的Memory # 所有节点可读写该状态实现上下文共享 class TravelState(TypedDict): user_input: str # 用户原始输入 city: str # 目的地城市 date: str 2026-03-18 # 默认日期 from_city: str # 出发城市机票用 weather: str # 天气结果 spots: str # 景点结果 flight: str # 机票结果 next_step: str # 下一步执行的节点 final_answer: str # 最终回答 # 定义默认值字典集中管理方便修改 TRAVEL_STATE_DEFAULTS { city: , date: 2026-03-18, from_city: , weather: , spots: , flight: , next_step: , final_answer: } # llm解析state def llm_parse(user_input: str) - TravelState: # -------------------------- # 关键用PromptLLM提取关键信息 # -------------------------- # 定义Prompt模板明确要求LLM输出JSON格式 prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅游信息提取助手需要从用户输入中提取以下信息输出JSON格式 1. from_city出发城市如无则为空字符串 2. city目的地城市如无则为空字符串 3. date出行日期格式YYYY-MM-DD如无则用默认值2026-03-18 注意 - 只输出JSON不要加任何额外解释 - 日期尽量从用户输入中提取比如明天对应2026-03-18后天对应2026-03-19 - 城市名必须是完整名称如上海而非沪 ), (user, 用户输入{user_input}) ]) # 创建JSON解析器 parser JsonOutputParser() # 组装ChainPrompt LLM 解析器 extract_chain prompt | llm | parser state: TravelState {user_input: user_input} try: # 调用LLM提取信息 extracted_info extract_chain.invoke({user_input: user_input}) # 将提取的信息赋值到State覆盖默认值 if extracted_info.get(from_city): state[from_city] extracted_info[from_city] if extracted_info.get(city): state[city] extracted_info[city] if extracted_info.get(date): state[date] extracted_info[date] except OutputParserException as e: # 解析失败时用默认值兜底 print(fLLM输出解析失败{e}使用默认值) except Exception as e: print(fLLM调用失败{e}使用默认值) return state # 5. 定义核心节点每个节点对应一个步骤替代Agent的“思考” # 节点1解析用户输入提取关键信息 def parse_input(state: TravelState) - TravelState: user_input state[user_input] state llm_parse(user_input) state[next_step] get_weather # 解析后下一步查天气 return state # 节点2查询天气 def get_weather_node(state: TravelState) - TravelState: city state[city] date state[date] state[weather] get_weather(city, date) state[next_step] recommend_spots # 下一步推荐景点 return state # 节点3推荐景点 def recommend_spots_node(state: TravelState) - TravelState: city state[city] weather state[weather] state[spots] recommend_scenic_spots(city, weather) # 分支逻辑有出发城市则查机票否则直接生成答案 if state[from_city]: state[next_step] check_flight else: state[next_step] generate_answer return state # 节点4查询机票 def check_flight_node(state: TravelState) - TravelState: from_city state[from_city] to_city state[city] date state[date] state[flight] check_flight(from_city, to_city, date) state[next_step] generate_answer # 下一步生成最终答案 return state # 节点5生成最终回答 def generate_answer_node(state: TravelState) - TravelState: weather state[weather] spots state[spots] flight state[flight] # 整合结果 final_answer f你的旅行建议\n1. 天气{weather}\n2. 景点{spots} if flight: final_answer f\n3. 机票{flight} state[final_answer] final_answer state[next_step] END # 结束流程 return state # 6. 定义节点跳转逻辑替代AgentExecutor的固定循环 def get_next_node(state: TravelState): return state[next_step] # 7. 构建LangGraph图核心节点边 graph_builder StateGraph(TravelState) # 添加节点每个步骤对应一个节点 graph_builder.add_node(parse_input, parse_input) graph_builder.add_node(get_weather, get_weather_node) graph_builder.add_node(recommend_spots, recommend_spots_node) graph_builder.add_node(check_flight, check_flight_node) graph_builder.add_node(generate_answer, generate_answer_node) # 设置起始节点 graph_builder.set_entry_point(parse_input) # 添加边自定义跳转规则 # 方式1固定边解析→查天气→推荐景点 graph_builder.add_edge(parse_input, get_weather) graph_builder.add_edge(get_weather, recommend_spots) # 方式2条件边推荐景点后根据状态跳转到机票/生成答案 graph_builder.add_conditional_edges( recommend_spots, get_next_node, # 跳转规则函数 { check_flight: check_flight, generate_answer: generate_answer } ) graph_builder.add_edge(check_flight, generate_answer) graph_builder.add_edge(generate_answer, END) # 生成答案后结束 # 8. 编译并运行图 app graph_builder.compile() # 测试运行 if __name__ __main__: # 输入包含机票的综合需求 input_state { user_input: 我明天从上海飞北京玩推荐适合的景点并查一下机票 } # 运行图LangGraph核心调用方式 result app.invoke(input_state) print( LangGraph 运行结果 ) print(result[final_answer])运行结果 LangGraph 运行结果 你的旅行建议 1. 天气晴10-22℃微风 2. 景点北京晴天推荐故宫、颐和园、天安门广场 3. 机票上海→北京2026-03-18 1. 东航MU510108:00-10:30800元 2. 国航CA150214:00-16:40750元图AgentExecutor和LangGraph实现差异维度AgentExecutor 版本LangGraph 版本核心载体Agent决策AgentExecutor执行循环State全局状态Node步骤Graph编排逻辑组织靠 Prompt 让 Agent 决策 “调用哪个工具”逻辑耦合在 Prompt 中拆分为独立节点解析→查天气→推荐景点→查机票逻辑分散在节点函数更易维护分支控制靠 Agent 思考实现分支需在 Prompt 中写规则调试困难用add_conditional_edges显式定义分支规则可视化、易调试上下文管理依赖Memory仅能保存对话历史无法灵活读写多维度数据全局TravelState所有节点可读写城市、日期、天气等任意字段灵活度拉满AgentExecutor实现用户输入 → Agent 思考“先查天气” → Executor 调用天气工具 → Agent 思考“再推荐景点” → Executor 调用景点工具 → Agent 思考“再查机票” → Executor 调用机票工具 → Agent 整合结果 → Executor 返回答案。核心Agent 主导决策Executor 仅负责 “执行调用”逻辑黑盒化靠 Prompt 控制。LangGraph实现用户输入 → parse_input 节点解析信息→ get_weather 节点查天气→ recommend_spots 节点推荐景点→ 条件判断有出发城市→check_flight 节点→generate_answer 节点 | 无→直接 generate_answer 节点→ 结束。核心开发者主导定义节点和跳转规则逻辑白盒化代码显式控制。若增加功能二者差异扩展需求AgentExecutor 实现难度LangGraph 实现难度新增 “酒店查询” 步骤需修改 Prompt让 Agent 决策 “何时调用酒店工具”易影响原有逻辑新增check_hotel节点修改跳转规则即可无侵入调整分支逻辑如 “雨天不查机票”需修改 Prompt 规则调试需反复运行看 Agent 决策结果修改recommend_spots_node中的分支判断逻辑直接、可控多智能体协作如 “景点推荐交给专门智能体”极难需自定义 Agent 通信逻辑新增 “景点推荐智能体节点”直接替换原有recommend_spots_node即可适用场景总结AgentExecutor 适用场景简单线性任务≤3 个工具、快速落地、不想写复杂代码核心优势代码少、学习成本低靠 Prompt 快速实现 “思考 - 行动” 循环。LangGraph 适用场景复杂分支逻辑、多步骤任务、多智能体协作、需要显式控制执行流程核心优势逻辑解耦、可视化、易扩展、可维护性高适合生产级复杂应用。