1. 从 LangChain 到 LangGraph为什么我们需要一个新的范式如果你和我一样在过去一两年里深度使用过 LangChain 来构建基于大语言模型的应用那你一定经历过那种“甜蜜的烦恼”。LangChain 的链Chain和代理Agent抽象确实极大地简化了早期开发流程让我们能快速地把提示词、工具调用、记忆等组件像搭积木一样组合起来。然而当应用逻辑变得稍微复杂一点——比如需要处理多轮对话中的条件分支、循环执行某个工具直到满足条件、或者在多个“执行者”之间协调工作流时传统的链式结构就开始显得力不从心了。你会发现代码里开始出现大量的if-else判断用来根据上一步的输出决定下一步的走向。或者你试图用AgentExecutor来实现一个循环但对其内部状态的管理和中断机制感到困惑。更复杂一些的比如构建一个模拟软件公司的多智能体系统里面有产品经理、工程师、测试员他们需要按照一定的流程需求评审、开发、测试、发布协作并且流程中可能存在反馈循环测试不通过打回重新开发。用传统的、线性的“链”来描绘这种带有状态、循环和条件分支的图状工作流就像试图用一根绳子来编织一张网既别扭又容易出错。这正是 LangGraph 诞生的背景。它不是要取代 LangChain而是作为 LangChain 生态系统中的一个更高级的、专门用于构建有状态、多步骤、带控制流的工作流Workflow的库。你可以把它理解为 LangChain 的“工作流引擎”或“编排层”。它的核心思想非常直观将应用逻辑建模为一个图Graph。在图中节点Node代表一个可执行的操作单元比如调用一次大模型、运行一个工具、或者执行一段自定义函数。边Edge则定义了节点之间的流转逻辑决定了在一个节点执行完毕后下一步应该去哪个节点。这里的边可以是简单的“无条件跳转”也可以是复杂的“条件判断边”根据当前整个工作流的状态State来决定下一步的路径。这完美契合了复杂业务逻辑中存在的分支、循环、并行等需求。所以HelloLangGraph这个标题对我们这些 LangChain 老用户而言更像是一把打开新世界大门的钥匙。它意味着我们从“链式思维”升级到了“图式思维”从编写线性的执行脚本转变为设计和编排一个灵活的、有状态的工作流。接下来我们就亲手构建第一个 LangGraph感受这种范式转换带来的清晰与强大。2. 核心概念初探State、Node、Edge 与 Graph在开始写代码之前我们必须先吃透 LangGraph 的几个核心抽象。理解它们就等于理解了 LangGraph 的“语法”。2.1 状态State工作流的记忆中枢在 LangGraph 中State是整个工作流运行时唯一的核心数据容器。它是一个字典Dict或类TypedDict定义了工作流中所有需要被记录和传递的变量。为什么是“唯一”的这很重要。在传统的链式编程中数据可能分散在各个回调函数或中间变量里。而在 LangGraph 的图模型中所有节点都读取和修改同一个 State 对象。这保证了数据流的高度一致性和可追溯性。State 的架构通常有两种方式Annotationed方式推荐这是 LangGraph 最新、最强大的模式。你定义一个类并使用pydantic的Field和 LangGraph 的annotation来声明状态字段及其更新规则。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END import operator from langchain_core.messages import HumanMessage, AIMessage from typing_extensions import TypedDict # 1. 定义状态结构 class AgentState(TypedDict): # messages 字段存储对话消息列表更新规则是“追加”operator.add messages: Annotated[list, operator.add] # count 字段存储一个计数器更新规则是“替换”直接赋值 count: int这里的Annotated[list, operator.add]是关键。它声明messages字段是一个列表并且当多个节点修改它时采用operator.add即列表的操作来合并更新。这意味着每个节点都可以向messages列表末尾追加新消息而不会覆盖之前的。count字段没有特殊注解默认采用“替换”策略后一个节点的赋值会覆盖前一个。字典方式传统直接使用一个字典类型来定义 State然后在节点函数中手动处理字段的读取和更新。这种方式更灵活但需要开发者自己管理更新冲突对于初学者更容易出错。我的经验是对于绝大多数应用尤其是涉及多轮对话或累积数据的场景强烈建议使用Annotationed方式。它通过声明式的更新规则如operator.add,operator.or_用于集合合并等自动且安全地处理了并发或顺序修改下的状态合并问题这是手动管理很难做到完美的。2.2 节点Node具体的工作单元Node就是一个普通的 Python 函数或可调用对象它接收当前的State作为参数并返回一个包含对State修改内容的字典。# 2. 定义节点函数 def call_model(state: AgentState): 节点调用大模型生成回复 print(f[Node - call_model] 被调用当前计数{state[count]}) # 从状态中获取最新的用户消息 last_message state[“messages”][-1] # 模拟大模型调用生成一个回复 # 这里为了简单我们直接返回一个固定回复。实际中会调用 ChatModel。 ai_response f”我已经收到了你的第{state[‘count’]}条消息{last_message.content}。我正在处理。” new_ai_message AIMessage(contentai_response) # 返回要更新到状态中的内容 # 根据 State 定义messages 字段会执行追加操作 return {“messages”: [new_ai_message]} def increment_counter(state: AgentState): 节点增加计数器 print(f”[Node - increment_counter] 被调用当前计数{state[‘count’]}“) # 返回要更新到状态中的内容 # count 字段会执行替换操作 return {”count“: state[‘count’] 1}关键点在于节点函数不直接修改传入的state对象而是返回一个“更新字典”。LangGraph 的运行时会根据 State 定义中声明的规则将这个字典合并到全局状态中。这符合函数式编程的理念使得每个节点都是无副作用的纯函数逻辑上更容易测试和推理。2.3 边Edge控制流的导航员Edge决定了执行完一个节点后接下来该去哪里。这是 LangGraph 实现条件分支和循环的魔法所在。边分为两种普通边固定转移无条件地指向下一个节点。条件边Conditional Edge根据 State 的内容动态决定下一个节点。这通过一个返回字符串下一个节点名的函数来实现。def should_continue(state: AgentState) - str: 条件边函数根据计数决定流程走向 # 如果计数小于3继续循环。否则结束。 if state[“count”] 3: return “continue_to_increment” else: return “end”这个函数检查state[‘count’]如果小于3就返回字符串”continue_to_increment”这是我们接下来要定义的一个节点名否则返回”end”LangGraph 预定义的结束标志。2.4 图Graph将一切组装起来最后我们用StateGraph这个类把 State、Node 和 Edge 组装成一个完整的工作流。# 3. 创建图构建器并传入我们定义的状态结构 workflow StateGraph(AgentState) # 4. 添加节点 workflow.add_node(“call_model”, call_model) # 节点名节点函数 workflow.add_node(“increment_counter”, increment_counter) # 5. 设置入口点从 START 到 “call_model” 节点 workflow.add_edge(START, “call_model”) # 6. 添加固定边从 “call_model” 到 “increment_counter” workflow.add_edge(“call_model”, “increment_counter”) # 7. 添加条件边从 “increment_counter” 出来由 should_continue 函数决定去向 workflow.add_conditional_edges( “increment_counter”, # 源节点 should_continue, # 条件判断函数 { “continue_to_increment”: “call_model”, # 如果返回”continue_to_increment”则跳转到”call_model” “end”: END # 如果返回”end”则工作流终止 } ) # 8. 编译图得到可执行对象 app workflow.compile()至此我们定义了一个简单的循环图START - call_model - increment_counter - (判断count) - 如果count3则回到call_model否则结束。3. 运行与调试可视化你的第一个工作流图编译好后就可以运行了。运行需要提供一个初始状态。# 9. 定义初始状态 initial_state: AgentState { “messages”: [HumanMessage(content”Hello, LangGraph!”)], “count”: 0 } # 10. 运行图 final_state app.invoke(initial_state) print(“\n 工作流执行完成 ) print(f”最终状态中的消息数{len(final_state[‘messages’])}“) print(f”最终计数器值{final_state[‘count’]}“) for msg in final_state[“messages”]: print(f” - {msg.type}: {msg.content}“)执行上述代码你会在控制台看到节点依次被调用的打印信息并最终输出结果。你会看到消息列表里包含了初始的人类消息和三轮循环中 AI 生成的三条消息计数器最终为 3。但是代码执行只是第一步。LangGraph 最强大的特性之一是其内置的可视化工具。对于理解复杂工作流和调试来说这几乎是不可或缺的。# 11. 可视化图结构 from IPython.display import Image, display try: # 生成图的PNG图片并显示需安装 graphviz display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片至少打印出图的拓扑结构 print(app.get_graph().draw_mermaid())如果你在 Jupyter Notebook 或支持图形显示的环境下get_graph().draw_mermaid_png()会生成一张清晰的流程图。图中矩形框代表节点箭头代表边菱形框代表条件判断。你一眼就能看出整个工作流的全貌它的起点、终点、循环路径和判断条件。提示在实际开发中我养成了一个习惯——每修改一次图结构就立刻可视化一次。这能帮你快速发现边连接错误、节点遗漏或条件逻辑矛盾比单步调试代码高效得多。确保你的环境安装了graphviz库和系统软件如brew install graphviz或apt-get install graphviz。4. 超越“Hello World”一个实用的对话处理流水线理解了基础概念后让我们构建一个更贴近真实场景的例子一个简单的对话处理流水线。它模拟了接收用户输入、进行敏感词检查、调用大模型生成回复、并根据回复情绪决定是否触发人工客服的流程。4.1 定义更复杂的状态from typing import Annotated, List import operator from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field from enum import Enum class Sentiment(str, Enum): POSITIVE “positive” NEUTRAL “neutral” NEGATIVE “negative” ANGRY “angry” # 特别负面的情绪 class DialogueState(TypedDict): # 对话历史 message_history: Annotated[List[BaseMessage], operator.add] # 最近一次用户输入 latest_input: str # 是否包含敏感词 has_sensitive_word: bool # 大模型生成的原始回复 llm_raw_response: str # 对回复的情绪分析结果 response_sentiment: Sentiment # 是否需要转人工 need_human_agent: bool # 给用户的最终回复 final_response: str这个状态结构包含了输入、多个中间处理结果和最终输出。4.2 实现各个功能节点# 节点1输入预处理与敏感词检查 def preprocess_input(state: DialogueState): print(“[节点] 输入预处理与敏感词检查”) user_input state[“latest_input”] # 模拟敏感词库 sensitive_words [“暴力”, “违禁词”, “攻击”] has_sensitive any(word in user_input for word in sensitive_words) # 将用户输入转换为 LangChain 消息格式存入历史 user_message HumanMessage(contentuser_input) updates { “has_sensitive_word”: has_sensitive, “message_history”: [user_message] } if has_sensitive: updates[“final_response”] “您的输入包含不合适内容请重新表述。” updates[“need_human_agent”] True # 敏感词直接转人工 return updates # 节点2调用大模型生成回复仅在无敏感词时执行 def call_llm_for_response(state: DialogueState): if state[“has_sensitive_word”]: # 如果有敏感词跳过此节点 return {} print(“[节点] 调用大模型生成回复”) # 模拟调用大模型这里用简单逻辑代替 history_text “\n”.join([f”{m.type}: {m.content}” for m in state[“message_history”]]) # 模拟一个简单的回复生成 simulated_response f”基于对话历史{history_text[-100:]}...\n我理解您的意思这是一个模拟的友好回复。” return {“llm_raw_response”: simulated_response} # 节点3分析回复情绪 def analyze_sentiment(state: DialogueState): if not state[“llm_raw_response”]: # 如果上一步没生成回复跳过 return {} print(“[节点] 分析回复情绪”) response state[“llm_raw_response”] # 简单的关键词情绪分析实际应用会用更复杂的模型 sentiment Sentiment.NEUTRAL if “开心” in response or “感谢” in response: sentiment Sentiment.POSITIVE elif “抱歉” in response or “遗憾” in response: sentiment Sentiment.NEGATIVE elif “愤怒” in response or “立即” in response: sentiment Sentiment.ANGRY return {“response_sentiment”: sentiment} # 节点4决策与最终回复生成 def make_decision_and_respond(state: DialogueState): print(“[节点] 决策与生成最终回复”) final_response “” need_human state[“need_human_agent”] if state[“has_sensitive_word”]: final_response state[“final_response”] # 沿用预处理节点设置的回复 elif state[“response_sentiment”] Sentiment.ANGRY: final_response “检测到您可能非常不满即将为您转接人工客服。” need_human True elif state[“response_sentiment”] Sentiment.NEGATIVE: final_response state[“llm_raw_response”] “\n检测到负面情绪如需进一步帮助请告知。” else: final_response state[“llm_raw_response”] # 将AI回复也存入历史 if final_response and not state[“has_sensitive_word”]: ai_message AIMessage(contentfinal_response) return { “final_response”: final_response, “need_human_agent”: need_human, “message_history”: [ai_message] } else: return {“final_response”: final_response, “need_human_agent”: need_human}4.3 编排带有条件分支的工作流现在我们用边来连接这些节点实现一个非线性的工作流。# 构建图 workflow StateGraph(DialogueState) # 添加节点 workflow.add_node(“preprocess”, preprocess_input) workflow.add_node(“call_llm”, call_llm_for_response) workflow.add_node(“analyze”, analyze_sentiment) workflow.add_node(“decide”, make_decision_and_respond) # 设置入口 workflow.add_edge(START, “preprocess”) # 从预处理节点出来根据是否有敏感词决定路径 def route_after_preprocess(state: DialogueState) - str: if state[“has_sensitive_word”]: return “to_decision” # 有敏感词直接跳转到决策节点 else: return “to_llm” # 无敏感词去调用大模型 workflow.add_conditional_edges( “preprocess”, route_after_preprocess, { “to_llm”: “call_llm”, “to_decision”: “decide” } ) # 无敏感词路径LLM - 情绪分析 - 决策 workflow.add_edge(“call_llm”, “analyze”) workflow.add_edge(“analyze”, “decide”) # 决策节点是终点 workflow.add_edge(“decide”, END) # 编译 dialogue_app workflow.compile()这个图清晰地表达了业务逻辑无论什么输入先做预处理和敏感词检查。分支点如果发现敏感词短路跳过 LLM 和情绪分析直接进入决策节点给出警告并标记转人工。如果没有敏感词则正常走LLM生成 - 情绪分析 - 决策的路径。在决策节点中情绪分析的结果如“愤怒”会再次影响是否转人工的决策。你可以用不同的输入来测试这个工作流test_cases [ “今天天气真好” “我对你们的服务感到非常愤怒” “这里有个暴力内容需要处理” ] for input_text in test_cases: print(f”\n 测试输入{input_text} “) init_state {“latest_input”: input_text} # 注意其他状态字段如未提供默认为None或空需要在节点函数中做防御性判断。 result dialogue_app.invoke(init_state) print(f”最终回复{result.get(‘final_response’, ‘N/A’)}“) print(f”是否需要人工{result.get(‘need_human_agent’, False)}“)通过这个例子你应该能感受到 LangGraph 如何将复杂的、带有分支的业务逻辑清晰地转化为一张可视化的、易于理解和维护的图。这比用一长串嵌套的if-else语句要优雅和健壮得多。5. 深入实践错误处理、持久化与并发思考一个健壮的生产级应用绝不能止步于功能实现。LangGraph 提供了一些高级特性来处理更复杂的需求。5.1 错误处理与超时在节点函数中可能会发生网络错误、模型调用失败、业务逻辑异常等。LangGraph 允许你设置中断Interruption和重试。一种常见模式是使用“监督节点”Supervisor Node或“错误处理边”。你可以定义一个专门的节点handle_error然后在构建图时通过add_edge或条件边在特定节点失败时跳转到这个错误处理节点。def unreliable_llm_call(state: State): import random if random.random() 0.3: # 30%概率模拟失败 raise Exception(“模拟LLM调用失败”) return {“response”: “调用成功”} def handle_llm_error(state: State): return {“response”: “抱歉服务暂时不可用已记录您的问题。”} workflow StateGraph(State) workflow.add_node(“llm”, unreliable_llm_call) workflow.add_node(“error_handler”, handle_llm_error) workflow.add_edge(START, “llm”) # 如何将错误连接到处理节点这需要更精细的“状态”设计和条件判断。 # 一种方法是在unreliable_llm_call中捕获异常并将错误信息写入State。 # 然后通过一个条件边函数读取State中的错误标志决定是去下一个正常节点还是去错误处理节点。更优雅的方式是使用 LangGraph 的prebuilt组件例如ToolNode或与LangChain的Runnable深度集成它们自带重试和错误处理机制。对于超时你可以结合asyncio或在使用invoke时设置超时参数。5.2 状态的持久化与恢复对于长时间运行或需要暂停/恢复的工作流例如一个需要用户多次交互的复杂任务状态持久化至关重要。LangGraph 的State对象通常是可序列化的尤其是使用Annotationed和基础类型时。你可以很容易地将state字典转换为 JSON 保存到数据库或文件import json # 保存状态 def save_checkpoint(state: DialogueState, checkpoint_id: str): # 注意BaseMessage 等复杂对象需要自定义序列化或使用LangChain提供的序列化方法。 # 简单示例将消息内容转为字典列表 serializable_state dict(state) serializable_state[“message_history”] [msg.dict() for msg in state[“message_history”]] with open(f”checkpoint_{checkpoint_id}.json”, “w”) as f: json.dump(serializable_state, f) # 加载状态 def load_checkpoint(checkpoint_id: str) - DialogueState: with open(f”checkpoint_{checkpoint_id}.json”, “r”) as f: data json.load(f) # 需要将字典列表反序列化为 BaseMessage 对象 from langchain_core.messages import message_to_dict, messages_from_dict data[“message_history”] messages_from_dict(data[“message_history”]) return data然后你可以从某个检查点重新invoke应用工作流会从保存的状态继续执行。这对于实现“会话恢复”或“长时间异步任务”非常有用。5.3 并发、异步与多线程LangGraph 本身是框架不强制并发模型。节点函数可以是普通的同步函数也可以是async def定义的异步函数。如果你的节点涉及大量 I/O 操作如网络请求、数据库查询使用异步可以极大提升吞吐量。import asyncio from langchain_openai import ChatOpenAI async def async_call_llm(state: State): model ChatOpenAI(model“gpt-4”, temperature0) # 假设 state[‘query’] 是用户查询 messages [HumanMessage(contentstate[“query”])] response await model.ainvoke(messages) # 使用异步调用 return {“answer”: response.content} # 编译时无需特殊处理LangGraph 能处理异步节点。 # 调用时使用 await app.ainvoke(initial_state)。当构建多智能体系统时多个“执行者”节点理论上可以并行运行。LangGraph 目前主要通过子图Subgraph和多线程/进程来实现某种程度的并行。你可以将可以并行的分支设计成独立的子图然后在主图中通过条件边同时触发它们并利用asyncio.gather在自定义节点中等待所有子图完成。需要注意的是完全的、自动化的并行调度还不是 LangGraph 的核心焦点它更擅长于编排清晰的、有依赖关系的步骤流。6. 避坑指南与最佳实践从我早期使用 LangGraph 的经历来看有几个坑特别容易踩到这里分享给你。坑一状态更新规则混淆这是新手最容易出错的地方。记住Annotationed声明中的更新操作符如operator.add决定了当多个节点返回对同一字段的修改时如何合并。如果你定义了一个字段为Annotated[int, operator.add]那么节点返回{“count”: 1}和另一个节点返回{“count”: 2}最终count会是3累加。如果你想要替换就不要加operator.add。对于列表operator.add是追加这通常是你想要的。但对于一个“当前查询”字符串你很可能需要的是替换而不是追加。坑二条件边函数过于复杂条件边函数should_continue(state)应该只做一件事根据当前状态返回下一个节点的名字字符串。它不应该有副作用不修改状态逻辑也应该尽量简单。如果判断逻辑非常复杂考虑将其拆解或者将部分判断逻辑前移到上一个节点将结果写入 State然后条件边函数只做简单的读取和跳转。坑三忽略了节点的“空返回”节点函数可以返回一个空字典{}这意味着“不修改任何状态”。这在某些条件下跳过节点执行时非常有用如我们上面例子中有敏感词时跳过 LLM 调用。确保你的下游节点能处理上游节点可能没有产出预期字段的情况使用state.get(‘field’, default)来安全访问。最佳实践建议从简单开始逐步迭代先用一个最简单的线性图跑通然后逐步添加分支、循环。每步都进行可视化确保图的结构符合你的预期。为状态和节点起好名字State的字段名、节点的函数名都应该清晰表达其用途。calculate_final_score比process_data要好得多。这在调试复杂图时能节省大量时间。充分利用类型提示为State使用TypedDict或Pydantic BaseModel为节点函数参数和返回值添加类型提示。这不仅能利用编辑器的自动补全和错误检查也让代码更易读、易维护。单元测试节点函数由于节点函数是纯函数给定输入状态产生输出更新它们非常容易进行单元测试。单独测试每个节点确保其逻辑正确再组装成图进行集成测试。文档化你的图对于业务逻辑复杂的图除了代码注释可以考虑用 Mermaid 语法或 draw.io 绘制一份业务流程图作为文档说明各个节点和边代表的业务含义。LangGraph 将复杂的工作流编排从“代码泥潭”中解放出来变成了一种声明式的、可视化的设计。它要求开发者更清晰地思考应用的状态流转和控制逻辑这本身就是一个巨大的进步。从HelloLangGraph这个简单的循环开始你已经掌握了它的核心思想。接下来就是用这张“图”去描绘你心中更复杂的智能应用蓝图了。