最近在尝试构建复杂的AI应用时你是否遇到过这样的困境单个大模型调用太简单无法处理多步骤任务而自己编写状态管理和流程控制代码又异常繁琐容易出错这正是LangGraph这类框架要解决的核心痛点。本文将为你系统梳理LangGraph的核心概念、设计哲学与实战应用无论你是想入门AI Agent开发还是希望将现有的大模型能力串联成自动化工作流都能从零开始构建出稳定、可维护的智能应用。1. LangGraphAI Agent工作流编排的核心引擎在深入代码之前我们首先要理解LangGraph究竟是什么以及它为何在当前的AI应用开发中变得如此重要。1.1 什么是LangGraph从LangChain到工作流编排LangGraph是LangChain生态系统中的一个库但它解决的是一个更具体、更工程化的问题复杂、有状态的工作流编排。你可以把它想象成AI应用领域的“Airflow”或“工作流引擎”专门为基于大语言模型LLM的智能体Agent设计。传统的LangChain提供了构建链Chain和代理Agent的基础能力但当任务步骤增多、状态复杂、且需要循环或条件分支时原始的Chain结构就显得力不从心。LangGraph应运而生它引入了“图”Graph的概念将应用逻辑抽象为节点Nodes和边Edges从而能够清晰地定义和控制任务的执行流。核心价值对比LangChain 更侧重于与大模型、工具、记忆模块的“连接”Chaining和基础组装。LangGraph 更侧重于多个步骤或智能体之间的“编排”Orchestration和状态管理。1.2 核心概念State、Node、Edge理解LangGraph的三个核心概念是上手的关键State状态 这是贯穿整个工作流的共享数据容器。它是一个字典dict或Pydantic模型定义了工作流中需要传递和修改的所有信息。例如在一个问答系统中State可能包含question用户问题、context检索到的文档、answer生成的答案等字段。Node节点 节点是工作流中的一个执行单元通常是一个函数。它接收当前的State执行一些操作如调用LLM、运行工具、处理数据然后返回更新后的State。每个节点负责一个明确的子任务。Edge边 边定义了节点之间的流转逻辑。它决定了在一个节点执行完毕后接下来应该执行哪个节点。边可以是条件边根据State中的某个值决定下一步也可以是普通边无条件指向下一个节点。这为实现“循环”和“分支”逻辑提供了可能。一个简单的比喻把AI应用看作一个厨房。State是共享的菜板和食材Node是切菜、炒菜、调味等具体动作Edge则是菜谱上写的“切完菜后如果肉已腌制好就下锅炒否则继续等待”这样的流程指令。LangGraph就是那位确保流程正确执行的“总厨”。1.3 典型应用场景LangGraph非常适合以下场景多步骤推理任务 如先检索、再分析、最后总结的报告生成。具备循环的Agent 如一个持续与用户对话、根据反馈修正答案的聊天助手。多智能体协作 如一个“研究员”Agent负责查找资料一个“写手”Agent负责撰写一个“评审”Agent负责润色。复杂业务流程自动化 结合外部API和工具实现如客户支持、数据审核等流程。2. 环境准备与项目初始化在开始构建第一个图之前我们需要搭建好开发环境。本文示例将使用Python和OpenAI的API但LangGraph的设计是模型无关的你可以轻松替换为其他LLM提供商如Anthropic、本地部署的Ollama模型等。2.1 安装依赖首先创建一个新的项目目录并建议使用虚拟环境如venv或conda。然后安装必要的包# 创建并激活虚拟环境以venv为例 python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install langgraph langchain-openailanggraph: 核心工作流编排库。langchain-openai: LangChain官方维护的OpenAI集成包用于方便地调用GPT系列模型。如果你还需要其他功能如向量数据库检索可以额外安装langchain-community等包。2.2 设置API密钥为了调用OpenAI的模型你需要设置API密钥。切勿将密钥硬编码在代码中提交到版本库。推荐使用环境变量管理。# 在终端中设置环境变量临时 # Windows (cmd): setx OPENAI_API_KEY your-api-key-here # Windows (PowerShell): $env:OPENAI_API_KEYyour-api-key-here # Linux/Mac: export OPENAI_API_KEYyour-api-key-here或者在代码中通过os.environ设置仅用于开发测试import os os.environ[“OPENAI_API_KEY”] “your-api-key-here”2.3 初始化LLM和基础State模型我们从一个最简单的例子开始创建一个能进行对话的链。首先定义我们将要使用的LLM和State的结构。# 文件basic_graph.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 1. 定义State结构 # 使用TypedDict或Pydantic BaseModel来定义State的“模式” class AgentState(TypedDict): # Annotated 用于在LangGraph中声明该字段的缩减操作如何合并多个节点的更新 # operator.add 表示将新的消息追加到列表中 messages: Annotated[list, operator.add] # 2. 初始化LLM llm ChatOpenAI(model“gpt-3.5-turbo”)这里我们定义了一个最简单的AgentState它只包含一个messages字段类型是列表。Annotated[list, operator.add]是一个关键声明它告诉LangGraph当多个节点都更新了messages字段时应该使用operator.add即列表的操作来合并这些更新。这是实现对话记忆的基础。3. 构建你的第一个LangGraph对话链让我们将上面的组件组装起来创建一个有开始和结束的线性对话链。3.1 创建节点函数节点就是一个普通的Python函数它接收一个State字典返回一个包含更新字段的字典。# 继续在 basic_graph.py 中编写 def call_model(state: AgentState): 节点函数调用LLM生成回复 # 从state中获取最新的对话历史 conversation_history state[“messages”] # 调用LLM。这里简单地将整个历史作为上下文。 response llm.invoke(conversation_history) # 返回更新后的state。我们将在messages列表中添加AI的回复。 # LangGraph会自动根据State定义中的operator.add来合并这个更新。 return {“messages”: [response]} def human_input(state: AgentState): 节点函数模拟人类用户输入 # 在实际应用中这里可能连接到一个Web接口或命令行输入。 # 此处我们模拟一个固定问题。 from langchain_core.messages import HumanMessage user_message HumanMessage(content“LangGraph是什么”) return {“messages”: [user_message]}3.2 构建图并设置流程现在我们创建图添加节点并定义它们的执行顺序。# 继续在 basic_graph.py 中编写 # 3. 创建图构建器 workflow StateGraph(AgentState) # 4. 添加节点 # 第一个节点模拟用户输入 workflow.add_node(“human”, human_input) # 第二个节点调用AI模型 workflow.add_node(“assistant”, call_model) # 5. 设置入口点 workflow.set_entry_point(“human”) # 6. 添加边定义流程human - assistant - END workflow.add_edge(“human”, “assistant”) workflow.add_edge(“assistant”, END) # 7. 编译图得到一个可执行的对象 app workflow.compile()3.3 运行并查看结果编译后的app就是一个可以运行的工作流。我们传入一个初始状态来执行它。# 继续在 basic_graph.py 中编写并执行 if __name__ “__main__”: # 定义初始状态通常messages是空的 initial_state {“messages”: []} # 运行图 final_state app.invoke(initial_state) # 查看最终状态 print(“ 最终对话记录 ) for msg in final_state[“messages”]: print(f“{msg.type}: {msg.content}”)运行这个脚本(python basic_graph.py)你将看到类似以下的输出 最终对话记录 human: LangGraph是什么 ai: LangGraph 是 LangChain 框架中的一个库用于构建和管理基于图结构的复杂工作流或代理系统...恭喜你已经创建并运行了第一个LangGraph。虽然它只是一个简单的线性流程但已经包含了状态管理和节点编排的核心思想。4. 进阶实战构建具备循环与条件的AI Agent一个真正的AI Agent往往需要根据模型或工具的输出决定下一步做什么循环或者选择不同的执行路径条件分支。下面我们构建一个更复杂的Agent它能够根据用户问题决定是否需要调用一个“计算器”工具。4.1 定义更丰富的State和工具# 文件agent_with_tools.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool import json # 1. 定义一个计算器工具 tool def calculator(expression: str) - str: “”“计算一个数学表达式。例如’(3 5) * 2‘”“” # 警告在生产环境中直接eval是危险的此处仅为演示。 # 应使用安全的计算库如numexpr或进行严格的输入过滤。 try: result eval(expression) return str(result) except Exception as e: return f“计算错误: {e}” # 将工具绑定到LLM llm_with_tools ChatOpenAI(model“gpt-3.5-turbo”).bind_tools([calculator]) # 2. 定义扩展的State class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话记录 needs_tool: Annotated[bool, operator.add] # 是否需要调用工具实际中可能用更复杂逻辑 # 注意对于布尔值operator.add 可能不是最佳选择这里为简化演示。 # 更常见的做法是在节点函数中直接覆盖该值。4.2 创建决策节点和工具调用节点现在我们需要多个节点来协作路由节点分析用户消息决定是直接回复还是调用工具。直接回复节点生成普通回复。工具调用节点执行工具并处理结果。# 继续在 agent_with_tools.py 中编写 def route_query(state: AgentState): 路由节点决定下一步是直接回答还是调用工具 last_message state[“messages”][-1] # 这里我们做一个简单的关键词判断。在实际应用中你会让LLM来决定。 if isinstance(last_message, HumanMessage): user_input last_message.content.lower() if “” in user_input or “-” in user_input or “*” in user_input or “/” in user_input or “calculate” in user_input: # 决定调用工具 # 首先让LLM以工具调用的格式进行思考 ai_msg_with_tool_call llm_with_tools.invoke(state[“messages”]) # 返回的消息中可能包含工具调用请求 return {“messages”: [ai_msg_with_tool_call]} else: # 决定直接回复 return {“needs_tool”: False} # 如果不是人类消息比如已经是AI的ToolCall消息则继续执行工具调用流程 return {“needs_tool”: True} def call_tool(state: AgentState): 工具调用节点执行AI请求的工具并将结果作为ToolMessage返回 last_message state[“messages”][-1] if not hasattr(last_message, ‘tool_calls’) or not last_message.tool_calls: # 如果没有工具调用直接返回空更新理论上不应走到这里 return {“messages”: []} tool_messages [] for tool_call in last_message.tool_calls: tool_name tool_call[‘name’] tool_args tool_call[‘args’] if tool_name “calculator”: # 执行计算器工具 result calculator.invoke(tool_args[“expression”]) # 构造ToolMessage这是LangChain中工具返回结果的标准格式 tool_msg ToolMessage( contentresult, tool_call_idtool_call[‘id’] ) tool_messages.append(tool_msg) return {“messages”: tool_messages} def generate_response(state: AgentState): 直接回复节点调用LLM生成最终答案 # 如果上一步是工具调用并返回了结果LLM会根据对话历史和工具结果进行总结。 response llm_with_tools.invoke(state[“messages”]) return {“messages”: [response]}4.3 构建带有条件边的图这是LangGraph最强大的部分根据状态值动态决定流程。# 继续在 agent_with_tools.py 中编写 # 3. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“router”, route_query) # 路由决策 workflow.add_node(“tool”, call_tool) # 调用工具 workflow.add_node(“responder”, generate_response) # 生成回复 # 设置入口点 workflow.set_entry_point(“router”) # 4. 添加条件边 # 从router节点出来后根据state中的needs_tool值决定去向 def decide_next_step(state: AgentState): # 这里我们简化逻辑如果最后一条消息是ToolCall就去tool节点否则去responder last_msg state[“messages”][-1] if hasattr(last_msg, ‘tool_calls’) and last_msg.tool_calls: return “tool” else: return “responder” workflow.add_conditional_edges( “router”, decide_next_step, # 这是一个判断函数返回下一个节点的名字 { “tool”: “tool”, # 如果返回”tool”则跳转到”tool”节点 “responder”: “responder” # 如果返回”responder”则跳转到”responder”节点 } ) # 5. 添加普通边 # 工具调用完成后必须回到LLMresponder来总结结果 workflow.add_edge(“tool”, “responder”) # 生成回复后流程结束 workflow.add_edge(“responder”, END) # 编译 app workflow.compile()4.4 运行与测试让我们用两个不同的问题来测试这个Agent。# 继续在 agent_with_tools.py 中编写并执行 if __name__ “__main__”: # 测试案例1需要计算的问题 print(“测试1: 计算问题”) initial_state {“messages”: [HumanMessage(content“请计算 (12 8) * 3 等于多少”)], “needs_tool”: None} final_state app.invoke(initial_state) print(f“最终回答: {final_state[‘messages’][-1].content}”) print(“-” * 50) # 测试案例2普通问题 print(“测试2: 普通问题”) initial_state {“messages”: [HumanMessage(content“你好请介绍一下你自己。”)], “needs_tool”: None} final_state app.invoke(initial_state) print(f“最终回答: {final_state[‘messages’][-1].content}”)运行后你会看到对于计算问题Agent成功调用了计算器工具并给出了正确答案对于普通问题它则直接进行了回复。这便是一个具备基础决策能力的AI Agent。5. 核心机制深度解析State管理与流程控制要熟练运用LangGraph必须理解其内部如何管理状态和控制流程。5.1 State的合并与更新策略在定义State时我们使用了Annotated[type, reducer]。这个reducer归约器决定了当多个节点对同一个字段进行更新时如何合并这些更新。常见的归约器有operator.add: 用于列表将新列表追加到旧列表之后。这是处理对话历史的典型方式。operator.or_: 用于集合set的并集操作。None或operator.setitem: 直接覆盖前一个值。这是大多数标量类型字符串、整数、字典的默认行为。对于字典如果你想合并update而非覆盖需要自定义函数。自定义归约器示例from typing import Any def merge_dicts(old: dict, new: dict) - dict: “”“合并两个字典新的覆盖旧的”“” return {**old, **new} class MyState(TypedDict): config: Annotated[dict, merge_dicts] # 使用自定义合并函数 log: Annotated[list, operator.add] # 日志列表追加 step_count: int # 整数默认覆盖5.2 条件边Conditional Edges与循环条件边是实现复杂逻辑的基石。add_conditional_edges方法接受一个判断函数该函数接收当前的State并返回下一个目标节点的名称字符串。这个返回值必须与提供的映射字典中的某个键匹配。实现循环循环本质上就是让边指回之前的某个节点。例如你可以创建一个“审核”节点如果审核不通过就指回“修改”节点直到审核通过才流向END。def reviewer(state: MyState): # ... 审核逻辑 if quality_score 0.8: return “revise” # 返回”revise”让流程循环 else: return “publish” workflow.add_conditional_edges( “reviewer”, reviewer, { “revise”: “modify_node”, # 指回修改节点形成循环 “publish”: “publish_node” } )5.3 图的持久化与可视化LangGraph提供了将编译好的图序列化的功能便于保存和加载。# 保存图 app.get_graph().to_json(“my_workflow.json”) # 加载图 (需要提前保存好) from langgraph.graph import StateGraph import json with open(“my_workflow.json”, “r”) as f: graph_config json.load(f) reconstructed_graph StateGraph.from_json(graph_config)对于可视化你可以使用graphviz来生成流程图这有助于理解和调试复杂的工作流。6. 工程化最佳实践与常见问题排查将LangGraph用于实际项目时遵循一些最佳实践可以避免很多坑。6.1 State设计原则最小化与清晰化State只应包含工作流真正需要共享和传递的数据。避免将临时变量或中间计算结果塞入State。使用Pydantic进行强类型校验对于复杂State推荐使用pydantic.BaseModel代替TypedDict因为它能提供运行时类型验证和更丰富的字段配置。合理设计归约器仔细思考每个字段的更新策略。列表追加add和字典合并update是最常见的。错误的归约器会导致状态混乱。6.2 节点函数设计指南单一职责每个节点应只做一件事。例如一个节点负责调用LLM另一个节点负责解析LLM的输出并更新特定状态。幂等性与错误处理节点函数应尽可能设计成幂等的多次执行结果相同。内部做好异常捕获避免因单个节点失败导致整个图崩溃。可以考虑将错误信息写入State由后续节点处理。日志与可观测性在关键节点添加日志记录输入、输出和关键决策点这对于调试生产环境问题至关重要。6.3 常见问题与解决方案问题现象可能原因排查思路与解决方案State更新不符合预期归约器reducer配置错误。例如期望列表追加却配置成了覆盖。1. 检查State字段的Annotated声明。2. 确认节点返回的字典键与State字段名匹配。3. 在节点函数开始和结束时打印State进行对比。条件边不生效总是走默认路径判断函数返回的字符串与add_conditional_edges中映射的键不匹配。1. 在判断函数内打印其返回值。2. 确保映射字典的键包含了所有可能的返回值。3. 可以使用add_conditional_edges的default参数设置一个默认目标节点。图编译或运行时报类型错误State的Pydantic模型定义与节点实际返回的数据类型不匹配。1. 检查Pydantic模型的字段类型定义。2. 确保节点函数返回的字典值类型与模型定义一致。3. 利用Pydantic的严格模式进行开发期校验。工作流陷入无限循环条件边的逻辑有误导致在几个节点间来回跳转无法满足结束条件。1. 在State中添加一个iteration_count字段并设置上限。2. 在条件判断函数中加入循环终止逻辑如超过最大步数则强制结束。3. 使用LangGraph的interrupt机制进行手动干预。调用外部API或工具超时网络问题或工具本身响应慢导致整个工作流卡住。1. 在节点函数中为外部调用设置超时timeout。2. 实现重试机制可以使用tenacity等库。3. 考虑将耗时操作异步化。6.4 生产环境部署考量配置管理将LLM API密钥、模型名称、超时时间等配置外置使用环境变量或配置中心管理。并发与性能LangGraph本身是同步的。对于高并发场景可以考虑将每个工作流的执行包装成异步任务放入任务队列如Celery、Dramatiq中处理。状态持久化默认State在内存中。对于长时间运行或需要故障恢复的工作流需要实现自定义的Checkpointer将State保存到数据库如Redis、PostgreSQL中。监控与告警集成APM工具如OpenTelemetry来追踪图的执行耗时、节点成功率。对关键节点的失败设置告警。7. 总结与进阶学习路线通过本文你已经掌握了LangGraph从基础到进阶的核心内容从State、Node、Edge的基本概念到线性对话链的构建再到具备条件判断和工具调用的复杂Agent开发。你理解了状态管理、条件边实现循环与分支的机制并了解了工程化实践中的关键点。下一步学习建议深入官方文档与示例LangGraph的官方文档和GitHub仓库提供了大量高级示例如多智能体协作、支持人类在环Human-in-the-loop的工作流、子图Subgraph等这是最好的学习资料。探索LangGraph Studio这是一个可视化的开发工具可以让你通过界面拖拽来设计和调试图非常适合原型设计和理解复杂流程。集成向量数据库与检索将LangGraph与LangChain的Retrieval模块结合构建能够基于知识库进行问答的RAG检索增强生成Agent。研究StateGraph与MessageGraphMessageGraph是另一种更简单的图类型专为纯消息传递场景设计在某些情况下比StateGraph更易用。实战项目尝试用LangGraph重构一个你现有的、流程较复杂的LangChain应用或者从头设计一个如自动客服、智能内容创作、数据分析报告生成等系统。LangGraph将AI应用的开发从“链式思维”提升到了“流程编排思维”为构建稳定、可靠、复杂的智能系统提供了强大的底层支持。掌握它意味着你能够将大模型的能力更精细、更可控地融入到真实的业务逻辑中去。