在构建复杂AI应用时你是否遇到过这样的困境多个LLM调用、工具执行和状态管理逻辑交织在一起代码迅速变得难以维护传统的LangChain虽然强大但在处理多步骤、有状态的工作流时其线性链式结构常常力不从心。这正是LangGraph要解决的核心问题。本文将从零开始手把手带你掌握LangGraph这一构建智能体Agent和复杂工作流的强大框架通过完整的实战案例让你不仅能理解其核心概念更能独立开发出功能强大的AI应用。1. LangGraph核心概念与背景1.1 什么是LangGraphLangGraph是LangChain生态系统中的一个库它扩展了LangChain的核心Runtime专门用于构建有状态、多参与者的应用程序。你可以把它想象成一个为AI应用设计的“工作流引擎”或“状态机”。与传统的线性链Chain不同LangGraph允许你定义包含循环、条件分支和并行执行的图Graph结构从而能够优雅地处理需要长时间运行、具备记忆能力或涉及多个决策点的复杂AI任务。1.2 LangGraph解决了什么问题在AI应用开发中尤其是智能体Agent场景我们经常需要维持对话状态记住之前的交互历史。根据条件选择路径例如根据用户问题的类型决定调用哪个工具或模型。处理循环和迭代比如让Agent反复思考直到得出满意答案。协调多个组件串联或并联地使用多个LLM、工具或链。使用基础的LangChain链来实现上述功能代码会充斥着大量的if-else和手动状态管理。LangGraph通过将工作流定义为“图”将状态管理抽象化使得这类应用的构建变得清晰、模块化且易于调试。1.3 LangGraph vs. LangChain核心区别这是一个常见的困惑点。简单来说LangChain是一个用于构建由语言模型驱动的应用程序的综合框架。它提供了模型调用、提示模板、记忆、索引、链等大量组件。链Chain是其核心抽象之一但通常是线性的。LangGraph是LangChain的一个库它建立在LangChain的组件之上提供了一个新的抽象——图Graph。它专注于管理有状态、多步骤的工作流。你可以使用LangChain的所有组件如LLM、工具作为LangGraph图中的节点。类比如果把构建AI应用比作造车LangChain提供了发动机LLM、方向盘提示、轮胎工具等各种零件和组装简单车辆的说明书链。而LangGraph则提供了一套设计并组装复杂传动系统、悬挂和控制系统即包含反馈循环、条件判断的复杂工作流的蓝图和工具。1.4 核心概念解析在深入学习前必须理解以下几个核心概念节点Node图中的一个执行单元。通常是一个函数它接收当前状态执行一些操作如调用LLM、运行工具并返回一个更新后的状态。边Edge连接节点的路径决定工作流的走向。边可以是条件边根据状态值决定下一步走哪个节点或普通边无条件指向下一个节点。状态State在整个图执行过程中传递和修改的数据。LangGraph使用TypedDict或Pydantic模型来严格定义状态的模式。检查点CheckpointLangGraph能够自动保存执行过程中的状态快照。这实现了两个关键功能长期记忆在应用重启后恢复对话和****从错误中恢复或人工干预后继续执行。2. 环境准备与项目初始化2.1 环境要求我们将使用Python进行开发。请确保你的环境满足以下要求操作系统Windows 10/11 macOS 或 Linux。Python版本 3.8.1, 3.13。推荐使用3.10或3.11以获得最佳兼容性。包管理工具pip。2.2 创建虚拟环境与安装依赖强烈建议使用虚拟环境来隔离项目依赖。# 1. 创建项目目录并进入 mkdir langgraph-tutorial cd langgraph-tutorial # 2. 创建Python虚拟环境 (以Python3.10为例) python3.10 -m venv venv # 3. 激活虚拟环境 # Windows (PowerShell) .\venv\Scripts\Activate.ps1 # macOS/Linux source venv/bin/activate # 4. 安装核心依赖 pip install langgraph langchain langchain-openai依赖说明langgraph 核心框架。langchain 提供LLM、提示、工具等基础组件。langchain-openai 用于调用OpenAI的模型如GPT-4 GPT-3.5-turbo。你也可以安装langchain-anthropic等适配其他模型的包。2.3 配置API密钥为了调用OpenAI的模型你需要设置API密钥。切勿将密钥硬编码在代码中或提交到版本控制系统。# 在命令行中设置环境变量临时 # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here # macOS/Linux export OPENAI_API_KEYyour-api-key-here更推荐的做法是使用.env文件管理在项目根目录创建.env文件。在文件中写入OPENAI_API_KEYyour-api-key-here。安装python-dotenv包pip install python-dotenv。在代码开头加载环境变量。# 示例使用dotenv加载密钥 from dotenv import load_dotenv import os load_dotenv() # 加载 .env 文件中的环境变量 api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. LangGraph核心语法与组件拆解3.1 定义状态State状态是LangGraph工作的核心。我们使用TypedDict来定义状态的“形状”。from typing import TypedDict, List, Annotated import operator from typing_extensions import TypedDict # 定义一个简单的对话状态 class AgentState(TypedDict): # 用户输入的问题 input: str # 对话历史 chat_history: List[str] # Agent生成的回复 response: str # 记录Agent调用过哪些工具 tool_calls: List[str]Annotated用于更复杂的场景例如定义状态的聚合方式operator.add表示列表合并。3.2 创建节点Node节点是一个普通的Python函数或可调用对象它接收一个状态字典并返回一个包含要更新字段的字典。def call_llm(state: AgentState): 一个调用LLM生成回复的节点 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage llm ChatOpenAI(modelgpt-3.5-turbo) # 构建消息历史 messages [ SystemMessage(content你是一个乐于助人的助手。), HumanMessage(contentstate[input]) ] # 调用模型 ai_message llm.invoke(messages) # 返回要更新的状态部分 return { response: ai_message.content, chat_history: state[chat_history] [fUser: {state[input]}, fAssistant: {ai_message.content}] }3.3 创建图Graph并添加节点StateGraph是构建工作流的主要类。from langgraph.graph import StateGraph, END # 1. 创建图并指定状态的结构 workflow StateGraph(AgentState) # 2. 将函数添加为节点 workflow.add_node(llm_node, call_llm) # 可以添加更多节点... # workflow.add_node(tool_node, call_tool) # 3. 设置入口点工作流从哪个节点开始 workflow.set_entry_point(llm_node) # 4. 添加边指定节点执行后的下一个节点 workflow.add_edge(llm_node, END) # 执行完llm_node后结束工作流 # 5. 编译图得到一个可执行的对象 app workflow.compile()3.4 执行图编译后的app就是一个可调用的应用程序其invoke方法接收初始状态。# 定义初始状态 initial_state: AgentState { input: 你好介绍一下LangGraph。, chat_history: [], response: , tool_calls: [] } # 执行图 final_state app.invoke(initial_state) print(final_state[response])4. 实战案例一构建基础对话Agent让我们构建一个简单的、带有工具调用能力的对话Agent。这个Agent可以根据用户问题决定是直接回答还是调用一个计算器工具。4.1 定义工具首先我们定义一个简单的计算器工具。from langchain.tools import tool tool def calculator(expression: str) - str: 计算一个数学表达式。支持 , -, *, / 和括号。 try: # 警告在生产环境中使用eval有安全风险此处仅作演示。 # 应考虑使用ast.literal_eval或专用数学库。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e}4.2 定义更复杂的状态和节点我们需要一个能决定是否调用工具的“路由”逻辑。from typing import Literal from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI class RouterState(TypedDict): messages: Annotated[List, operator.add] # 使用注解实现消息列表的自动合并 next_step: Literal[respond, call_tool] # 用于决定下一步 # 创建LLM并绑定我们定义的工具 llm_with_tools ChatOpenAI(modelgpt-3.5-turbo).bind_tools([calculator]) def router_node(state: RouterState): 路由节点分析最新用户消息决定下一步是直接回复还是调用工具。 messages state[messages] last_message messages[-1] # 调用LLM并指示它可以使用工具 ai_msg llm_with_tools.invoke([SystemMessage(content根据用户问题决定是否需要计算。如果需要请调用计算器工具。), last_message]) # 检查LLM是否决定调用工具 if ai_msg.tool_calls: # 有工具调用下一步进入工具节点 return {messages: [ai_msg], next_step: call_tool} else: # 没有工具调用LLM已生成回复下一步结束 return {messages: [ai_msg], next_step: respond} def tool_node(state: RouterState): 工具节点执行LLM请求的工具调用。 messages state[messages] last_message messages[-1] # 这是一个AIMessage包含了tool_calls results [] for tool_call in last_message.tool_calls: # 根据工具名找到对应的函数 if tool_call[name] calculator: tool_result calculator.invoke(tool_call[args]) results.append(ToolMessage(contenttool_result, tool_call_idtool_call[id])) # 将工具执行结果作为消息添加到历史中 return {messages: results} def response_node(state: RouterState): 响应节点最终将LLM的回复返回给用户。 messages state[messages] last_message messages[-1] # 对于简单情况最后一个AIMessage的内容就是回复 response_text last_message.content if hasattr(last_message, content) else str(last_message) print(fAgent回复: {response_text}) return {messages: []} # 清空或维持状态取决于你的设计4.3 构建并运行带有条件边的图关键点在于使用add_conditional_edges来实现路由。from langgraph.graph import StateGraph, END # 构建图 workflow StateGraph(RouterState) # 添加节点 workflow.add_node(router, router_node) workflow.add_node(tool, tool_node) workflow.add_node(response, response_node) # 设置入口点 workflow.set_entry_point(router) # 添加条件边router节点之后根据state中的next_step值决定去向 workflow.add_conditional_edges( router, # 这是一个路由函数根据状态返回下一个节点的名称 lambda state: state[next_step], { call_tool: tool, # 如果next_step是call_tool去tool节点 respond: response # 如果next_step是respond去response节点 } ) # 添加普通边 workflow.add_edge(tool, router) # 执行完工具后返回router节点重新决策 workflow.add_edge(response, END) # 生成响应后结束流程 # 编译 app workflow.compile() # 运行Agent initial_state RouterState(messages[HumanMessage(content123乘以456等于多少)], next_step) final_state app.invoke(initial_state)这个Agent会分析问题“123乘以456等于多少”LLM会决定调用计算器工具执行计算后将结果返回。你可以尝试问“今天天气怎么样”LLM会判断无需调用工具直接生成回复。5. 实战案例二实现具有长期记忆的对话助手LangGraph的检查点Checkpoint系统可以轻松实现对话记忆的持久化。5.1 使用内存持久化我们需要一个MemorySaver来存储检查点。from langgraph.checkpoint.memory import MemorySaver # 创建内存检查点存储器 memory MemorySaver() # 在编译图时传入检查点管理器 app workflow.compile(checkpointermemory)5.2 配置线程与对话现在每次调用都可以关联一个唯一的thread_id从而隔离不同对话的上下文。# 定义配置包含线程ID from langgraph.graph import MessagesState config {configurable: {thread_id: user_123_session_1}} # 带有配置的调用 initial_state MessagesState(messages[HumanMessage(content我叫小明。)]) result app.invoke(initial_state, configconfig) print(result.messages[-1].content) # 输出: 你好小明 # 在同一个线程中继续对话 next_state MessagesState(messages[HumanMessage(content我刚才说我叫什么名字)]) result2 app.invoke(next_state, configconfig) # 使用相同的thread_id # Agent应该能回答“你叫小明”因为它记住了之前的对话上下文。 print(result2.messages[-1].content)5.3 持久化到数据库进阶MemorySaver仅用于内存重启后数据丢失。生产环境应使用SqliteSaver或自定义存储。# 需要安装 langgraph-checkpoint-sqlite # pip install langgraph-checkpoint-sqlite from langgraph.checkpoint.sqlite import SqliteSaver # 连接到SQLite数据库文件 checkpointer SqliteSaver.from_conn_string(:memory:) # 内存数据库也可用文件路径如 checkpoints.db app workflow.compile(checkpointercheckpointer)使用方式与MemorySaver完全相同。通过更换checkpointer我们就实现了从内存到持久化存储的升级。6. 常见问题与排查思路在开发LangGraph应用时你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案KeyError当访问状态键时状态State的TypedDict定义与实际返回的字典键不匹配。1. 检查StateGraph初始化时传入的State类。2. 确保所有节点函数返回的字典键都存在于State定义中。3. 使用from typing import NotRequired来定义可选键。图编译失败提示节点/边错误节点名称拼写错误边引用了不存在的节点循环依赖。1. 仔细检查add_node和add_edge中的节点名称字符串。2. 使用workflow.get_graph().draw_mermaid()输出图形可视化检查结构。3. 确保没有形成无法到达END的无限循环除非是设计如此。工具调用不生效LLM模型没有正确绑定工具工具定义不符合规范提示词未引导模型使用工具。1. 确认使用llm.bind_tools([tool1, tool2])绑定工具。2. 检查工具函数是否有tool装饰器且文档字符串清晰。3. 在SystemMessage中明确指示模型在合适时使用工具。4. 打印ai_msg.tool_calls查看模型是否返回了工具调用请求。检查点记忆功能无效编译时未传入checkpointer每次调用使用了不同的thread_id。1. 确认编译图时app workflow.compile(checkpointercheckpointer)。2. 调用invoke时确保需要共享上下文的对话使用相同的config配置特别是thread_id。3. 对于SqliteSaver检查数据库文件路径和写入权限。工作流陷入无限循环图中存在未设置终止条件的循环边。1. 分析业务逻辑确保循环有跳出条件例如最大迭代次数、特定状态判断。2. 可以使用条件边在某个节点判断状态并指向END。3. 在节点函数中添加日志打印每次执行的状态帮助调试循环逻辑。执行速度慢节点中的操作如LLM调用、网络请求本身耗时图结构复杂序列执行。1. 优化单个节点的性能如使用更快的模型、缓存。2. 考虑是否可以将无依赖的节点并行化。LangGraph支持通过StateGraph的add_edge和并发语义来设计并行执行流。7. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下实践能大幅提升代码质量和可维护性。7.1 状态设计规范使用Pydantic模型替代TypedDict推荐Pydantic提供运行时数据验证、更丰富的类型提示和序列化支持更适合复杂状态。from pydantic import BaseModel, Field from typing import List class AgentState(BaseModel): input: str chat_history: List[str] Field(default_factorylist) response: str tool_calls: List[dict] Field(default_factorylist)状态扁平化尽量避免在状态中嵌套过深的数据结构这有助于节点的读写清晰度。明确读写范围每个节点只修改状态中它负责的部分返回对应的子字典即可。LangGraph会自动合并。7.2 节点函数设计原则单一职责一个节点只做一件事如调用LLM、执行特定工具、格式化输出。纯函数化节点函数应尽量是纯函数输出只由输入状态决定避免依赖和修改外部全局变量。这使测试和调试更容易。完善的日志记录在节点关键步骤添加日志便于跟踪执行流程和排查问题。import logging logger logging.getLogger(__name__) def my_node(state): logger.info(f进入节点my_node当前输入: {state.get(input)}) # ... 业务逻辑 logger.debug(f节点my_node执行完成返回: {result}) return result7.3 图的构建与可视化模块化构建对于复杂应用将子图构建成函数然后在主图中引用保持代码清晰。def build_decision_subgraph(): subgraph StateGraph(State) # ... 构建子图逻辑 return subgraph main_graph StateGraph(State) decision_app build_decision_subgraph().compile() # 将子图作为一个“超级节点”加入主图 main_graph.add_node(decision_maker, decision_app)利用可视化调试在开发阶段使用graph.get_graph().draw_mermaid()生成Mermaid图表直观检查工作流逻辑是否正确。可以将其粘贴到 Mermaid Live Editor 中查看。7.4 错误处理与健壮性节点级错误处理在节点函数内部使用try-except捕获可能发生的异常如API调用失败并返回一个标识错误的状态而不是让整个图崩溃。def call_external_api(state): try: result risky_api_call(state[query]) return {api_result: result, status: success} except Exception as e: logger.error(fAPI调用失败: {e}) return {api_result: None, status: ferror: {str(e)}}设置全局超时对于可能长时间运行或卡住的图在调用时设置超时。import asyncio # 假设app是异步编译的 try: final_state await asyncio.wait_for(app.ainvoke(initial_state, config), timeout30.0) except asyncio.TimeoutError: # 处理超时逻辑7.5 测试策略单元测试节点单独测试每个节点函数模拟输入状态断言输出状态。集成测试图针对整个编译后的app使用典型的初始状态进行测试验证最终输出是否符合预期。模拟外部依赖在测试时使用unittest.mock来模拟LLM调用、工具执行等外部IO操作使测试快速且稳定。掌握LangGraph意味着你拥有了构建下一代复杂、有状态AI应用的核心能力。从简单的条件路由到具备长期记忆和复杂循环的智能体其基于图的工作流模型提供了无与伦比的清晰度和控制力。建议你从本文的案例出发先尝试修改和扩展其中的功能例如增加更多工具网络搜索、数据库查询、实现更复杂的多Agent协作流程或将其集成到你的Web服务如FastAPI中。随着实践的深入你会越发体会到它将AI应用开发从“脚本编写”提升到“系统设计”层面的强大之处。