拓冰网站定制 
如果你正在尝试构建一个能自主决策、调用工具、并完成复杂任务的AI智能体却发现现有的框架要么过于简单只能做单轮问答要么过于复杂需要自己管理状态、编排流程那么LangGraph很可能就是你一直在寻找的解决方案。过去一年AI智能体开发从概念走向落地但开发者面临的核心矛盾日益凸显如何平衡开发的灵活性与工程的规范性用LangChain写一个简单的Agent不难但一旦涉及多步骤推理、循环执行、状态持久化或团队协作代码很快就会变得难以维护。而LangGraph的出现正是为了解决这个“从Demo到生产”的工程化鸿沟。它不是一个全新的工具而是对现有LangChain生态的一次“架构升级”将智能体的工作流从“链式调用”提升为“图状态机”。本文不会重复那些官网已有的基础概念。我们将直击要害通过一个完整的实战项目带你彻底掌握LangGraph的核心如何用“图”的思维来设计和实现一个具备长期记忆、能调用外部API、并可稳定运行的智能体系统。你将学到的不只是几个API的调用而是智能体应用从设计、开发到调试的完整工程实践。文末附带了可直接运行的代码仓库建议收藏备用。1. 为什么是LangGraph重新理解智能体开发范式在深入代码之前我们必须先厘清一个关键问题LangGraph解决了什么根本问题这决定了我们是否值得投入学习。传统的LangChain Agent开发模式可以概括为“提示词驱动循环”。你定义一个工具集Tools和一个大模型LLM然后通过一个AgentExecutor来循环执行模型思考→选择工具→执行工具→观察结果→继续思考。这个模式对于简单任务很有效但其状态管理是隐式的、线性的且难以定制。当你的智能体需要处理以下场景时传统模式的短板就会暴露复杂流程控制需要根据中间结果决定走分支A还是分支B。长期记忆与状态持久化让智能体记住多轮对话的上下文或任务执行的中间状态。多智能体协作需要多个“角色”智能体相互通信、协同完成任务。人类介入审批在关键步骤需要暂停等待人工确认后再继续。异步与超时控制某些工具调用耗时很长需要异步处理和超时机制。LangGraph的答案是将智能体工作流建模为一个有状态图Stateful Graph。图中的节点Node可以是调用LLM、执行工具、或者任何自定义函数边Edge定义了节点之间的流转条件。整个系统的状态State是一个可自定义的字典随着在图中的流转而不断更新。这种范式转变带来了巨大优势可视化与可调试性工作流可以被直观地绘制出来执行路径一目了然。极强的灵活性你可以实现循环、条件分支、并行、子图等任何复杂流程。明确的状态管理所有中间数据都存储在状态对象中易于持久化和恢复。生产就绪内置了对并发、检查点、流式输出的支持。接下来我们将通过构建一个“旅行规划智能体”来具体感受这些优势。这个智能体能根据用户模糊的需求如“我想去一个温暖的海边城市度三天假”自动调用搜索工具获取信息进行多轮规划和确认最终生成一份结构化的旅行计划。2. 核心概念拆解图、状态、节点与边开始编码前需要准确理解LangGraph的四个核心抽象。理解它们就理解了LangGraph的全部。2.1 状态State状态是一个字典TypedDict它是工作流运行时信息的唯一载体。你可以把它想象成智能体的“工作内存”。通常它会包含messages: 对话消息列表这是与LLM交互的核心。next: 指示下一个应该执行哪个节点可选。任何你自定义的键如destination目的地、budget预算、plan计划等。在LangGraph中你需要首先定义一个状态类来明确数据的结构。2.2 节点Node节点是工作流中的基本执行单元。每个节点是一个函数它接收当前状态作为输入并返回一个对该状态的更新字典。节点可以调用LLM生成文本。执行一个工具如搜索、计算、查询数据库。运行一段业务逻辑代码。调用另一个子图。2.3 边Edge边定义了节点之间的流转逻辑。分为两种起始边Start Edge定义工作流从哪个节点开始。普通边Regular Edge定义从一个节点执行完后下一个该去哪个节点。边可以是有条件的Conditional Edge根据状态中的某个值来决定下一步走向从而实现分支逻辑。2.4 图Graph图是节点和边的集合。你创建图添加节点然后通过边将它们连接起来。最后将图编译成一个可执行的CompiledGraph对象它就可以像函数一样被调用。一个生动的类比你可以把LangGraph构建的智能体看作一个游戏剧本。状态是游戏角色的所有属性血量、位置、道具。节点是一个个剧情点或任务点如“与NPC对话”、“打开宝箱”。边是剧情的选择分支如果“拥有钥匙”则去“打开宝箱”否则去“寻找钥匙”。图就是整个游戏的剧本大纲。编译器CompiledGraph则是游戏引擎它根据玩家用户输入的互动按照剧本来推动游戏进程。3. 环境准备与项目初始化我们使用Python进行开发。确保你的环境满足以下要求Python版本: 3.10 或更高版本推荐3.11。包管理工具: pip 或 poetry。LLM服务: 我们将使用OpenAI的GPT模型你需要一个有效的API Key。你也可以替换为其他兼容OpenAI API的模型如Azure OpenAI, Ollama等。3.1 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir langgraph-travel-agent cd langgraph-travel-agent # 创建并激活虚拟环境以venv为例 python -m venv .venv # Windows .venv\Scripts\activate # Linux/Mac source .venv/bin/activate # 安装核心依赖 pip install langgraph langchain-openai tavily-python python-dotenv依赖说明langgraph: 核心框架。langchain-openai: LangChain对OpenAI的官方集成。tavily-python: 我们将使用Tavily作为搜索工具API。它是一个专为AI优化的搜索引擎。python-dotenv: 用于管理环境变量。3.2 配置API密钥在项目根目录创建.env文件用于安全存储密钥。# .env 文件内容 OPENAI_API_KEY你的OpenAI_API密钥 TAVILY_API_KEY你的Tavily_API密钥接下来创建一个config.py文件来加载配置。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) # 简单的配置检查 if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) if not TAVILY_API_KEY: print(警告未设置 TAVILY_API_KEY搜索功能将不可用。你可以去 https://tavily.com 申请免费额度。)3.3 初始化LLM和工具创建agents/core.py文件初始化我们将要用到的LLM客户端和搜索工具。# agents/core.py from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from config import OPENAI_API_KEY, TAVILY_API_KEY def get_llm(model_name: str gpt-4o-mini, temperature: float 0.1): 获取ChatOpenAI LLM实例。 使用gpt-4o-mini平衡性能与成本temperature调低使输出更稳定。 return ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyOPENAI_API_KEY) def get_search_tool(max_results: int 3): 获取Tavily搜索工具实例。 Tavily返回的结果已经是结构化摘要非常适合AI智能体使用。 if not TAVILY_API_KEY: # 返回一个模拟工具避免因密钥缺失导致程序崩溃 from langchain_core.tools import Tool def mock_search(query: str): return [{content: f模拟搜索结果关于{query}的信息。请配置TAVILY_API_KEY以获取真实数据。}] return Tool( namemock_web_search, description一个模拟的网络搜索工具。请配置TAVILY_API_KEY启用真实搜索。, funcmock_search ) return TavilySearchResults(max_resultsmax_results, tavily_api_keyTAVILY_API_KEY) # 预初始化常用实例懒加载模式 _llm None _search_tool None def get_cached_llm(): global _llm if _llm is None: _llm get_llm() return _llm def get_cached_search_tool(): global _search_tool if _search_tool is None: _search_tool get_search_tool() return _search_tool至此基础环境搭建完成。我们有了LLM、有了搜索工具接下来开始构建LangGraph的核心——状态图。4. 定义智能体状态与节点函数我们的旅行规划智能体需要经历几个阶段理解需求、搜索信息、生成计划、确认计划。我们首先定义状态然后为每个阶段创建节点。4.1 定义状态结构在agents/state.py中我们使用TypedDict来定义状态。这是确保类型安全的关键。# agents/state.py from typing import TypedDict, List, Optional, Annotated import operator from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class TravelAgentState(TypedDict): 旅行规划智能体的状态定义。 使用Annotated和add_messages是为了让LangGraph能自动处理消息列表的合并。 # 对话消息历史 messages: Annotated[List[BaseMessage], add_messages] # 用户原始输入 user_request: str # 解析出的目的地 destination: Optional[str] # 解析出的旅行天数 days: Optional[int] # 解析出的预算范围字符串如“中等” budget: Optional[str] # 从网络搜索到的目的地信息 destination_info: Optional[str] # 生成的详细旅行计划 detailed_plan: Optional[str] # 一个标志位表示是否需要用户确认 needs_human_approval: boolAnnotated[List[BaseMessage], add_messages]是LangGraph的一个精妙设计。它告诉框架当多个节点都返回对messages字段的更新时不要覆盖而是使用add_messages函数将它们追加到列表中。这完美契合了对话历史的累积特性。4.2 创建节点函数节点就是普通的Python函数它接收状态返回一个包含更新字段的字典。我们在agents/nodes.py中创建它们。# agents/nodes.py from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from .state import TravelAgentState from .core import get_cached_llm, get_cached_search_tool import json def parse_user_request(state: TravelAgentState) - TravelAgentState: 节点1解析用户模糊请求。 从对话历史中提取最后一条用户消息解析出目的地、天数、预算等结构化信息。 llm get_cached_llm() messages state[messages] # 获取最新的用户消息 last_user_msg None for msg in reversed(messages): if isinstance(msg, HumanMessage): last_user_msg msg.content break if not last_user_msg: last_user_msg state.get(user_request, ) # 构建一个提示词让LLM进行信息提取 parser_prompt f 你是一个旅行请求解析器。请从用户的以下请求中提取关键信息。 用户请求{last_user_msg} 请以JSON格式返回包含以下字段 - destination: 旅行目的地城市或地区。如果用户描述模糊如“海边城市”请推断一个最可能的具体目的地。 - days: 旅行天数整数。如果未明确提及请推断一个合理的值如3-5天。 - budget: 预算水平取值为“经济”、“中等”、“豪华”之一。根据描述推断。 只返回JSON不要有其他任何解释。 system_msg SystemMessage(content你是一个精准的信息提取助手只返回JSON。) human_msg HumanMessage(contentparser_prompt) response llm.invoke([system_msg, human_msg]) try: parsed json.loads(response.content) except json.JSONDecodeError: # 如果解析失败使用默认值 parsed {destination: None, days: 3, budget: 中等} # 更新状态 return { destination: parsed.get(destination), days: parsed.get(days, 3), budget: parsed.get(budget, 中等), user_request: last_user_msg } def search_destination_info(state: TravelAgentState) - TravelAgentState: 节点2搜索目的地信息。 使用搜索工具获取目的地的景点、美食、文化等实用信息。 destination state.get(destination) if not destination: # 如果没有目的地则跳过搜索直接返回空信息 return {destination_info: 未指定目的地。} search_tool get_cached_search_tool() # 构建搜索查询词 days state.get(days, 3) budget state.get(budget, 中等) query f{destination} {days}天旅行攻略 {budget}预算 必去景点 美食推荐 try: search_results search_tool.invoke(query) # Tavily工具返回的是包含‘content’的字典列表 info \n\n.join([res.get(content, ) for res in search_results]) except Exception as e: info f搜索目的地{destination}时出错{e} return {destination_info: info} def generate_travel_plan(state: TravelAgentState) - TravelAgentState: 节点3生成详细旅行计划。 综合用户需求、解析出的信息和搜索到的资料生成一份结构化的旅行计划。 llm get_cached_llm() user_request state[user_request] destination state[destination] days state[days] budget state[budget] destination_info state.get(destination_info, ) planner_prompt f 你是一个专业的旅行规划师。请为以下需求制定一份详细的旅行计划。 **用户原始需求**{user_request} **目的地**{destination} **天数**{days}天 **预算水平**{budget} **参考信息** {destination_info} **请生成一份包含以下部分的计划** 1. 行程概览总体思路 2. 每日详细安排上午、下午、晚上分别做什么包括景点、活动、餐饮建议 3. 预算分配估算交通、住宿、餐饮、门票、购物 4. 行前准备与注意事项签证、天气、衣物、必备物品等 5. 备用方案或贴士 请确保计划符合{days}天的时间安排和{budget}预算水平。计划应具体、可行、有吸引力。 system_msg SystemMessage(content你是一个经验丰富、考虑周到的旅行规划师。) human_msg HumanMessage(contentplanner_prompt) response llm.invoke([system_msg, human_msg]) plan response.content # 同时将生成的计划以AI消息的形式添加到对话历史中 new_message AIMessage(contentf我已为您生成了一份{destination}{days}日游的初步计划请查阅。) return { detailed_plan: plan, messages: [new_message], # 这个列表会被 add_messages 自动追加 needs_human_approval: True # 生成计划后需要用户确认 } def human_approval_node(state: TravelAgentState) - TravelAgentState: 节点4模拟人类确认环节。 在实际应用中这里可以连接到一个用户界面等待输入。 在此示例中我们模拟用户总是同意。 # 在实际系统中这里会暂停等待前端或用户输入“同意”或“拒绝”。 # 我们简化处理假设用户同意并添加一条模拟的人类确认消息。 approval_msg HumanMessage(content计划看起来不错我同意。) return { messages: [approval_msg], needs_human_approval: False } def finalize_plan(state: TravelAgentState) - TravelAgentState: 节点5最终确认并输出计划。 在获得人类确认后对计划进行最终润色并输出。 llm get_cached_llm() detailed_plan state[detailed_plan] finalizer_prompt f 以下是一份旅行计划的草稿已经获得用户同意。请对其进行最终润色使其语言更加优美、鼓舞人心并以“【最终旅行计划】”的标题开头。 计划草稿 {detailed_plan} response llm.invoke([SystemMessage(content你是文案编辑。), HumanMessage(contentfinalizer_prompt)]) final_plan response.content final_message AIMessage(contentf完美这是为您定制的最终版旅行计划\n\n{final_plan}\n\n祝您旅途愉快) return { messages: [final_message], detailed_plan: final_plan # 更新为最终版 }现在我们有了五个功能明确的节点解析请求、搜索信息、生成计划、等待确认、最终润色。下一步就是用边把它们连接起来形成一个完整的工作流。5. 构建并编译LangGraph状态图这是LangGraph最核心的部分。我们将节点组装成图并定义它们之间的流转逻辑。创建agents/graph.py文件。# agents/graph.py from langgraph.graph import StateGraph, END from .state import TravelAgentState from .nodes import ( parse_user_request, search_destination_info, generate_travel_plan, human_approval_node, finalize_plan ) def create_travel_agent_graph(): 创建并编译旅行规划智能体的状态图。 # 1. 初始化一个状态图指定状态类型 workflow StateGraph(TravelAgentState) # 2. 添加节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(search_info, search_destination_info) workflow.add_node(generate_plan, generate_travel_plan) workflow.add_node(wait_approval, human_approval_node) workflow.add_node(finalize, finalize_plan) # 3. 设置起始节点从解析请求开始 workflow.set_entry_point(parse_request) # 4. 添加边定义工作流 # 解析完请求后无论结果如何都去搜索信息 workflow.add_edge(parse_request, search_info) # 搜索完信息后去生成计划 workflow.add_edge(search_info, generate_plan) # 生成计划后去等待人工确认 workflow.add_edge(generate_plan, wait_approval) # 5. 定义条件边根据用户确认结果决定下一步 # 这里我们简化在wait_approval节点中我们模拟用户同意所以直接去finalize # 但在真实场景中这里需要一个条件判断。 # 演示条件边的用法 def should_finalize(state: TravelAgentState): 根据状态决定下一步。如果不需要确认或已确认则终稿否则等待。 # 在我们的流程中wait_approval节点会将needs_human_approval设为False # 所以这里总是返回”finalize“ return finalize # 添加从“wait_approval”出发的条件边 workflow.add_conditional_edges( wait_approval, should_finalize, { finalize: finalize, # 如果should_finalize返回”finalize“则前往finalize节点 } ) # 6. 最终节点指向END workflow.add_edge(finalize, END) # 7. 编译图 return workflow.compile() # 创建图的单例 travel_agent_graph create_travel_agent_graph()上面的图是一个简单的线性流程解析→搜索→生成→确认→终稿。但我们已经引入了add_conditional_edges来展示如何实现分支。你可以轻松地修改should_finalize函数例如检查状态中是否有“用户拒绝”的标记然后跳转到一个“修改计划”的节点从而实现循环。6. 运行智能体与效果验证图已经编译好现在让我们运行它看看这个旅行规划智能体如何工作。创建主程序main.py。# main.py from agents.graph import travel_agent_graph from langchain_core.messages import HumanMessage import asyncio async def main(): 主函数运行旅行规划智能体。 print( 旅行规划智能体启动 ) # 模拟用户输入 user_input 我想下个月去一个温暖、有美食的海边城市放松3天预算中等。 print(f用户请求: {user_input}) # 构建初始状态 initial_state { messages: [HumanMessage(contentuser_input)], user_request: user_input, destination: None, days: None, budget: None, destination_info: None, detailed_plan: None, needs_human_approval: False, } # 运行图 print(\n--- 智能体开始执行 ---) try: # 使用流式输出可以看到执行到哪个节点 async for event in travel_agent_graph.astream_events(initial_state, versionv1): kind event[event] node event.get(name, N/A) if kind on_chain_start and event[metadata].get(step) 1: # 这里可以打印节点开始信息 pass elif kind on_chain_end and event[metadata].get(step) 1: print(f[执行完成] 节点: {node}) except Exception as e: print(f执行过程中出现错误: {e}) # 即使流式事件出错我们也尝试获取最终结果 final_state await travel_agent_graph.ainvoke(initial_state) else: # 正常执行后获取最终状态 final_state await travel_agent_graph.ainvoke(initial_state) print(\n--- 执行完成 ---) # 输出关键结果 print(f\n✅ 解析出的目的地: {final_state.get(destination)}) print(f✅ 解析出的天数: {final_state.get(days)}) print(f✅ 解析出的预算: {final_state.get(budget)}) print(f\n 生成的旅行计划 (摘要):) plan final_state.get(detailed_plan, 无) # 只打印前500字符作为预览 print(plan[:500] ... if len(plan) 500 else plan) print(f\n 最终对话消息数: {len(final_state.get(messages, []))}) # 可以选择保存计划到文件 save_to_file input(\n是否将完整计划保存到文件(y/n): ).lower().strip() if save_to_file y: with open(ftravel_plan_{final_state.get(destination, unknown)}.txt, w, encodingutf-8) as f: f.write(plan) print(f计划已保存至 travel_plan_{final_state.get(destination, unknown)}.txt) if __name__ __main__: asyncio.run(main())6.1 运行程序在终端中执行python main.py6.2 预期输出与解读程序会依次执行各个节点并在控制台输出类似以下内容 旅行规划智能体启动 用户请求: 我想下个月去一个温暖、有美食的海边城市放松3天预算中等。 --- 智能体开始执行 --- [执行完成] 节点: parse_request [执行完成] 节点: search_info [执行完成] 节点: generate_plan [执行完成] 节点: wait_approval [执行完成] 节点: finalize --- 执行完成 --- ✅ 解析出的目的地: 厦门 ✅ 解析出的天数: 3 ✅ 解析出的预算: 中等 生成的旅行计划 (摘要): 【最终旅行计划】 **行程概览** 本次厦门3日休闲美食之旅以“慢生活、深体验”为主题聚焦鼓浪屿、环岛路、沙坡尾等核心区域品尝地道海鲜、沙茶面、土笋冻等闽南风味预算控制在中等水平... 最终对话消息数: 4执行流程解读parse_request节点LLM从“温暖、有美食的海边城市”推断出具体目的地“厦门”并提取出天数“3”和预算“中等”。search_info节点调用Tavily搜索API获取厦门3日游、中等预算的攻略、景点和美食信息。generate_plan节点综合所有信息生成一份详细的、结构化的旅行计划草案并将needs_human_approval设为True。wait_approval节点模拟用户确认环节。在我们的代码中它自动模拟用户同意并将needs_human_approval设为False。finalize节点对草案进行最终润色输出鼓舞人心的最终版计划。整个流程清晰可见状态目的地、天数、计划草案、确认标志在节点间有序传递和更新。7. 深入探索实现条件循环与错误处理上面的例子是线性流程。LangGraph的强大之处在于处理非线性逻辑。让我们增强智能体使其在用户“拒绝”计划时能够循环回去修改。7.1 修改状态与节点首先在状态中增加一个字段来记录用户反馈。# agents/state.py (修改部分) class TravelAgentState(TypedDict): # ... 保留原有字段 ... # 新增用户对上一版计划的反馈 user_feedback: Optional[str] # 新增计划修改次数 revision_count: int然后修改human_approval_node和generate_travel_plan节点并新增一个revise_plan_node。# agents/nodes.py (新增和修改函数) def human_approval_node_with_feedback(state: TravelAgentState) - TravelAgentState: 修改后的确认节点模拟获取用户反馈。 在实际应用中这里从UI获取“同意”、“拒绝并给出修改意见”。 # 模拟交互我们假设第一版计划被拒绝用户给出了反馈 if state.get(revision_count, 0) 0: # 第一次生成计划模拟用户拒绝并要求修改 feedback 我不太想去鼓浪屿人太多了。能不能推荐一些更小众、安静的地方 approval_msg HumanMessage(contentf我拒绝这个计划。反馈{feedback}) needs_approval True else: # 第二次修改后模拟用户同意 feedback 这次修改得很好我同意了 approval_msg HumanMessage(contentfeedback) needs_approval False return { messages: [approval_msg], user_feedback: feedback, needs_human_approval: needs_approval } def revise_plan_node(state: TravelAgentState) - TravelAgentState: 新节点根据用户反馈修改旅行计划。 llm get_cached_llm() old_plan state[detailed_plan] feedback state[user_feedback] revision_count state.get(revision_count, 0) 1 revision_prompt f 这是一份旅行计划收到了用户的修改意见。请根据意见重新规划。 **原计划** {old_plan} **用户反馈** {feedback} **修改要求** 1. 充分考虑用户反馈。 2. 保留原计划中用户未提及的合理部分。 3. 生成一份新的、完整的旅行计划。 这是第{revision_count}次修改。 response llm.invoke([SystemMessage(content你是耐心细致的旅行规划师善于根据反馈调整方案。), HumanMessage(contentrevision_prompt)]) revised_plan response.content new_message AIMessage(contentf已根据您的反馈完成了第{revision_count}次修改请查阅新计划。) return { detailed_plan: revised_plan, messages: [new_message], revision_count: revision_count, needs_human_approval: True # 修改后再次等待确认 }7.2 修改图定义以实现循环现在修改agents/graph.py引入条件循环。# agents/graph.py (修改后的create_travel_agent_graph函数) def create_travel_agent_graph_with_loop(): workflow StateGraph(TravelAgentState) # 添加所有节点包括新的revise_plan节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(search_info, search_destination_info) workflow.add_node(generate_plan, generate_travel_plan) workflow.add_node(wait_approval, human_approval_node_with_feedback) # 使用新的确认节点 workflow.add_node(revise_plan, revise_plan_node) # 新增修改节点 workflow.add_node(finalize, finalize_plan) workflow.set_entry_point(parse_request) workflow.add_edge(parse_request, search_info) workflow.add_edge(search_info, generate_plan) workflow.add_edge(generate_plan, wait_approval) # 关键定义从 wait_approval 出发的条件边 def route_after_approval(state: TravelAgentState): 根据用户确认状态决定下一步 if state.get(needs_human_approval): # 如果需要确认即用户拒绝则去修改计划 return revise_plan else: # 如果不需要确认即用户同意则去最终定稿 return finalize workflow.add_conditional_edges( wait_approval, route_after_approval, { revise_plan: revise_plan, finalize: finalize } ) # 从修改计划节点再次回到等待确认节点形成循环 workflow.add_edge(revise_plan, wait_approval) # 为防止无限循环可以设置一个最大修改次数检查在revise_plan节点或条件边中逻辑判断 # 这里我们在条件边函数中添加简单判断 def route_after_approval_with_limit(state: TravelAgentState): revision_count state.get(revision_count, 0) if revision_count 3: # 最多修改3次 print(已达到最大修改次数强制通过。) return finalize return route_after_approval(state) # 更新条件边函数 # workflow.add_conditional_edges(wait_approval, route_after_approval_with_limit, ...) workflow.add_edge(finalize, END) return workflow.compile()这个新的图实现了一个完整的“生成-确认-修改”循环。只有当用户同意needs_human_approval为False时才会跳出循环进入finalize节点。我们还添加了revision_count和简单的限制逻辑来防止无限循环。8. 常见问题与排查思路在实际使用LangGraph时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案ValueError: Missing keys: ...状态StateTypedDict中定义的某些键在节点返回的更新字典中没有提供初始值。检查状态类定义和每个节点函数的返回值。确保所有在TypedDict中声明为非Optional的键在初始状态和每次状态更新中都存在。1. 将状态字段改为Optional[...]。2. 确保所有节点都返回完整的字段更新或使用...操作符保留其他字段。图编译成功但执行时无任何输出或卡住1. 边Edge没有正确连接导致图没有终点END。2. 使用了异步astream但未在异步环境中运行。1. 使用graph.get_graph().draw_mermaid()输出图结构可视化检查路径是否通达。2. 确认在异步函数中调用astream或用asyncio.run()包装。1. 确保所有路径最终都能到达END节点。2. 使用同步的invoke方法测试或确保在async def函数中运行。工具调用失败如搜索API报错1. API密钥未设置或错误。2. 网络问题。3. 工具参数格式错误。1. 检查.env文件和环境变量。2. 在节点函数中添加try...except打印具体错误。3. 单独测试工具调用。1. 正确配置API密钥。2. 实现降级策略如返回模拟数据。3. 查阅对应工具的官方文档。LLM输出格式不符合预期导致后续节点解析失败提示词Prompt不够精确LLM返回了非结构化文本。打印出LLM的原始响应内容。1. 在提示词中明确要求输出格式如“请以JSON格式返回”。2. 使用LangChain的OutputParser如PydanticOutputParser来强制结构化输出。3. 在节点代码中添加更健壮的异常处理如json.loads失败后的默认值。条件边Conditional Edge逻辑不生效条件函数返回的值与add_conditional_edges中定义的映射键不匹配。打印条件函数的输入state和返回值。确保条件函数返回的字符串与add_conditional_edges的映射字典中的某个键完全一致大小写敏感。状态更新不符合预期如消息被覆盖对messages这类列表字段没有使用Annotated和add_messages。检查状态类中列表字段的定义。对于需要追加而非替换的字段使用Annotated[List[BaseMessage], add_messages]语法。对于其他自定义列表可以定义自己的归约函数。9. 最佳实践与工程化建议将LangGraph智能体用于生产环境需要遵循以下工程最佳实践9.1 状态设计原则最小化与清晰化只把真正需要在节点间传递的数据放入状态。避免状态字典过于庞大。使用TypedDict强烈推荐使用TypedDict并配合mypy或pyright进行类型检查能在开发阶段避免大量键错误。区分对话历史与内部状态像messages这样的对话历史使用add_messages自动管理。业务状态如destination,plan单独存储。9.2 节点函数设计单一职责每个节点只做一件事如解析、搜索、生成、验证。纯函数化节点函数应尽可能只依赖于输入状态和注入的依赖如LLM、工具避免副作用。这便于测试和调试。完善的错误处理节点内部应对可能失败的操作网络调用、JSON解析进行try-catch并返回错误信息或默认状态避免整个图执行中断。9.3 图的构建与调试可视化是利器使用graph.get_graph().draw_mermaid()生成Mermaid图表直观理解工作流。这对于复杂流程图和向团队解释设计至关重要。版本化配置将图的构建逻辑封装在函数中便于根据配置如不同环境、不同功能开关创建不同的图实例。利用检查点Checkpoint对于长时运行或需要中断恢复的智能体使用LangGraph的检查点功能持久化状态。这对于需要“人工介入”然后继续的任务非常有用。9.4 测试与监控单元测试节点单独测试每个节点函数模拟输入状态验证输出状态。集成测试全图使用固定的初始状态和Mock的LLM、工具来测试整个图的执行路径和最终输出。记录与追踪利用astream_events或LangSmith等工具详细记录每个节点的输入、输出和执行时间便于问题排查和性能分析。9.5 与MCPModel Context Protocol集成MCP协议正在成为连接AI智能体与外部工具、数据源的重要标准。LangGraph可以与MCP Server无缝集成将MCP Server作为工具你可以将任何MCP Server如连接数据库、内部API的Server封装成LangChain Tool然后在LangGraph的节点中调用。构建更强大的智能体通过MCP你的智能体可以安全、标准化地访问代码库如Git、项目管理工具如Jira、设计文件如Figma实现真正的“AI员工”。实践步骤部署或连接一个MCP Server例如用于查询数据库的Server。使用langchain-mcp-adapters等库将MCP Server提供的工具动态加载为LangChain Tool。将这些Tool添加到你的LangGraph智能体的工具集中。通过遵循这些实践你可以构建出结构清晰、易于维护、可扩展且稳健的AI智能体应用。LangGraph提供的不仅是一个框架更是一种构建复杂、可维护AI工作流的工程范式。从简单的线性流程开始逐步引入条件分支、循环、子图和持久化你将能驾驭越来越复杂的智能体应用场景。 LangGraph实战：构建具备长期记忆与复杂流程控制的AI智能体
如果你正在尝试构建一个能自主决策、调用工具、并完成复杂任务的AI智能体却发现现有的框架要么过于简单只能做单轮问答要么过于复杂需要自己管理状态、编排流程那么LangGraph很可能就是你一直在寻找的解决方案。过去一年AI智能体开发从概念走向落地但开发者面临的核心矛盾日益凸显如何平衡开发的灵活性与工程的规范性用LangChain写一个简单的Agent不难但一旦涉及多步骤推理、循环执行、状态持久化或团队协作代码很快就会变得难以维护。而LangGraph的出现正是为了解决这个“从Demo到生产”的工程化鸿沟。它不是一个全新的工具而是对现有LangChain生态的一次“架构升级”将智能体的工作流从“链式调用”提升为“图状态机”。本文不会重复那些官网已有的基础概念。我们将直击要害通过一个完整的实战项目带你彻底掌握LangGraph的核心如何用“图”的思维来设计和实现一个具备长期记忆、能调用外部API、并可稳定运行的智能体系统。你将学到的不只是几个API的调用而是智能体应用从设计、开发到调试的完整工程实践。文末附带了可直接运行的代码仓库建议收藏备用。1. 为什么是LangGraph重新理解智能体开发范式在深入代码之前我们必须先厘清一个关键问题LangGraph解决了什么根本问题这决定了我们是否值得投入学习。传统的LangChain Agent开发模式可以概括为“提示词驱动循环”。你定义一个工具集Tools和一个大模型LLM然后通过一个AgentExecutor来循环执行模型思考→选择工具→执行工具→观察结果→继续思考。这个模式对于简单任务很有效但其状态管理是隐式的、线性的且难以定制。当你的智能体需要处理以下场景时传统模式的短板就会暴露复杂流程控制需要根据中间结果决定走分支A还是分支B。长期记忆与状态持久化让智能体记住多轮对话的上下文或任务执行的中间状态。多智能体协作需要多个“角色”智能体相互通信、协同完成任务。人类介入审批在关键步骤需要暂停等待人工确认后再继续。异步与超时控制某些工具调用耗时很长需要异步处理和超时机制。LangGraph的答案是将智能体工作流建模为一个有状态图Stateful Graph。图中的节点Node可以是调用LLM、执行工具、或者任何自定义函数边Edge定义了节点之间的流转条件。整个系统的状态State是一个可自定义的字典随着在图中的流转而不断更新。这种范式转变带来了巨大优势可视化与可调试性工作流可以被直观地绘制出来执行路径一目了然。极强的灵活性你可以实现循环、条件分支、并行、子图等任何复杂流程。明确的状态管理所有中间数据都存储在状态对象中易于持久化和恢复。生产就绪内置了对并发、检查点、流式输出的支持。接下来我们将通过构建一个“旅行规划智能体”来具体感受这些优势。这个智能体能根据用户模糊的需求如“我想去一个温暖的海边城市度三天假”自动调用搜索工具获取信息进行多轮规划和确认最终生成一份结构化的旅行计划。2. 核心概念拆解图、状态、节点与边开始编码前需要准确理解LangGraph的四个核心抽象。理解它们就理解了LangGraph的全部。2.1 状态State状态是一个字典TypedDict它是工作流运行时信息的唯一载体。你可以把它想象成智能体的“工作内存”。通常它会包含messages: 对话消息列表这是与LLM交互的核心。next: 指示下一个应该执行哪个节点可选。任何你自定义的键如destination目的地、budget预算、plan计划等。在LangGraph中你需要首先定义一个状态类来明确数据的结构。2.2 节点Node节点是工作流中的基本执行单元。每个节点是一个函数它接收当前状态作为输入并返回一个对该状态的更新字典。节点可以调用LLM生成文本。执行一个工具如搜索、计算、查询数据库。运行一段业务逻辑代码。调用另一个子图。2.3 边Edge边定义了节点之间的流转逻辑。分为两种起始边Start Edge定义工作流从哪个节点开始。普通边Regular Edge定义从一个节点执行完后下一个该去哪个节点。边可以是有条件的Conditional Edge根据状态中的某个值来决定下一步走向从而实现分支逻辑。2.4 图Graph图是节点和边的集合。你创建图添加节点然后通过边将它们连接起来。最后将图编译成一个可执行的CompiledGraph对象它就可以像函数一样被调用。一个生动的类比你可以把LangGraph构建的智能体看作一个游戏剧本。状态是游戏角色的所有属性血量、位置、道具。节点是一个个剧情点或任务点如“与NPC对话”、“打开宝箱”。边是剧情的选择分支如果“拥有钥匙”则去“打开宝箱”否则去“寻找钥匙”。图就是整个游戏的剧本大纲。编译器CompiledGraph则是游戏引擎它根据玩家用户输入的互动按照剧本来推动游戏进程。3. 环境准备与项目初始化我们使用Python进行开发。确保你的环境满足以下要求Python版本: 3.10 或更高版本推荐3.11。包管理工具: pip 或 poetry。LLM服务: 我们将使用OpenAI的GPT模型你需要一个有效的API Key。你也可以替换为其他兼容OpenAI API的模型如Azure OpenAI, Ollama等。3.1 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir langgraph-travel-agent cd langgraph-travel-agent # 创建并激活虚拟环境以venv为例 python -m venv .venv # Windows .venv\Scripts\activate # Linux/Mac source .venv/bin/activate # 安装核心依赖 pip install langgraph langchain-openai tavily-python python-dotenv依赖说明langgraph: 核心框架。langchain-openai: LangChain对OpenAI的官方集成。tavily-python: 我们将使用Tavily作为搜索工具API。它是一个专为AI优化的搜索引擎。python-dotenv: 用于管理环境变量。3.2 配置API密钥在项目根目录创建.env文件用于安全存储密钥。# .env 文件内容 OPENAI_API_KEY你的OpenAI_API密钥 TAVILY_API_KEY你的Tavily_API密钥接下来创建一个config.py文件来加载配置。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) # 简单的配置检查 if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) if not TAVILY_API_KEY: print(警告未设置 TAVILY_API_KEY搜索功能将不可用。你可以去 https://tavily.com 申请免费额度。)3.3 初始化LLM和工具创建agents/core.py文件初始化我们将要用到的LLM客户端和搜索工具。# agents/core.py from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from config import OPENAI_API_KEY, TAVILY_API_KEY def get_llm(model_name: str gpt-4o-mini, temperature: float 0.1): 获取ChatOpenAI LLM实例。 使用gpt-4o-mini平衡性能与成本temperature调低使输出更稳定。 return ChatOpenAI(modelmodel_name, temperaturetemperature, api_keyOPENAI_API_KEY) def get_search_tool(max_results: int 3): 获取Tavily搜索工具实例。 Tavily返回的结果已经是结构化摘要非常适合AI智能体使用。 if not TAVILY_API_KEY: # 返回一个模拟工具避免因密钥缺失导致程序崩溃 from langchain_core.tools import Tool def mock_search(query: str): return [{content: f模拟搜索结果关于{query}的信息。请配置TAVILY_API_KEY以获取真实数据。}] return Tool( namemock_web_search, description一个模拟的网络搜索工具。请配置TAVILY_API_KEY启用真实搜索。, funcmock_search ) return TavilySearchResults(max_resultsmax_results, tavily_api_keyTAVILY_API_KEY) # 预初始化常用实例懒加载模式 _llm None _search_tool None def get_cached_llm(): global _llm if _llm is None: _llm get_llm() return _llm def get_cached_search_tool(): global _search_tool if _search_tool is None: _search_tool get_search_tool() return _search_tool至此基础环境搭建完成。我们有了LLM、有了搜索工具接下来开始构建LangGraph的核心——状态图。4. 定义智能体状态与节点函数我们的旅行规划智能体需要经历几个阶段理解需求、搜索信息、生成计划、确认计划。我们首先定义状态然后为每个阶段创建节点。4.1 定义状态结构在agents/state.py中我们使用TypedDict来定义状态。这是确保类型安全的关键。# agents/state.py from typing import TypedDict, List, Optional, Annotated import operator from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class TravelAgentState(TypedDict): 旅行规划智能体的状态定义。 使用Annotated和add_messages是为了让LangGraph能自动处理消息列表的合并。 # 对话消息历史 messages: Annotated[List[BaseMessage], add_messages] # 用户原始输入 user_request: str # 解析出的目的地 destination: Optional[str] # 解析出的旅行天数 days: Optional[int] # 解析出的预算范围字符串如“中等” budget: Optional[str] # 从网络搜索到的目的地信息 destination_info: Optional[str] # 生成的详细旅行计划 detailed_plan: Optional[str] # 一个标志位表示是否需要用户确认 needs_human_approval: boolAnnotated[List[BaseMessage], add_messages]是LangGraph的一个精妙设计。它告诉框架当多个节点都返回对messages字段的更新时不要覆盖而是使用add_messages函数将它们追加到列表中。这完美契合了对话历史的累积特性。4.2 创建节点函数节点就是普通的Python函数它接收状态返回一个包含更新字段的字典。我们在agents/nodes.py中创建它们。# agents/nodes.py from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from .state import TravelAgentState from .core import get_cached_llm, get_cached_search_tool import json def parse_user_request(state: TravelAgentState) - TravelAgentState: 节点1解析用户模糊请求。 从对话历史中提取最后一条用户消息解析出目的地、天数、预算等结构化信息。 llm get_cached_llm() messages state[messages] # 获取最新的用户消息 last_user_msg None for msg in reversed(messages): if isinstance(msg, HumanMessage): last_user_msg msg.content break if not last_user_msg: last_user_msg state.get(user_request, ) # 构建一个提示词让LLM进行信息提取 parser_prompt f 你是一个旅行请求解析器。请从用户的以下请求中提取关键信息。 用户请求{last_user_msg} 请以JSON格式返回包含以下字段 - destination: 旅行目的地城市或地区。如果用户描述模糊如“海边城市”请推断一个最可能的具体目的地。 - days: 旅行天数整数。如果未明确提及请推断一个合理的值如3-5天。 - budget: 预算水平取值为“经济”、“中等”、“豪华”之一。根据描述推断。 只返回JSON不要有其他任何解释。 system_msg SystemMessage(content你是一个精准的信息提取助手只返回JSON。) human_msg HumanMessage(contentparser_prompt) response llm.invoke([system_msg, human_msg]) try: parsed json.loads(response.content) except json.JSONDecodeError: # 如果解析失败使用默认值 parsed {destination: None, days: 3, budget: 中等} # 更新状态 return { destination: parsed.get(destination), days: parsed.get(days, 3), budget: parsed.get(budget, 中等), user_request: last_user_msg } def search_destination_info(state: TravelAgentState) - TravelAgentState: 节点2搜索目的地信息。 使用搜索工具获取目的地的景点、美食、文化等实用信息。 destination state.get(destination) if not destination: # 如果没有目的地则跳过搜索直接返回空信息 return {destination_info: 未指定目的地。} search_tool get_cached_search_tool() # 构建搜索查询词 days state.get(days, 3) budget state.get(budget, 中等) query f{destination} {days}天旅行攻略 {budget}预算 必去景点 美食推荐 try: search_results search_tool.invoke(query) # Tavily工具返回的是包含‘content’的字典列表 info \n\n.join([res.get(content, ) for res in search_results]) except Exception as e: info f搜索目的地{destination}时出错{e} return {destination_info: info} def generate_travel_plan(state: TravelAgentState) - TravelAgentState: 节点3生成详细旅行计划。 综合用户需求、解析出的信息和搜索到的资料生成一份结构化的旅行计划。 llm get_cached_llm() user_request state[user_request] destination state[destination] days state[days] budget state[budget] destination_info state.get(destination_info, ) planner_prompt f 你是一个专业的旅行规划师。请为以下需求制定一份详细的旅行计划。 **用户原始需求**{user_request} **目的地**{destination} **天数**{days}天 **预算水平**{budget} **参考信息** {destination_info} **请生成一份包含以下部分的计划** 1. 行程概览总体思路 2. 每日详细安排上午、下午、晚上分别做什么包括景点、活动、餐饮建议 3. 预算分配估算交通、住宿、餐饮、门票、购物 4. 行前准备与注意事项签证、天气、衣物、必备物品等 5. 备用方案或贴士 请确保计划符合{days}天的时间安排和{budget}预算水平。计划应具体、可行、有吸引力。 system_msg SystemMessage(content你是一个经验丰富、考虑周到的旅行规划师。) human_msg HumanMessage(contentplanner_prompt) response llm.invoke([system_msg, human_msg]) plan response.content # 同时将生成的计划以AI消息的形式添加到对话历史中 new_message AIMessage(contentf我已为您生成了一份{destination}{days}日游的初步计划请查阅。) return { detailed_plan: plan, messages: [new_message], # 这个列表会被 add_messages 自动追加 needs_human_approval: True # 生成计划后需要用户确认 } def human_approval_node(state: TravelAgentState) - TravelAgentState: 节点4模拟人类确认环节。 在实际应用中这里可以连接到一个用户界面等待输入。 在此示例中我们模拟用户总是同意。 # 在实际系统中这里会暂停等待前端或用户输入“同意”或“拒绝”。 # 我们简化处理假设用户同意并添加一条模拟的人类确认消息。 approval_msg HumanMessage(content计划看起来不错我同意。) return { messages: [approval_msg], needs_human_approval: False } def finalize_plan(state: TravelAgentState) - TravelAgentState: 节点5最终确认并输出计划。 在获得人类确认后对计划进行最终润色并输出。 llm get_cached_llm() detailed_plan state[detailed_plan] finalizer_prompt f 以下是一份旅行计划的草稿已经获得用户同意。请对其进行最终润色使其语言更加优美、鼓舞人心并以“【最终旅行计划】”的标题开头。 计划草稿 {detailed_plan} response llm.invoke([SystemMessage(content你是文案编辑。), HumanMessage(contentfinalizer_prompt)]) final_plan response.content final_message AIMessage(contentf完美这是为您定制的最终版旅行计划\n\n{final_plan}\n\n祝您旅途愉快) return { messages: [final_message], detailed_plan: final_plan # 更新为最终版 }现在我们有了五个功能明确的节点解析请求、搜索信息、生成计划、等待确认、最终润色。下一步就是用边把它们连接起来形成一个完整的工作流。5. 构建并编译LangGraph状态图这是LangGraph最核心的部分。我们将节点组装成图并定义它们之间的流转逻辑。创建agents/graph.py文件。# agents/graph.py from langgraph.graph import StateGraph, END from .state import TravelAgentState from .nodes import ( parse_user_request, search_destination_info, generate_travel_plan, human_approval_node, finalize_plan ) def create_travel_agent_graph(): 创建并编译旅行规划智能体的状态图。 # 1. 初始化一个状态图指定状态类型 workflow StateGraph(TravelAgentState) # 2. 添加节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(search_info, search_destination_info) workflow.add_node(generate_plan, generate_travel_plan) workflow.add_node(wait_approval, human_approval_node) workflow.add_node(finalize, finalize_plan) # 3. 设置起始节点从解析请求开始 workflow.set_entry_point(parse_request) # 4. 添加边定义工作流 # 解析完请求后无论结果如何都去搜索信息 workflow.add_edge(parse_request, search_info) # 搜索完信息后去生成计划 workflow.add_edge(search_info, generate_plan) # 生成计划后去等待人工确认 workflow.add_edge(generate_plan, wait_approval) # 5. 定义条件边根据用户确认结果决定下一步 # 这里我们简化在wait_approval节点中我们模拟用户同意所以直接去finalize # 但在真实场景中这里需要一个条件判断。 # 演示条件边的用法 def should_finalize(state: TravelAgentState): 根据状态决定下一步。如果不需要确认或已确认则终稿否则等待。 # 在我们的流程中wait_approval节点会将needs_human_approval设为False # 所以这里总是返回”finalize“ return finalize # 添加从“wait_approval”出发的条件边 workflow.add_conditional_edges( wait_approval, should_finalize, { finalize: finalize, # 如果should_finalize返回”finalize“则前往finalize节点 } ) # 6. 最终节点指向END workflow.add_edge(finalize, END) # 7. 编译图 return workflow.compile() # 创建图的单例 travel_agent_graph create_travel_agent_graph()上面的图是一个简单的线性流程解析→搜索→生成→确认→终稿。但我们已经引入了add_conditional_edges来展示如何实现分支。你可以轻松地修改should_finalize函数例如检查状态中是否有“用户拒绝”的标记然后跳转到一个“修改计划”的节点从而实现循环。6. 运行智能体与效果验证图已经编译好现在让我们运行它看看这个旅行规划智能体如何工作。创建主程序main.py。# main.py from agents.graph import travel_agent_graph from langchain_core.messages import HumanMessage import asyncio async def main(): 主函数运行旅行规划智能体。 print( 旅行规划智能体启动 ) # 模拟用户输入 user_input 我想下个月去一个温暖、有美食的海边城市放松3天预算中等。 print(f用户请求: {user_input}) # 构建初始状态 initial_state { messages: [HumanMessage(contentuser_input)], user_request: user_input, destination: None, days: None, budget: None, destination_info: None, detailed_plan: None, needs_human_approval: False, } # 运行图 print(\n--- 智能体开始执行 ---) try: # 使用流式输出可以看到执行到哪个节点 async for event in travel_agent_graph.astream_events(initial_state, versionv1): kind event[event] node event.get(name, N/A) if kind on_chain_start and event[metadata].get(step) 1: # 这里可以打印节点开始信息 pass elif kind on_chain_end and event[metadata].get(step) 1: print(f[执行完成] 节点: {node}) except Exception as e: print(f执行过程中出现错误: {e}) # 即使流式事件出错我们也尝试获取最终结果 final_state await travel_agent_graph.ainvoke(initial_state) else: # 正常执行后获取最终状态 final_state await travel_agent_graph.ainvoke(initial_state) print(\n--- 执行完成 ---) # 输出关键结果 print(f\n✅ 解析出的目的地: {final_state.get(destination)}) print(f✅ 解析出的天数: {final_state.get(days)}) print(f✅ 解析出的预算: {final_state.get(budget)}) print(f\n 生成的旅行计划 (摘要):) plan final_state.get(detailed_plan, 无) # 只打印前500字符作为预览 print(plan[:500] ... if len(plan) 500 else plan) print(f\n 最终对话消息数: {len(final_state.get(messages, []))}) # 可以选择保存计划到文件 save_to_file input(\n是否将完整计划保存到文件(y/n): ).lower().strip() if save_to_file y: with open(ftravel_plan_{final_state.get(destination, unknown)}.txt, w, encodingutf-8) as f: f.write(plan) print(f计划已保存至 travel_plan_{final_state.get(destination, unknown)}.txt) if __name__ __main__: asyncio.run(main())6.1 运行程序在终端中执行python main.py6.2 预期输出与解读程序会依次执行各个节点并在控制台输出类似以下内容 旅行规划智能体启动 用户请求: 我想下个月去一个温暖、有美食的海边城市放松3天预算中等。 --- 智能体开始执行 --- [执行完成] 节点: parse_request [执行完成] 节点: search_info [执行完成] 节点: generate_plan [执行完成] 节点: wait_approval [执行完成] 节点: finalize --- 执行完成 --- ✅ 解析出的目的地: 厦门 ✅ 解析出的天数: 3 ✅ 解析出的预算: 中等 生成的旅行计划 (摘要): 【最终旅行计划】 **行程概览** 本次厦门3日休闲美食之旅以“慢生活、深体验”为主题聚焦鼓浪屿、环岛路、沙坡尾等核心区域品尝地道海鲜、沙茶面、土笋冻等闽南风味预算控制在中等水平... 最终对话消息数: 4执行流程解读parse_request节点LLM从“温暖、有美食的海边城市”推断出具体目的地“厦门”并提取出天数“3”和预算“中等”。search_info节点调用Tavily搜索API获取厦门3日游、中等预算的攻略、景点和美食信息。generate_plan节点综合所有信息生成一份详细的、结构化的旅行计划草案并将needs_human_approval设为True。wait_approval节点模拟用户确认环节。在我们的代码中它自动模拟用户同意并将needs_human_approval设为False。finalize节点对草案进行最终润色输出鼓舞人心的最终版计划。整个流程清晰可见状态目的地、天数、计划草案、确认标志在节点间有序传递和更新。7. 深入探索实现条件循环与错误处理上面的例子是线性流程。LangGraph的强大之处在于处理非线性逻辑。让我们增强智能体使其在用户“拒绝”计划时能够循环回去修改。7.1 修改状态与节点首先在状态中增加一个字段来记录用户反馈。# agents/state.py (修改部分) class TravelAgentState(TypedDict): # ... 保留原有字段 ... # 新增用户对上一版计划的反馈 user_feedback: Optional[str] # 新增计划修改次数 revision_count: int然后修改human_approval_node和generate_travel_plan节点并新增一个revise_plan_node。# agents/nodes.py (新增和修改函数) def human_approval_node_with_feedback(state: TravelAgentState) - TravelAgentState: 修改后的确认节点模拟获取用户反馈。 在实际应用中这里从UI获取“同意”、“拒绝并给出修改意见”。 # 模拟交互我们假设第一版计划被拒绝用户给出了反馈 if state.get(revision_count, 0) 0: # 第一次生成计划模拟用户拒绝并要求修改 feedback 我不太想去鼓浪屿人太多了。能不能推荐一些更小众、安静的地方 approval_msg HumanMessage(contentf我拒绝这个计划。反馈{feedback}) needs_approval True else: # 第二次修改后模拟用户同意 feedback 这次修改得很好我同意了 approval_msg HumanMessage(contentfeedback) needs_approval False return { messages: [approval_msg], user_feedback: feedback, needs_human_approval: needs_approval } def revise_plan_node(state: TravelAgentState) - TravelAgentState: 新节点根据用户反馈修改旅行计划。 llm get_cached_llm() old_plan state[detailed_plan] feedback state[user_feedback] revision_count state.get(revision_count, 0) 1 revision_prompt f 这是一份旅行计划收到了用户的修改意见。请根据意见重新规划。 **原计划** {old_plan} **用户反馈** {feedback} **修改要求** 1. 充分考虑用户反馈。 2. 保留原计划中用户未提及的合理部分。 3. 生成一份新的、完整的旅行计划。 这是第{revision_count}次修改。 response llm.invoke([SystemMessage(content你是耐心细致的旅行规划师善于根据反馈调整方案。), HumanMessage(contentrevision_prompt)]) revised_plan response.content new_message AIMessage(contentf已根据您的反馈完成了第{revision_count}次修改请查阅新计划。) return { detailed_plan: revised_plan, messages: [new_message], revision_count: revision_count, needs_human_approval: True # 修改后再次等待确认 }7.2 修改图定义以实现循环现在修改agents/graph.py引入条件循环。# agents/graph.py (修改后的create_travel_agent_graph函数) def create_travel_agent_graph_with_loop(): workflow StateGraph(TravelAgentState) # 添加所有节点包括新的revise_plan节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(search_info, search_destination_info) workflow.add_node(generate_plan, generate_travel_plan) workflow.add_node(wait_approval, human_approval_node_with_feedback) # 使用新的确认节点 workflow.add_node(revise_plan, revise_plan_node) # 新增修改节点 workflow.add_node(finalize, finalize_plan) workflow.set_entry_point(parse_request) workflow.add_edge(parse_request, search_info) workflow.add_edge(search_info, generate_plan) workflow.add_edge(generate_plan, wait_approval) # 关键定义从 wait_approval 出发的条件边 def route_after_approval(state: TravelAgentState): 根据用户确认状态决定下一步 if state.get(needs_human_approval): # 如果需要确认即用户拒绝则去修改计划 return revise_plan else: # 如果不需要确认即用户同意则去最终定稿 return finalize workflow.add_conditional_edges( wait_approval, route_after_approval, { revise_plan: revise_plan, finalize: finalize } ) # 从修改计划节点再次回到等待确认节点形成循环 workflow.add_edge(revise_plan, wait_approval) # 为防止无限循环可以设置一个最大修改次数检查在revise_plan节点或条件边中逻辑判断 # 这里我们在条件边函数中添加简单判断 def route_after_approval_with_limit(state: TravelAgentState): revision_count state.get(revision_count, 0) if revision_count 3: # 最多修改3次 print(已达到最大修改次数强制通过。) return finalize return route_after_approval(state) # 更新条件边函数 # workflow.add_conditional_edges(wait_approval, route_after_approval_with_limit, ...) workflow.add_edge(finalize, END) return workflow.compile()这个新的图实现了一个完整的“生成-确认-修改”循环。只有当用户同意needs_human_approval为False时才会跳出循环进入finalize节点。我们还添加了revision_count和简单的限制逻辑来防止无限循环。8. 常见问题与排查思路在实际使用LangGraph时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案ValueError: Missing keys: ...状态StateTypedDict中定义的某些键在节点返回的更新字典中没有提供初始值。检查状态类定义和每个节点函数的返回值。确保所有在TypedDict中声明为非Optional的键在初始状态和每次状态更新中都存在。1. 将状态字段改为Optional[...]。2. 确保所有节点都返回完整的字段更新或使用...操作符保留其他字段。图编译成功但执行时无任何输出或卡住1. 边Edge没有正确连接导致图没有终点END。2. 使用了异步astream但未在异步环境中运行。1. 使用graph.get_graph().draw_mermaid()输出图结构可视化检查路径是否通达。2. 确认在异步函数中调用astream或用asyncio.run()包装。1. 确保所有路径最终都能到达END节点。2. 使用同步的invoke方法测试或确保在async def函数中运行。工具调用失败如搜索API报错1. API密钥未设置或错误。2. 网络问题。3. 工具参数格式错误。1. 检查.env文件和环境变量。2. 在节点函数中添加try...except打印具体错误。3. 单独测试工具调用。1. 正确配置API密钥。2. 实现降级策略如返回模拟数据。3. 查阅对应工具的官方文档。LLM输出格式不符合预期导致后续节点解析失败提示词Prompt不够精确LLM返回了非结构化文本。打印出LLM的原始响应内容。1. 在提示词中明确要求输出格式如“请以JSON格式返回”。2. 使用LangChain的OutputParser如PydanticOutputParser来强制结构化输出。3. 在节点代码中添加更健壮的异常处理如json.loads失败后的默认值。条件边Conditional Edge逻辑不生效条件函数返回的值与add_conditional_edges中定义的映射键不匹配。打印条件函数的输入state和返回值。确保条件函数返回的字符串与add_conditional_edges的映射字典中的某个键完全一致大小写敏感。状态更新不符合预期如消息被覆盖对messages这类列表字段没有使用Annotated和add_messages。检查状态类中列表字段的定义。对于需要追加而非替换的字段使用Annotated[List[BaseMessage], add_messages]语法。对于其他自定义列表可以定义自己的归约函数。9. 最佳实践与工程化建议将LangGraph智能体用于生产环境需要遵循以下工程最佳实践9.1 状态设计原则最小化与清晰化只把真正需要在节点间传递的数据放入状态。避免状态字典过于庞大。使用TypedDict强烈推荐使用TypedDict并配合mypy或pyright进行类型检查能在开发阶段避免大量键错误。区分对话历史与内部状态像messages这样的对话历史使用add_messages自动管理。业务状态如destination,plan单独存储。9.2 节点函数设计单一职责每个节点只做一件事如解析、搜索、生成、验证。纯函数化节点函数应尽可能只依赖于输入状态和注入的依赖如LLM、工具避免副作用。这便于测试和调试。完善的错误处理节点内部应对可能失败的操作网络调用、JSON解析进行try-catch并返回错误信息或默认状态避免整个图执行中断。9.3 图的构建与调试可视化是利器使用graph.get_graph().draw_mermaid()生成Mermaid图表直观理解工作流。这对于复杂流程图和向团队解释设计至关重要。版本化配置将图的构建逻辑封装在函数中便于根据配置如不同环境、不同功能开关创建不同的图实例。利用检查点Checkpoint对于长时运行或需要中断恢复的智能体使用LangGraph的检查点功能持久化状态。这对于需要“人工介入”然后继续的任务非常有用。9.4 测试与监控单元测试节点单独测试每个节点函数模拟输入状态验证输出状态。集成测试全图使用固定的初始状态和Mock的LLM、工具来测试整个图的执行路径和最终输出。记录与追踪利用astream_events或LangSmith等工具详细记录每个节点的输入、输出和执行时间便于问题排查和性能分析。9.5 与MCPModel Context Protocol集成MCP协议正在成为连接AI智能体与外部工具、数据源的重要标准。LangGraph可以与MCP Server无缝集成将MCP Server作为工具你可以将任何MCP Server如连接数据库、内部API的Server封装成LangChain Tool然后在LangGraph的节点中调用。构建更强大的智能体通过MCP你的智能体可以安全、标准化地访问代码库如Git、项目管理工具如Jira、设计文件如Figma实现真正的“AI员工”。实践步骤部署或连接一个MCP Server例如用于查询数据库的Server。使用langchain-mcp-adapters等库将MCP Server提供的工具动态加载为LangChain Tool。将这些Tool添加到你的LangGraph智能体的工具集中。通过遵循这些实践你可以构建出结构清晰、易于维护、可扩展且稳健的AI智能体应用。LangGraph提供的不仅是一个框架更是一种构建复杂、可维护AI工作流的工程范式。从简单的线性流程开始逐步引入条件分支、循环、子图和持久化你将能驾驭越来越复杂的智能体应用场景。