如果你是一名开发者最近在尝试构建一个具备复杂逻辑和自主决策能力的智能体Agent或者想为你的应用注入一个能“思考”和“行动”的“大脑”那么你很可能已经接触过或听说过LangGraph。LangGraph 不是一个新模型而是一个构建在 LangChain 之上的框架。很多人初看会以为它只是 LangChain 的一个“流程图”工具用来画一画 Agent 的执行步骤。但如果你只停留在这个认知层面就错过了它最核心的价值——它真正解决的是 Agent 工作流中“状态”的持久化、流转与复杂编排问题。过去我们构建一个多步骤的 Agent常常需要自己管理对话历史、工具调用结果、中间决策等一堆状态变量代码很快就会变得混乱且难以维护。LangGraph 通过引入“图”Graph和“状态”State这两个核心抽象将 Agent 的执行过程标准化、可视化并且最重要的是可持久化和可回溯。这意味着你可以构建出能处理复杂分支、循环、甚至具备长期记忆的智能体。本文不会只停留在概念介绍。我们将通过一个完整的实战项目从零开始构建一个具备“记忆”和“规划”能力的旅行规划智能体。你将看到LangGraph 的核心概念如何映射到代码。如何定义和管理智能体的“状态”。如何编排工具调用、LLM 决策和条件分支。如何让智能体记住之前的对话和决策。最终你将获得一个可以直接运行和扩展的代码库。无论你是想深入了解 LangGraph 的机制还是急需一个可复用的 Agent 项目模板这篇文章都将为你提供清晰的路径和落地方案。1. 理解 LangGraph不止是“画流程图”在深入代码之前我们必须先厘清几个关键概念否则很容易在后续的配置中感到困惑。核心概念一图Graph你可以把图理解为一个由节点Nodes和边Edges组成的工作流。节点代表一个执行单元比如调用一次 LLM、运行一个工具边代表执行流的方向。LangGraph 允许你定义复杂的拓扑结构包括顺序、分支、循环这远超过简单的线性链Chain。核心概念二状态State这是 LangGraph 的灵魂。整个图在一个共享的State对象上运行。State是一个字典或 Pydantic 模型定义了工作流中需要传递和更新的所有数据。例如它可能包含messages对话历史、next下一步该执行哪个节点、destination用户输入的目的地等。每个节点读取并修改这个共享状态从而实现了信息的无缝流转。核心概念三节点Node与边Edge节点一个普通的 Python 函数它接收当前的State执行一些操作如调用 LLM、查询数据库然后返回一个包含更新内容的字典。这个字典会被合并到全局State中。边决定在某个节点执行完毕后接下来应该执行哪个节点。边可以是固定的always_go_to也可以是基于State内容动态决定的conditional_edges。它解决了什么问题想象一下你要构建一个旅行助手 Agent用户说“我想去北京玩。”Agent 需要先询问出行时间和预算LLM 对话。根据时间查询北京的天气调用天气工具。根据预算和天气推荐室内或室外景点调用推荐工具并需要之前步骤的结果。最后生成一个行程草稿。如果没有 LangGraph你需要手动维护“用户输入”、“出行时间”、“预算”、“天气数据”、“景点列表”这一系列状态并在不同的函数间传递逻辑分支会让代码像一团乱麻。而 LangGraph 通过一个明确定义的State和清晰的图结构让这一切变得井然有序。2. 环境准备与项目初始化我们将使用 Python 作为开发语言。请确保你的环境满足以下要求Python 版本 3.8包管理工具pip 或 condaLLM 服务我们将使用 OpenAI 的 GPT 模型你需要准备一个有效的OPENAI_API_KEY。你也可以替换为其他兼容 OpenAI API 的模型如 Azure OpenAI, Ollama 等。第一步创建项目目录并安装依赖# 创建项目目录 mkdir travel_agent_with_langgraph cd travel_agent_with_langgraph # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心依赖 pip install langgraph langchain-openai langchain-communitylanggraph: 核心框架。langchain-openai: 用于调用 OpenAI 模型。langchain-community: 包含一些社区维护的工具如网络搜索我们可能会用到。第二步准备配置文件为了安全地管理 API Key我们创建一个.env文件。# 在项目根目录创建 .env 文件 touch .env在.env文件中填入你的 OpenAI API KeyOPENAI_API_KEYsk-your-actual-openai-api-key-here第三步创建主程序文件touch main.py现在你的项目结构应该如下所示travel_agent_with_langgraph/ ├── venv/ # 虚拟环境目录 ├── .env # 环境变量文件 ├── main.py # 主程序文件 └── requirements.txt # 可选依赖列表文件3. 定义智能体的“状态”State状态是工作流的“记忆中枢”。我们使用 Pydantic 的BaseModel来严格定义状态的结构这有助于类型检查和自动补全。在main.py中我们开始编写代码# main.py from typing import List, Optional, Literal, Annotated from typing_extensions import TypedDict from pydantic import BaseModel, Field from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() # --- 第1部分定义状态State--- class AgentState(TypedDict): 定义智能体的全局状态。 这是一个 TypedDictLangGraph 推荐使用它来获得更好的类型提示。 关键字段 - messages: 完整的对话历史是工作流的核心输入和输出。 - destination: 用户输入的目的地。 - travel_date: 用户输入的出行日期。 - budget: 用户输入的预算。 - weather_info: 查询到的天气信息。 - recommendations: 生成的景点推荐。 - itinerary: 最终生成的行程草案。 - next: 指示下一步该执行哪个节点可选用于条件路由。 messages: Annotated[List[HumanMessage | AIMessage], 对话历史记录] destination: Annotated[Optional[str], 旅行目的地] travel_date: Annotated[Optional[str], 出行日期] budget: Annotated[Optional[str], 预算范围] weather_info: Annotated[Optional[str], 天气查询结果] recommendations: Annotated[Optional[str], 景点推荐结果] itinerary: Annotated[Optional[str], 行程草案] next: Annotated[Optional[str], 下一步节点]代码解释我们定义了一个AgentState类它继承自TypedDict。这明确规定了状态对象中每个字段的名称和类型。Annotated用于为字段添加描述这在后续的可视化中可能会有用。messages字段是核心它存储了所有HumanMessage用户消息和AIMessageAI 回复LangGraph 和 LangChain 的很多组件都依赖这个格式的对话历史。其他字段如destination、weather_info等都是我们为这个特定旅行 Agent 设计的自定义状态用于在节点间传递信息。4. 构建图的工作节点Nodes节点是执行具体任务的地方。我们将创建四个关键节点路由节点Router分析用户输入决定工作流走向是收集信息还是直接规划。信息收集节点CollectInfo与用户交互补全出行所需的关键信息目的地、日期、预算。查询天气节点GetWeather调用外部工具获取目的地天气。规划行程节点PlanItinerary综合所有信息调用 LLM 生成最终行程。首先初始化 LLM 和模拟工具# --- 第2部分初始化 LLM 和工具 --- llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 模拟一个天气查询工具在实际应用中你会连接真实的天气API def mock_weather_api(destination: str, date: str) - str: 模拟天气查询工具。 # 这里应该是一个真实的 API 调用例如 # response requests.get(fhttps://weather.api?city{destination}date{date}) # return response.json()[forecast] print(f[模拟工具调用] 查询 {destination} 在 {date} 的天气) return f{destination}在{date}的天气预计为晴朗气温20-25度。适合户外活动。 # 模拟一个景点推荐工具 def mock_recommendation_api(destination: str, interest: str) - str: 模拟景点推荐工具。 print(f[模拟工具调用] 为{destination}推荐{interest}相关的景点) return f根据您的兴趣推荐{destination}的故宫、天坛和颐和园。然后开始定义节点函数# --- 第3部分定义节点函数 --- def router_node(state: AgentState) - dict: 路由节点分析最新用户消息决定下一步是收集信息还是直接规划。 规则 - 如果状态中已包含目的地、日期、预算则跳转到 plan。 - 否则跳转到 collect_info。 print(f\n--- [路由节点] 正在分析状态 ---) # 检查关键信息是否齐全 if state.get(destination) and state.get(travel_date) and state.get(budget): print( - 信息已齐全转向规划节点。) return {next: plan} else: print( - 信息不全转向信息收集节点。) return {next: collect_info} def collect_info_node(state: AgentState) - dict: 信息收集节点与用户对话补全缺失的关键信息。 它读取对话历史让LLM判断缺失哪项信息并生成询问。 print(f\n--- [信息收集节点] 正在运行 ---) # 从状态中获取当前对话 messages state[messages] # 构建一个系统提示指导LLM进行信息收集 system_prompt 你是一个旅行助手正在帮助用户规划行程。你需要收集以下关键信息 1. 旅行目的地例如北京、上海 2. 出行日期例如下周末、国庆假期 3. 预算范围例如经济型、豪华型 请根据当前的对话历史判断哪一项信息还缺失然后生成一个自然的问题来询问用户。 只输出你的问题不要输出其他内容。 # 将系统提示和历史对话一起发给LLM prompt [{role: system, content: system_prompt}] messages response llm.invoke(prompt) # 将LLM生成的问题添加到消息历史中作为AI的回复 ai_message AIMessage(contentresponse.content) new_messages messages [ai_message] print(f - AI 询问: {response.content}) # 注意这个节点不更新 destination/date/budget这些将由后续的“更新状态节点”处理。 # 它只负责推进对话。 return {messages: new_messages} def get_weather_node(state: AgentState) - dict: 查询天气节点调用外部工具获取目的地天气信息。 print(f\n--- [查询天气节点] 正在运行 ---) destination state[destination] travel_date state[travel_date] if not destination or not travel_date: weather 无法查询天气目的地或日期缺失。 else: weather mock_weather_api(destination, travel_date) print(f - 天气信息: {weather}) return {weather_info: weather} def plan_itinerary_node(state: AgentState) - dict: 规划行程节点综合所有信息生成旅行行程草案。 print(f\n--- [规划行程节点] 正在运行 ---) destination state[destination] travel_date state[travel_date] budget state[budget] weather state.get(weather_info, 未知) # 构建给LLM的提示词包含所有收集到的信息 prompt f你是一个专业的旅行规划师。请根据以下信息为用户生成一份详细的旅行行程草案。 目的地{destination} 出行日期{travel_date} 预算{budget} 天气情况{weather} 行程草案需要包含 1. 每日的上午、下午、晚上的活动安排。 2. 活动类型观光、美食、休闲等与具体地点推荐。 3. 简要的交通和餐饮建议。 4. 根据预算和天气给出的贴心提示。 请以清晰、友好的格式输出。 response llm.invoke([HumanMessage(contentprompt)]) itinerary response.content print(f - 行程生成完毕长度: {len(itinerary)} 字符) # 将最终行程添加到消息历史中返回给用户 ai_message AIMessage(contentitinerary) new_messages state[messages] [ai_message] return {itinerary: itinerary, messages: new_messages}关键点解析每个节点函数都接收AgentState作为参数并返回一个字典。这个字典中的键值对将被“合并”到全局状态中。router_node不修改messages它只根据现有状态计算下一个节点。collect_info_node的核心是调用 LLM 来生成对话它更新了messages但并未直接解析出destination等具体信息。在实际更复杂的系统中你可能需要另一个节点或工具来从对话中提取实体并更新状态。我们使用了模拟工具mock_weather_api。在真实项目中这里应替换为真实的 API 调用并做好错误处理。5. 组装图定义节点与边现在我们将各个节点组装成一个有向图并定义它们之间的执行顺序。# --- 第4部分组装图Graph--- print(\n 正在构建 LangGraph 工作流 ) # 1. 创建一个图构建器并指定状态的结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(router, router_node) workflow.add_node(collect_info, collect_info_node) workflow.add_node(get_weather, get_weather_node) workflow.add_node(plan, plan_itinerary_node) # 3. 设置入口点所有流程都从 router 节点开始 workflow.set_entry_point(router) # 4. 添加条件边Conditional Edges # 从 router 节点出来根据其返回状态中的 next 字段值决定去往哪个节点。 workflow.add_conditional_edges( router, # 这是一个路由函数它检查状态中的 next 字段 lambda state: state.get(next), # 映射关系next 字段的值 - 下一个节点的名称 { collect_info: collect_info, plan: plan, } ) # 5. 添加普通边固定流转 # 信息收集完成后自动去查询天气 workflow.add_edge(collect_info, get_weather) # 查询天气完成后自动去规划行程 workflow.add_edge(get_weather, plan) # 规划行程完成后工作流结束 workflow.add_edge(plan, END) # 6. 编译图生成可执行对象 app workflow.compile()图结构解读 这个图定义了一个清晰的执行逻辑起点是router节点。router检查状态如果信息不全就去collect_info如果信息齐全就直接去plan。collect_info节点会与用户进行一轮对话在我们的示例中是AI主动提问。完成后无条件流向get_weather。get_weather节点调用工具获取天气完成后无条件流向plan。plan节点生成最终行程然后工作流到达END终点。这个流程体现了“条件分支”router和“顺序执行”collect_info - get_weather - plan的组合。6. 运行与测试智能体图编译完成后我们就可以用它来处理用户请求了。我们需要初始化一个初始状态然后调用app.invoke()。# --- 第5部分运行智能体 --- def run_agent(user_input: str): 运行智能体的主函数。 print(f\n 用户输入: {user_input}) # 初始化状态。初始消息是用户的输入。 initial_state: AgentState { messages: [HumanMessage(contentuser_input)], destination: None, travel_date: None, budget: None, weather_info: None, recommendations: None, itinerary: None, next: None, } # 执行图 final_state app.invoke(initial_state) print(f\n 执行完成 ) # 打印最终的行程结果 if final_state.get(itinerary): print(\n【生成的行程草案】) print(final_state[itinerary]) else: print(未生成行程。) # 打印完整的对话历史 print(f\n【完整对话历史】) for msg in final_state[messages]: print(f{msg.type}: {msg.content[:100]}...) # 只打印前100字符 # 测试场景1用户直接提供了完整信息 print(\n *50) print(测试场景1用户提供完整信息) print(*50) run_agent(我想下周末去北京玩预算中等。) # 测试场景2用户只提供了部分信息 print(\n\n *50) print(测试场景2用户信息不全) print(*50) # 注意为了模拟多轮对话我们需要手动“接力”状态。 # 在实际的聊天机器人中这个状态是持久化的每次调用都是接着上次的最终状态。 initial_state_2: AgentState { messages: [HumanMessage(content帮我规划一个旅行)], # 信息不全 destination: None, travel_date: None, budget: None, weather_info: None, recommendations: None, itinerary: None, next: None, } # 第一轮执行 state_after_round1 app.invoke(initial_state_2) print(f第一轮AI回复: {state_after_round1[messages][-1].content}) # 假设用户回答了AI的问题“我想去上海国庆期间预算宽松。” # 我们需要手动更新状态模拟用户的第二次输入。 new_human_message HumanMessage(content我想去上海国庆期间预算宽松。) state_after_round1[messages].append(new_human_message) # 关键更新从对话中提取出的实体信息这里简化处理直接赋值 state_after_round1[destination] 上海 state_after_round1[travel_date] 国庆期间 state_after_round1[budget] 宽松 # 重新设置next让流程继续 state_after_round1[next] None # 第二轮执行 print(\n 用户第二次输入: 我想去上海国庆期间预算宽松。) final_state_2 app.invoke(state_after_round1) if final_state_2.get(itinerary): print(\n【生成的行程草案】) print(final_state_2[itinerary])运行与观察 执行python main.py你会看到控制台打印出详细的节点执行日志并最终输出生成的行程草案。通过两个测试场景你可以清晰地看到场景1由于初始状态就包含了目的地、日期、预算router节点直接指向plan但由于plan需要天气信息而get_weather节点没有被执行所以行程中天气部分是“未知”。这暴露了我们图逻辑的一个缺陷规划前应该确保天气信息已获取。场景2展示了多轮对话。第一轮AI会提问第二轮用户补充信息后流程会继续执行get_weather和plan节点。7. 优化与进阶让智能体更完善上面的基础版本揭示了几个问题也是你实际项目中需要优化的方向问题1信息提取我们的collect_info_node只负责提问没有从用户的回答中自动提取destination、travel_date等信息并更新状态。这需要增强。解决方案添加信息提取节点# 新增一个节点函数用于从最新对话中提取结构化信息 def extract_info_node(state: AgentState) - dict: 信息提取节点从最新的对话消息中提取目的地、日期、预算等信息。 print(f\n--- [信息提取节点] 正在运行 ---) messages state[messages] # 取最后两条消息假设是用户的最新回复和AI的上一个问题 recent_msgs messages[-2:] if len(messages) 2 else messages extraction_prompt f请从以下对话中提取旅行规划相关的结构化信息。如果提到请提取 1. 目的地城市或景点名 2. 出行日期或时间段 3. 预算范围如经济、中等、豪华 对话内容 {recent_msgs} 请以JSON格式输出只包含这三个字段destination, travel_date, budget。如果某个字段未提及其值为null。 response llm.invoke([HumanMessage(contentextraction_prompt)]) # 这里需要解析LLM返回的JSON。实际应用中应使用LangChain的OutputParser或做try-catch。 # 为简化示例我们假设它返回了正确的JSON。 import json try: extracted json.loads(response.content) print(f - 提取到信息: {extracted}) # 只更新非None的值 update_dict {k: v for k, v in extracted.items() if v is not None} return update_dict except json.JSONDecodeError: print( - 信息提取失败无法解析JSON。) return {}然后你需要修改图的结构在collect_info节点后get_weather节点前插入这个extract_info节点。问题2图逻辑缺陷规划行程前必须确保天气信息已查询。解决方案修改路由逻辑我们可以修改router_node或图的结构确保规划路径 (plan) 必须经过get_weather。一个更健壮的设计是引入一个“规划准备”节点它检查所有前置条件信息齐全、天气已查如果满足则进入plan否则触发相应的补全流程。问题3持久化与多轮对话我们的示例中多轮对话是通过手动传递state模拟的。在生产环境中你需要将state与用户的会话ID绑定存储到数据库如Redis、PostgreSQL中每次新的用户消息传入时加载对应的历史状态调用app.invoke再保存新状态。8. 常见问题与排查思路在开发 LangGraph 应用时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案KeyError当节点访问状态字段状态State的 TypedDict 定义与实际返回的字典键不匹配。1. 检查AgentState定义的所有字段名。2. 检查每个节点函数return的字典键是否与定义一致。3. 使用print(state.keys())在节点内调试。确保节点返回的字典键是AgentState中定义的字段的子集。图编译错误提示节点未定义在add_edge或add_conditional_edges时引用了不存在的节点名。仔细检查add_node时使用的名称与add_edge时引用的名称是否完全一致大小写敏感。统一节点名称的字符串建议使用常量定义。条件边不生效总是走固定分支add_conditional_edges中的路由函数返回的值与提供的映射字典的键不匹配。1. 在路由函数中打印其返回值。2. 检查映射字典的键是否包含了所有可能的返回值。确保路由函数返回的值是映射字典中存在的键。也可以使用default参数指定默认流向。工作流陷入无限循环图中存在环且没有设置终止条件。例如collect_info完成后又指回router而router又指回collect_info。1. 可视化你的图结构LangGraph 支持导出为PNG。2. 检查是否存在A-B-A这样的环。确保每个执行路径最终都能到达END节点。对于需要循环的场景如持续对话应明确设置循环退出条件。LLM 调用超时或报错API Key 错误、网络问题、模型服务不可用、提示词导致输出过长。1. 检查.env文件和环境变量。2. 尝试简单的llm.invoke(“Hello”)测试连通性。3. 简化提示词再试。配置正确的 API Key 和 Base URL。增加超时设置。对提示词进行分步优化和测试。状态更新不符合预期多个节点修改了同一个状态字段后者覆盖了前者。LangGraph 的默认合并策略是“覆盖”。理解状态合并策略。默认是“浅合并”后执行的节点返回值会覆盖先执行节点的同名字段。对于需要“追加”操作的字段如messages在节点中返回{“messages”: existing_messages [new_message]}。对于其他字段注意执行顺序。9. 最佳实践与工程建议基于项目经验以下建议能帮助你构建更稳健、可维护的 LangGraph 应用状态设计要精简且明确State不是数据垃圾桶。只定义工作流真正需要流转的数据。过多的字段会增加复杂度和调试难度。使用Optional类型明确哪些字段可能为空。节点功能保持单一职责一个节点最好只做一件事如路由、调用一个工具、调用一次LLM。这提高了节点的可测试性和可复用性。善用条件边实现复杂逻辑add_conditional_edges是实现分支、循环的核心。路由函数应尽量简单只基于State做判断。复杂的决策逻辑可以封装到一个专门的“决策节点”中。为图添加可视化LangGraph 支持将图导出为图片这对于理解复杂工作流和向他人解释设计至关重要。在开发初期就养成可视化的习惯。from langgraph.graph import StateGraph # ... 构建图 workflow ... # 导出为PNG image_data workflow.get_graph().draw_mermaid_png() with open(travel_agent_workflow.png, wb) as f: f.write(image_data) print(工作流图已保存为 travel_agent_workflow.png)实现持久化存储对于生产环境必须将State与用户会话关联并持久化。可以定义一个StateStore类使用数据库来save_state和load_state。在每次app.invoke前后调用它们。加入错误处理与回退在节点函数中对 LLM 调用和工具调用使用try...except。可以设计一个fallback_node当关键节点失败时将流程导向此处给用户一个友好的错误提示而不是让整个工作流崩溃。编写单元测试由于节点是纯函数输入状态输出状态变更它们非常容易进行单元测试。为每个节点编写测试用例模拟不同的输入状态验证输出是否符合预期。通过本文的旅程我们从零构建了一个具备基本逻辑的 LangGraph 智能体。你掌握了其状态驱动和图编排的核心思想并看到了一个从路由、信息收集、工具调用到最终规划的全流程实现。LangGraph 的强大之处在于当你需要为智能体增加新功能例如查询机票、预订酒店、生成预算清单时你只需要定义新的节点和工具然后将它们以清晰的逻辑“插入”到现有的图中即可而无需重写整个混乱的控制流。这种模块化和可维护性正是构建复杂 Agent 系统时所急需的。下一步你可以尝试将模拟工具替换为真实的 API 调用如 SerpAPI 搜索、真实天气接口。实现更强大的信息提取逻辑使用 LangChain 的PydanticOutputParser来确保 LLM 输出结构化数据。引入Human-in-the-loop节点在关键决策点如确认高额预算暂停流程等待用户确认。探索 LangGraph 的“检查点”Checkpointing功能它能自动保存每个步骤的状态实现智能体的暂停、恢复和回溯这是构建长周期、复杂任务 Agent 的终极利器。