LangGraph：构建复杂AI工作流的状态驱动图编程框架

1. 从LangChain到LangGraph为什么我们需要一个新的“图”如果你在过去一年里折腾过大模型应用开发LangChain这个名字大概率不会陌生。它像是一套乐高积木把提示词模板、记忆、工具调用这些组件都给你准备好了让你能相对快速地拼出一个能跑起来的AI应用。但用久了尤其是当你开始构建稍微复杂一点的、带有多步骤决策或循环逻辑的应用时你可能会感到一种掣肘。比如你想做一个客服机器人它需要先理解用户意图然后可能去查知识库再根据查询结果决定是直接回答、反问用户还是转接人工。这种带“状态”和“分支”的流程用LangChain的Chain或Agent来写代码往往会变得有点“拧巴”状态管理分散流程逻辑也不够直观。这就是LangGraph诞生的背景。它不是要取代LangChain而是LangChain生态系统中的一个专门用于构建有状态、多环节工作流的框架。你可以把它理解为LangChain在“编排”维度上的一个强力补充和进化。如果说LangChain提供了丰富的“零件”Models, Prompts, Tools, Memory那么LangGraph则提供了一套强大的“装配蓝图”和“流水线控制系统”专门用来设计那些零件如何协同工作尤其是当工作流中存在循环、条件判断和持久化状态的时候。它的核心思想非常直观用“图”Graph来定义应用逻辑。在LangGraph中节点Node代表一个执行单元比如调用一次LLM、执行一个工具函数边Edge代表执行路径。通过定义节点和边你就能清晰地描绘出应用的完整执行流程图。这带来的最大好处是可维护性和可观测性的大幅提升。当你面对一个复杂的业务逻辑时一张可视化的图远比层层嵌套的if-else代码更容易让人理解。而且由于整个流程被抽象成了图LangGraph可以原生支持一些高级特性比如检查点中断后可从中间状态恢复、并行执行、以及更精细的流程控制。所以简单来说当你需要构建的AI应用不再是简单的“一问一答”而是涉及多轮对话、复杂决策、回溯、或需要严格顺序执行的一系列步骤时LangGraph就是你该认真考虑的工具。它把Agent的概念从“一个黑盒”变成了“一张可设计、可调试的蓝图”。2. LangGraph核心三要素State、Node、Edge的深度解析理解LangGraph最关键的就是吃透它的三个核心概念State状态、Node节点和Edge边。这是构建任何LangGraph应用的基石。2.1 State工作流的“记忆中枢”State是LangGraph中最核心的抽象。它定义了在整个工作流执行过程中需要被传递和修改的所有数据。你可以把它想象成一个共享的、结构化的“白板”或者“上下文字典”每个节点都可以读取它也可以修改它。在代码中State通常用一个Pydantic的BaseModel来定义。这不仅仅是类型提示更是LangGraph进行状态序列化和验证的基础。from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END import operator # 定义一个强类型的State class AgentState(TypedDict): # 用户输入的问题 input: str # 模型生成的思考过程或中间答案 reasoning: str # 最终要返回给用户的答案 answer: str # 记录调用过哪些工具 tools_called: List[str]这里有一个非常重要的设计模式状态是累加式的。在LangGraph中你通常不会直接覆盖整个状态而是声明每个节点会修改状态的哪一部分。这是通过Annotated类型提示和operator模块来实现的。class AgentState(TypedDict): messages: Annotated[List[str], operator.add] # 关键使用operator.add表示追加 latest_response: str上面这个定义中messages字段被标记为Annotated[List[str], operator.add]。这意味着任何节点如果更新messages它提供的值应该是一个列表会被追加到现有的messages列表后面而不是替换它。这对于对话历史记录这类场景至关重要。latest_response字段则没有这种修饰意味着节点会直接覆盖它的值。为什么这么设计这解决了分布式或异步执行时的状态冲突问题。如果多个节点可能并发修改同一列表直接赋值会导致数据丢失。而“追加”是一个相对安全的操作。LangGraph内部利用这种声明来合并来自不同节点的状态更新。2.2 Node执行具体任务的“工作单元”Node就是一个普通的Python函数或可调用对象它接收当前的State作为输入执行一些操作比如调用LLM、查询数据库、运行计算然后返回一个对State的更新。关键点在于Node函数返回的是一个字典这个字典的键对应State中的字段名值是你想要更新到该字段的内容。LangGraph会根据你之前对State字段的定义如是否用operator.add来合并这些更新。def call_llm_node(state: AgentState) - dict: 节点调用大模型生成思考过程 # 从状态中获取用户输入 user_query state[“input”] # 构建提示词这里简化处理 prompt f“请思考以下问题{user_query}。分步骤给出你的推理过程。” # 模拟调用LLM实际中会使用ChatOpenAI等 llm_response “首先我需要理解问题的核心...其次我需要查找相关数据...” # 返回要更新的状态部分 return {“reasoning”: llm_response}这个节点只更新了reasoning字段。其他字段如input,answer保持不变。这种设计使得每个节点的职责非常单一易于测试和复用。2.3 Edge决定流程走向的“路标”Edge定义了执行完一个节点后下一步应该去哪里。它决定了工作流的控制流。Edge分为两种普通边Fixed Edge直接连接两个节点无条件转移。条件边Conditional Edge根据State中的某个条件值决定下一个要执行的节点。这是实现分支和循环的关键。条件边通常连接到一个“路由函数”Router。这个函数检查State并返回下一个节点的名称。def should_use_tool(state: AgentState) - str: 路由函数判断是否需要调用工具 reasoning state.get(“reasoning”, “”) # 一个简单的规则如果思考中提到“查询”或“计算”则去调用工具 if “查询” in reasoning or “计算” in reasoning: return “call_tool_node” # 下一个节点名 else: return “generate_answer_node” # 另一个节点名在构建图时你会这样添加条件边from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“think”, call_llm_node) workflow.add_node(“use_tool”, call_tool_node) workflow.add_node(“answer”, generate_answer_node) # 设置入口 workflow.set_entry_point(“think”) # 添加条件边从“think”节点出来后根据路由函数决定去向 workflow.add_conditional_edges( “think”, should_use_tool, # 路由函数 { “call_tool_node”: “use_tool”, # 如果返回“call_tool_node”则跳转到“use_tool”节点 “generate_answer_node”: “answer” # 如果返回“generate_answer_node”则跳转到“answer”节点 } ) # 添加固定边从“use_tool”节点出来后固定前往“answer”节点 workflow.add_edge(“use_tool”, “answer”) # 添加固定边“answer”节点是终点 workflow.add_edge(“answer”, END)通过State、Node、Edge的组合你就能像搭积木一样构建出任意复杂的、带状态的工作流。图的结构直观地反映了业务逻辑这是传统线性代码难以比拟的优势。3. 构建你的第一个LangGraph智能体代码逐行解读理论说再多不如动手写一个。我们来构建一个简单的“研究助手”智能体。它的流程是接收用户问题 - LLM思考并决定是否需要联网搜索 - 如需搜索则调用工具 - 综合信息生成最终答案。3.1 环境准备与状态定义首先确保安装必要的包。我们使用OpenAI的模型和Tavily搜索工具作为示例。pip install langgraph langchain-openai tavily-python然后定义State。这个智能体需要管理对话消息、最新的LLM回复以及是否使用了工具的标志。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults # 定义状态结构 class ResearchState(TypedDict): # 消息列表记录整个对话。使用operator.add确保消息是追加的。 messages: Annotated[List, operator.add] # 一个标志位记录LLM是否决定要搜索 should_search: bool这里messages字段是关键。在LangChain/LangGraph的常见模式中messages是一个由HumanMessage、AIMessage、ToolMessage等组成的列表完整记录了人机交互和工具调用的历史。operator.add确保了每个节点对消息的补充都不会丢失。3.2 创建工具与模型初始化我们要用到的外部资源。# 初始化LLM。请替换为你的API KEY。 llm ChatOpenAI(model“gpt-4o”, temperature0, api_key“your-key”) # 初始化搜索工具Tavily是一个专为AI优化的搜索API tavily_tool TavilySearchResults(max_results2, tavily_api_key“your-key”)3.3 实现核心节点一个典型的ReAct推理行动风格智能体至少需要两个节点一个负责“思考”决定行动一个负责“行动”执行工具。节点1思考节点agent_node这个节点的职责是分析当前对话历史和状态决定下一步该做什么直接回答还是调用工具。我们通过给LLM绑定工具并开启函数调用来实现。from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.prebuilt import ToolExecutor # 将工具包装成执行器 tool_executor ToolExecutor([tavily_tool]) def agent_node(state: ResearchState): 智能体节点分析消息决定下一步。 它可能直接生成回复也可能请求调用工具。 # 从状态中获取最新的消息列表 messages state[‘messages’] # 最后一条消息应该是用户的提问HumanMessage last_message messages[-1] # 关键步骤将工具绑定到LLM并开启函数调用。 # 这意味着LLM在思考后可能会输出一个符合工具调用格式的响应。 llm_with_tools llm.bind_tools([tavily_tool]) # 调用LLM传入所有历史消息。LLM会基于上下文决定是直接回答还是调用工具。 response llm_with_tools.invoke(messages) # 将LLM的响应消息添加到状态中会被自动追加到messages列表 return {“messages”: [response]}这里发生了什么llm.bind_tools([tavily_tool])告诉LLM它有一个叫tavily_search_results_json的工具可用。当LLM认为需要搜索时它不会输出普通文本而是输出一个特殊的AIMessage其tool_calls属性里包含了调用tavily_search_results_json的请求参数。如果不需要工具它就输出普通的文本回复。节点2工具执行节点tool_node这个节点的职责很简单执行LLM请求调用的工具并将工具返回的结果格式化成一条消息放回对话历史。def tool_node(state: ResearchState): 工具执行节点执行LLM请求的工具调用并将结果返回。 messages state[‘messages’] # 获取最后一条消息它应该是一个包含工具调用请求的AIMessage last_message messages[-1] # 遍历LLM请求的所有工具调用通常一次一个 for tool_call in last_message.tool_calls: # 执行工具。tool_call[‘name’]是工具名tool_call[‘args’]是参数 result tool_executor.invoke(tool_call) # 创建一条ToolMessage包含执行结果。这是LangChain约定的消息类型用于告知LLM工具执行完毕。 tool_message ToolMessage(contentstr(result), tool_call_idtool_call[‘id’]) # 将工具结果消息返回追加到状态中 return {“messages”: [tool_message]}注意ToolMessage的tool_call_id必须与请求中的id对应这样LangGraph才能正确地将结果与请求关联起来。3.4 组装图与条件路由现在我们把节点和边组装起来形成完整的工作流。# 创建状态图 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(“agent”, agent_node) workflow.add_node(“tool”, tool_node) # 设置入口从agent节点开始 workflow.set_entry_point(“agent”) # 定义路由函数判断下一步是去工具节点还是结束 def route_after_agent(state: ResearchState) - str: 在agent节点执行后检查LLM的响应。 如果响应中包含工具调用请求则路由到‘tool’节点。 否则路由到END结束流程。 messages state[‘messages’] last_message messages[-1] # 判断最后一条AIMessage是否包含工具调用 if last_message.tool_calls: return “tool” else: return END # 添加条件边从agent节点出来后根据路由函数决定去向 workflow.add_conditional_edges( “agent”, route_after_agent, {“tool”: “tool”, END: END} # 映射路由函数的返回值 - 下一个节点名 ) # 添加固定边从tool节点执行完毕后必须回到agent节点进行下一步思考 workflow.add_edge(“tool”, “agent”) # 编译图得到一个可执行的对象 app workflow.compile()这个图形成了一个经典的循环agent - (可能) - tool - agent - ...。智能体会持续地“思考-行动-再思考”直到LLM认为不再需要调用工具输出最终答案为止。3.5 运行与可视化现在我们可以运行这个智能体了。# 初始化输入状态 initial_state {“messages”: [HumanMessage(content“2024年巴黎奥运会中国代表团拿了多少枚金牌”)], “should_search”: False} # 运行图 final_state app.invoke(initial_state) # 打印最终的所有消息 for msg in final_state[“messages”]: print(f“{msg.type}: {msg.content}”)运行后你会看到类似这样的消息序列human: 2024年巴黎奥运会中国代表团拿了多少枚金牌ai: 思考后发现需要最新数据决定调用搜索工具实际上是一条包含tool_calls的AIMessagetool: [{title: ..., content: 中国代表团在2024年巴黎奥运会共获得40枚金牌..., ...}]ai: 根据最新搜索结果中国代表团在2024年巴黎奥运会共获得40枚金牌...可视化是LangGraph的一大亮点。你可以轻松地将图导出查看。from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本结构 print(app.get_graph().draw_ascii())这能生成一张清晰的流程图直观展示agent和tool节点之间的循环关系以及条件边如何工作。对于调试复杂流程来说这是无价之宝。4. 高级特性与实战技巧超越基础工作流当你掌握了基础构建方法后LangGraph的一些高级特性能让你的应用更健壮、更强大。4.1 持久化与检查点实现“长期记忆”和断点续跑这是LangGraph区别于简单脚本的核心能力之一。Checkpointer允许你将图执行过程中的任意状态保存下来后续可以从这个检查点重新开始执行。这对于以下场景至关重要长时运行/异步任务任务可以被中断稍后恢复。错误恢复某个步骤失败后可以从上一步检查点重试而不是从头开始。用户会话持久化将多轮对话的状态保存到数据库实现“长期记忆”用户下次回来可以接着聊。LangGraph支持多种存储后端内存、文件系统、SQLite、Redis等。以下是一个使用SQLite的示例from langgraph.checkpoint.sqlite import SqliteSaver # 创建一个SQLite检查点存储器 checkpointer SqliteSaver.from_conn_string(“:memory:”) # 内存数据库也可用文件路径 # 在编译图时传入检查点存储器 app workflow.compile(checkpointercheckpointer) # 使用一个唯一的线程ID来标识这次会话 config {“configurable”: {“thread_id”: “user_123_session_1”}} # 第一次调用状态会被保存 initial_state {“messages”: [HumanMessage(content“你好”)]} result1 app.invoke(initial_state, configconfig) print(“第一次回复:”, result1[“messages”][-1].content) # 模拟一段时间后用户再次提问。我们使用相同的thread_id。 # LangGraph会自动加载上次的检查点状态并在此基础上继续。 new_message {“messages”: [HumanMessage(content“我刚才问的是什么”)]} result2 app.invoke(new_message, configconfig) # 此时result2的messages里包含了历史对话LLM能回答“你刚才说的是‘你好’” print(“第二次回复:”, result2[“messages”][-1].content)通过configurable配置你可以管理无数个独立的会话流。这对于构建生产级的、有状态的聊天应用是基础功能。4.2 子图模块化与层次化设计当你的工作流变得非常复杂时一张大图会难以维护。子图Subgraph允许你将一部分节点和边打包成一个独立的、可复用的单元作为主图的一个节点。这促进了模块化设计。例如你可以把一个“数据验证与清洗”的流程封装成一个子图然后在多个主图中复用。from langgraph.graph import StateGraph # 1. 首先定义一个子图内部流程 def validate_and_clean_subgraph(state: MainState): # 子图内部也可以有自己的状态定义和逻辑 # 这里简化为一个函数实际中可以编译另一个StateGraph raw_data state[“raw_input”] # ... 执行验证和清洗逻辑 ... cleaned_data raw_data.strip().upper() return {“cleaned_data”: cleaned_data} # 2. 在主图中你可以像添加普通节点一样添加这个子图函数 # 但更常见的做法是子图本身也是一个编译好的Graph对象 # workflow.add_node(“data_preprocessor”, validate_and_clean_subgraph)子图在可视化时会被折叠成一个节点点击可以展开查看内部细节极大地提升了复杂流程的可读性。4.3 并行与分支执行LangGraph支持在图中定义并行分支这对于需要同时处理多个独立任务的场景非常有用。通过add_node和add_edge的组合你可以创建出汇聚fork和合并join的节点结构。# 假设我们需要同时调用两个不同的API获取数据 def fetch_from_api_a(state: State): # 调用API A return {“data_a”: “result_a”} def fetch_from_api_b(state: State): # 调用API B return {“data_b”: “result_b”} def merge_results(state: State): # 合并两个API的结果 data_a state.get(“data_a”) data_b state.get(“data_b”) return {“merged”: f“{data_a} {data_b}”} workflow StateGraph(State) workflow.add_node(“fetch_a”, fetch_from_api_a) workflow.add_node(“fetch_b”, fetch_from_api_b) workflow.add_node(“merge”, merge_results) workflow.set_entry_point(“fetch_a”) workflow.set_entry_point(“fetch_b”) # 注意这需要特定的设置来支持多入口通常需要引入一个“开始”节点来分发 # 更标准的做法是使用一个起始节点然后同时指向fetch_a和fetch_b workflow.add_node(“start”, lambda state: {}) workflow.add_edge(“start”, “fetch_a”) workflow.add_edge(“start”, “fetch_b”) workflow.add_edge(“fetch_a”, “merge”) workflow.add_edge(“fetch_b”, “merge”) workflow.set_entry_point(“start”)在这个简化示例中fetch_a和fetch_b会并行执行取决于执行引擎它们都完成后merge节点才会执行。LangGraph的内部调度器会处理这种依赖关系。4.4 中断与暂停compiled_state_graph.stream()的控制在使用app.stream()进行流式交互时你可能会需要中断或暂停流程。例如在聊天中用户可能想中途打断AI的思考。LangGraph的流式接口返回一个异步生成器。控制中断的核心在于外部驱动循环和状态检查。# 流式调用 config {“configurable”: {“thread_id”: “stream_demo”}} inputs {“messages”: [HumanMessage(content“请写一首关于春天的诗。”)]} stream app.stream(inputs, configconfig, stream_mode“values”) for chunk in stream: # chunk 包含状态更新 node_name chunk[“node_name”] # 当前执行的节点名 state_update chunk[“state”] # 状态更新 # 你可以在这里检查状态例如检查用户是否发送了“停止”指令 # if user_sent_stop_signal: # break # 中断循环即停止了流的消费 print(f“节点 [{node_name}] 执行完毕。最新消息: {state_update.get(‘messages’, [])[-1]}”)真正的“暂停/恢复”能力依赖于前面提到的检查点Checkpointer。当你中断流时状态已经通过检查点保存。下次你可以用相同的thread_id重新调用app.invoke()或app.stream()它会从最后一个持久化的检查点继续执行而不是从头开始。要实现用户主动的“暂停”按钮你需要在UI逻辑中捕获中断信号并妥善保存当前的thread_id。5. 常见陷阱与性能优化指南在实际项目中应用LangGraph我踩过不少坑也总结出一些让应用更稳健、更高效的经验。5.1 状态设计陷阱避免过度耦合与数据污染问题把所有数据都塞进一个庞大的State里导致节点间依赖不清晰难以调试。建议遵循“最小权限原则”。每个节点只声明它需要读写的那部分State字段。使用TypedDict和Annotated进行严格类型约束。将关联紧密的数据放在同一个嵌套对象里但需注意序列化/反序列化的成本。# 不推荐扁平化的大状态 class BadState(TypedDict): user_query: str search_results: list analysis: str final_answer: str error_log: list metadata: dict # 推荐分组的状态职责清晰 class GoodState(TypedDict): input: UserInput # 嵌套对象 processing: ProcessingContext output: OutputBundle system: SystemLogs5.2 节点函数设计保持纯净与幂等问题在节点函数内执行有副作用的操作如直接写入数据库、发送邮件导致在重试或调试时产生重复操作。建议节点函数应尽量是“纯函数”或接近纯函数。它的核心逻辑是根据输入State计算输出State更新。如果需要副作用应该将其封装在Tool中由LangGraph的工具调用机制来管理。或者在节点内通过检查State中的标志位如is_processed来避免重复执行。def risky_node(state: State): # 危险直接发送邮件 # send_email(state[“recipient”], state[“content”]) # return {“email_sent”: True} # 安全将发送邮件设计为一个Tool由LLM决定何时调用 # 或者在节点内做幂等检查 if not state.get(“email_sent”, False): # 执行发送... pass return {“email_sent”: True}5.3 循环与超时控制防止无限循环问题在agent - tool - agent的循环中如果LLM逻辑有误可能陷入死循环不断调用工具。解决LangGraph提供了内置的interrupt机制和超时控制。最实用的方法是在路由函数或状态中引入循环计数器。class SafeState(TypedDict): messages: Annotated[List, operator.add] iteration_count: int # 新增迭代计数器 def route_with_limit(state: SafeState) - str: # 检查迭代次数超过5次则强制结束 if state.get(“iteration_count”, 0) 5: return “force_end_node” # 一个专门处理超限的节点 messages state[“messages”] last_message messages[-1] if last_message.tool_calls: return “tool” else: return END # 在每个agent节点中需要更新计数器 def agent_node_with_count(state: SafeState): # ... LLM调用逻辑 ... new_iteration_count state.get(“iteration_count”, 0) 1 return {“messages”: [response], “iteration_count”: new_iteration_count}5.4 性能优化减少LLM调用与上下文长度LangGraph应用的性能瓶颈通常在于LLM调用慢且贵和过长的上下文影响效果和成本。优化策略1条件短路在进入昂贵的LLM节点前先用简单的规则判断是否可以跳过。例如如果用户输入是“谢谢”可以直接路由到结束节点返回固定回复而无需调用LLM。优化策略2消息窗口与摘要对于长对话messages列表会不断增长。每次都将全部历史喂给LLM成本高昂且可能超过上下文窗口。使用“滑动窗口”只保留最近N轮对话。使用“摘要”节点在对话轮次达到一定数量后触发一个子图调用LLM对早期对话历史进行摘要然后用摘要替换掉详细的历史消息。这需要精心设计State结构来维护“原始历史”和“当前上下文”。class StateWithSummary(TypedDict): full_history: Annotated[List, operator.add] # 完整的、不可变的历史记录可用于持久化 context_messages: List # 当前用于LLM上下文的摘要或最近消息 # ... 其他字段优化策略3并行化独立节点如前所述利用LangGraph的图结构将没有依赖关系的节点并行化。例如一个需要用户画像和产品信息的推荐系统可以并行调用用户数据库和产品目录API。5.5 调试与监控利用可视化与日志当流程出错时清晰的日志和可视化图表是救命稻草。结构化日志在每个节点的开始和结束记录日志包含节点名、输入State的快照、输出更新。可以将这些日志发送到ELK或类似系统。善用app.get_graph().draw_mermaid_png()在开发阶段经常输出流程图确保逻辑与设计一致。状态快照在关键节点后将State的内容记录到文件或数据库便于事后分析流程在哪一步出现了数据异常。LangGraph不是一个“开箱即用”的魔法盒而是一套严谨的工程框架。它要求你对应用的状态流有清晰的事前设计。一旦设计得当它带来的可维护性、可观测性和灵活性提升是巨大的。从简单的线性链到复杂的、带循环和分支的智能体工作流它都能优雅地胜任。