1. 从LangChain到LangGraph：为什么我们需要“图”来编排AI应用？

如果你在过去一年里折腾过基于大语言模型的应用开发，大概率听说过LangChain。它像是一套乐高积木，把提示词模板、记忆、工具调用这些组件都给你准备好了，让你能快速搭出一个能聊天的AI应用。但当你真的想做一个稍微复杂点的东西，比如一个能根据用户问题自动决定是查数据库、调用API还是生成代码的智能客服，或者一个多轮对话的决策分析助手时，你可能会发现，用LangChain那一套链式调用，写着写着代码就变成了一团乱麻。状态在各个函数间传来传去，逻辑分支像蜘蛛网一样交织，调试起来让人头皮发麻。

这就是LangGraph要解决的核心问题。它不是来替代LangChain的，而是LangChain生态中的一个专门用于构建有状态、多环节、可循环的复杂AI应用编排框架。你可以把它理解为一个专为AI工作流设计的“可视化编程”引擎，只不过它的底层抽象是“图”。为什么是图？因为现实世界中的复杂任务，尤其是涉及决策、循环和状态维护的对话或业务流程，其本质就是一个由节点和边构成的有向图。每个节点执行一个操作（比如调用LLM、执行工具、查询状态），边则定义了操作的流转逻辑（比如根据LLM的输出决定下一步去哪）。

LangGraph把这种思想变成了代码。它让你能清晰地定义工作流中的每一步，以及步骤之间的流转条件，从而将复杂的、容易出错的流程控制，转化为可维护、可调试的图形化结构。这对于构建智能体、复杂对话系统、审批流程自动化等场景来说，是一个游戏规则的改变者。它解决的痛点正是传统链式结构在复杂逻辑面前的无力感，将开发者的心智负担从“如何用代码控制流程”转移到“如何设计工作流本身”。

2. 核心基石：深入理解LangGraph中的“状态机”思想

要玩转LangGraph，必须吃透它的核心设计模式：状态机。这不是计算机科学课本里那个冷冰冰的概念，而是LangGraph赋予AI应用“记忆”和“逻辑”的活生生骨架。

2.1 什么是LangGraph语境下的状态？

在LangGraph中，状态是一个中心化的、可变的字典对象。它贯穿整个工作流的生命周期，是所有节点共享的“工作记忆”。这个状态字典里可以存放任何东西：用户的输入、LLM的回复、中间计算结果、工具的执行结果、甚至是循环的计数器。

举个例子，你构建一个旅行规划助手，状态里可能包含：

• "messages": 对话历史列表。
• "user_preferences": 从对话中提取的用户偏好（如预算、目的地）。
• "search_results": 调用航班/酒店API返回的原始数据。
• "current_step": 当前进行到的步骤（如“收集信息”、“搜索”、“推荐”）。

关键在于，这个状态是随着工作流执行而不断演化的。每个节点读取状态的某些部分，执行操作，然后更新状态。这种设计使得工作流具备了“记忆”，能够进行多轮交互和基于上下文的决策。

2.2 节点与边：构建工作流的乐高积木

有了状态，就需要定义谁来操作它，以及操作完后该去哪。这就是节点和边。

• 节点：一个可调用的函数。它接收整个状态字典作为输入，执行一些操作（如调用LLM、运行代码），然后返回一个包含对状态更新内容的字典。LangGraph会将这个返回的字典与原有状态进行合并更新。节点通常职责单一，比如call_llm,search_web,validate_input。

• 边：连接节点的规则。它决定了在当前节点执行完毕后，下一个该执行哪个节点。边可以分为两种：

  1. 起始边：定义工作流从哪个节点开始。
  2. 条件边：这是状态机灵活性的关键。它不是一个简单的“跳转到节点A”，而是一个根据当前状态来决定下一跳的函数。例如，在旅行助手决定下一步时，条件边函数会检查状态中的“current_step”值，如果是“collect_info”就跳转到信息收集节点，如果是“recommend”就跳转到推荐生成节点。

2.3 “编译”与执行：从蓝图到运行实例

在LangGraph中，你并不是直接“运行”代码，而是先“编译”一个图。这个过程就像把设计好的电路图（你的节点和边定义）生成为一个可执行的程序。

from langgraph.graph import StateGraph, END # 1. 定义状态结构（通常使用TypedDict或Pydantic Model进行类型提示） from typing import TypedDict, List, Annotated import operator class State(TypedDict): messages: Annotated[List[str], operator.add] # 关键：使用注解定义如何合并该字段 user_query: str answer: str # 2. 定义节点函数 def retrieve_node(state: State): # 模拟检索过程 retrieved_info = f"根据查询‘{state['user_query']}’检索到的信息。" return {"messages": [f"检索节点：{retrieved_info}"], "answer": retrieved_info} def generate_node(state: State): # 基于检索结果生成回答 response = f"综合信息，答案是：{state['answer']}" return {"messages": [f"生成节点：{response}"]} # 3. 构建图 workflow = StateGraph(State) workflow.add_node("retrieve", retrieve_node) workflow.add_node("generate", generate_node) # 4. 设置边 workflow.set_entry_point("retrieve") # 起始边：从retrieve节点开始 workflow.add_edge("retrieve", "generate") # 无条件边：retrieve完后一定去generate workflow.add_edge("generate", END) # 结束边：generate完后工作流结束 # 5. 编译图 app = workflow.compile()

现在，app就是一个编译好的、可执行的工作流对象。你可以通过一个初始状态来运行它：

# 6. 执行图 initial_state = {"messages": [], "user_query": "LangGraph是什么？", "answer": ""} final_state = app.invoke(initial_state) print(final_state["messages"])

这个简单的例子展示了一个线性链：检索 -> 生成。但真正的威力在于引入条件边，构建非线性工作流。

注意：Annotated[List[str], operator.add]这个类型注解是LangGraph处理列表状态合并的秘诀。它告诉框架，当多个节点都返回对“messages”字段的更新时，不要覆盖，而是用operator.add（即列表的+操作）来合并它们。这是实现对话历史累积的关键。

3. 超越链式调用：实战构建一个条件分支工作流

让我们构建一个更贴近现实的例子：一个智能路由助手。它需要根据用户问题的类型，决定调用不同的处理节点。

场景：用户输入一个问题。工作流需要：

1. 判断问题类型（是“技术概念解释”、“代码生成”还是“闲聊”）。
2. 根据类型，路由到不同的专家节点处理。
3. 所有节点处理完后，汇总到一个统一格式化的节点。

3.1 定义状态与节点

首先，我们定义更丰富的状态和几个节点函数：

from typing import Literal from langgraph.graph import StateGraph, END class RouterState(TypedDict): messages: Annotated[List[dict], operator.add] # 存储消息对象 user_input: str problem_type: Literal["concept", "code", "chitchat", None] # 问题类型 concept_answer: str code_answer: str chitchat_answer: str final_output: str # 节点1：分类路由节点 def classify_node(state: RouterState): input_text = state["user_input"].lower() if "什么是" in input_text or "原理" in input_text: p_type = "concept" elif "代码" in input_text or "写一个" in input_text or "实现" in input_text: p_type = "code" else: p_type = "chitchat" return {"problem_type": p_type, "messages": [{"role": "system", "content": f"分类为：{p_type}"}]} # 节点2-4：三个专家处理节点（这里用模拟逻辑） def explain_concept_node(state: RouterState): # 模拟概念解释 answer = f"关于‘{state['user_input']}’的概念解释：这是一个重要的架构模式。" return {"concept_answer": answer, "messages": [{"role": "assistant", "content": f"[概念专家] {answer}"}]} def write_code_node(state: RouterState): # 模拟代码生成 answer = f"# 示例代码\nprint('Hello, {state[\"user_input\"]}')" return {"code_answer": answer, "messages": [{"role": "assistant", "content": f"[代码专家] {answer}"}]} def make_chitchat_node(state: RouterState): # 模拟闲聊 answer = f"您问‘{state['user_input']}’？今天天气真不错！" return {"chitchat_answer": answer, "messages": [{"role": "assistant", "content": f"[闲聊专家] {answer}"}]} # 节点5：汇总节点 def format_output_node(state: RouterState): # 根据类型，选取对应的答案作为最终输出 if state["problem_type"] == "concept": final = state["concept_answer"] elif state["problem_type"] == "code": final = state["code_answer"] else: final = state["chitchat_answer"] return {"final_output": final, "messages": [{"role": "system", "content": f"最终输出已就绪：{final}"}]}

3.2 构建带条件边的图

这是关键步骤。我们需要让classify_node之后，根据state[‘problem_type’]的值，动态决定下一个节点。

# 构建图 workflow = StateGraph(RouterState) # 添加所有节点 workflow.add_node("classify", classify_node) workflow.add_node("explain_concept", explain_concept_node) workflow.add_node("write_code", write_code_node) workflow.add_node("make_chitchat", make_chitchat_node) workflow.add_node("format_output", format_output_node) # 设置入口 workflow.set_entry_point("classify") # 关键：从classify节点出发，定义条件边 def route_after_classify(state: RouterState): # 这个函数返回下一个节点的 *名称* if state["problem_type"] == "concept": return "explain_concept" elif state["problem_type"] == "code": return "write_code" elif state["problem_type"] == "chitchat": return "make_chitchat" else: return END # 如果未分类，直接结束 workflow.add_conditional_edges( "classify", # 源节点 route_after_classify, # 路由函数 # 可选：列出所有可能的目标节点，有助于可视化 ["explain_concept", "write_code", "make_chitchat", END] ) # 将三个专家节点都连接到汇总节点 workflow.add_edge("explain_concept", "format_output") workflow.add_edge("write_code", "format_output") workflow.add_edge("make_chitchat", "format_output") # 汇总节点后结束 workflow.add_edge("format_output", END) # 编译 app = workflow.compile()

3.3 执行与可视化

现在，我们可以运行并观察状态如何流转：

# 执行1：询问概念 state1 = app.invoke({"user_input": "什么是状态机？", "messages": []}) print(f"最终输出: {state1['final_output']}") print(f"消息历史: {state1['messages']}") # 执行2：请求代码 state2 = app.invoke({"user_input": "写一个Python的hello world代码", "messages": []}) print(f"\n最终输出: {state2['final_output']}")

更强大的是，LangGraph内置了可视化功能，让你能直观看到你构建的“状态机”：

from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法显示图片，可以输出文本表示 print(app.get_graph().print_ascii())

这张图会清晰地显示classify节点如何分叉到三个不同的专家节点，然后汇聚到format_output。这种可视化对于理解复杂工作流和调试至关重要。

实操心得：在定义条件边函数route_after_classify时，务必确保它对所有可能的状态分支都有明确的返回值，并且返回值必须是图中已添加的节点名称或END。一个常见的坑是路由逻辑遗漏了某些边界情况，导致运行时抛出KeyError。建议在开发初期，用简单的if-elif-else覆盖所有枚举值，并在最后加一个return END或return “fallback_node”作为兜底。

4. 高级模式：循环、持久化与多智能体协作

当你的应用需要多轮对话、长期记忆或多个AI智能体分工合作时，LangGraph的高级特性就派上用场了。

4.1 实现循环：构建多轮对话智能体

循环是状态机的核心能力。在LangGraph中，通过让边指向之前的节点，可以轻松实现循环。一个典型的模式是“人类在环”：智能体执行 -> 等待用户输入 -> 继续执行。

from typing import Optional class ConversationState(TypedDict): messages: Annotated[List[dict], operator.add] needs_human_input: bool def assistant_node(state: ConversationState): # 模拟助理处理逻辑 last_msg = state["messages"][-1]["content"] if state["messages"] else "" if "帮我订票" in last_msg: response = "我需要您提供出行日期和目的地。" next_action = "need_human" else: response = f"我收到了您的消息：‘{last_msg}’。" next_action = "wait" return { "messages": [{"role": "assistant", "content": response}], "needs_human_input": (next_action == "need_human") } def human_input_node(state: ConversationState): # 在实际应用中，这里会是一个等待外部输入（如API调用、前端事件）的节点 # 此处模拟用户回复 simulated_input = "明天去上海" # 这通常来自外部 return { "messages": [{"role": "user", "content": simulated_input}], "needs_human_input": False } # 构建循环图 workflow = StateGraph(ConversationState) workflow.add_node("assistant", assistant_node) workflow.add_node("human_input", human_input_node) workflow.set_entry_point("assistant") def decide_next(state: ConversationState): if state.get("needs_human_input"): return "human_input" else: # 可以设置一个结束条件，比如对话轮次 if len(state["messages"]) > 10: return END return "assistant" # 循环回助理节点 workflow.add_conditional_edges( "assistant", decide_next, ["human_input", "assistant", END] ) workflow.add_edge("human_input", "assistant") # 人类输入后回到助理 app = workflow.compile() # 运行这个app，它会根据`needs_human_input`标志在assistant和human_input间循环

4.2 状态持久化：让应用记住“上一次”

对于需要长期运行的智能体（如聊天机器人），状态必须能够持久化到数据库，并在下次调用时恢复。LangGraph通过Checkpointer抽象支持这一点。

from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 1. 创建一个SQLite检查点存储（生产环境可用Postgres等） dir = tempfile.mkdtemp() conn_str = f"sqlite:///{os.path.join(dir, 'checkpoints.db')}" checkpointer = SqliteSaver.from_conn_string(conn_str) # 2. 在编译图时传入检查点器 app = workflow.compile(checkpointer=checkpointer) # 3. 运行并保存检查点（通常以线程ID或用户ID作为配置标识） config = {"configurable": {"thread_id": "user_123"}} initial_state = {"messages": [{"role": "user", "content": "你好"}], "needs_human_input": False} # 第一次调用，生成检查点 result1, checkpoint_info1 = app.invoke(initial_state, config=config) print(f"第一次回复: {result1['messages'][-1]['content']}") # 模拟应用重启后，从检查点恢复 # 我们使用相同的config（thread_id）来调用，LangGraph会自动加载上次的状态 new_initial_state = {"messages": [{"role": "user", "content": "我昨天问过什么？"}]} result2, checkpoint_info2 = app.invoke(new_initial_state, config=config) # result2中的messages会包含上一次对话的历史！ print(f"历史消息: {[m['content'] for m in result2['messages']]}")

注意事项：状态持久化是生产级应用的关键。SqliteSaver适合轻量级或演示使用，线上环境强烈建议使用PostgresSaver或RedisSaver这类更健壮的后端。另外，要注意状态字典中存储的数据必须是可序列化的（如基本类型、列表、字典），自定义类对象可能需要特殊处理。

4.3 子图与多智能体协作：模块化复杂系统

对于极其复杂的工作流，你可以将其分解为多个子图，每个子图负责一个特定的子任务，然后通过主图进行编排。这对应着多智能体系统中的“主管-工作者”模式。

from langgraph.graph import StateGraph, START, END # 子图A：研究智能体 class ResearchState(TypedDict): topic: str findings: list def research_subgraph(state: ResearchState): # 模拟研究过程 return {"findings": [f"关于{state['topic']}的发现1", f"关于{state['topic']}的发现2"]} research_graph = StateGraph(ResearchState) research_graph.add_node("research", research_subgraph) research_graph.add_edge(START, "research") research_graph.add_edge("research", END) compiled_research = research_graph.compile() # 子图B：写作智能体 class WritingState(TypedDict): findings: list report: str def write_subgraph(state: WritingState): # 模拟写作过程 report = "报告：\n" + "\n".join(state["findings"]) return {"report": report} writing_graph = StateGraph(WritingState) writing_graph.add_node("write", write_subgraph) writing_graph.add_edge(START, "write") writing_graph.add_edge("write", END) compiled_write = writing_graph.compile() # 主图：协调智能体 class ManagerState(TypedDict): query: str research_results: list final_report: str def manager_node(state: ManagerState): # 主节点调用子图 # 1. 调用研究子图 research_state = compiled_research.invoke({"topic": state["query"]}) # 2. 调用写作子图 write_state = compiled_write.invoke({"findings": research_state["findings"]}) return {"research_results": research_state["findings"], "final_report": write_state["report"]} main_graph = StateGraph(ManagerState) main_graph.add_node("manager", manager_node) main_graph.add_edge(START, "manager") main_graph.add_edge("manager", END) main_app = main_graph.compile() result = main_app.invoke({"query": "LangGraph的优势"}) print(result["final_report"])

这种方式实现了极致的模块化和复用。每个子图可以独立开发、测试和优化，主图则像一个 orchestrator，负责流程控制和数据传递。

5. 避坑指南与最佳实践：来自实战的经验

在项目中大规模使用LangGraph后，我积累了一些关键的经验和教训，能帮你节省大量调试时间。

5.1 状态合并的陷阱与正确姿势

LangGraph默认使用“浅合并”来更新状态。这意味着如果两个节点都修改了状态字典中同一个嵌套对象（比如一个列表里的某个字典元素），可能会发生意外覆盖。

错误示例：

class ProblematicState(TypedDict): data: dict # 这是一个嵌套字典 def node_a(state): return {"data": {"step": "a", "value": 1}} # 设置整个data def node_b(state): # 本意是想在data中添加一个字段，但... new_data = state["data"] new_data["status"] = "processed" return {"data": new_data} # 这实际上返回了整个data，会覆盖node_a的`value`吗？

这里的结果取决于合并顺序，行为不确定。

正确做法：对于嵌套结构的更新，使用operator.add对列表是有效的，但对字典不直接支持。推荐两种策略：

1. 扁平化状态：尽量避免深层嵌套。将data_step,data_value,data_status作为状态的同级键。
2. 使用Pydantic模型与langgraph的add_messages范式：对于消息列表，LangGraph有内置最佳实践。对于其他复杂结构，可以在节点函数中返回一个只包含增量更新的字典，并确保你的合并逻辑能处理。

from typing import List from pydantic import BaseModel from langgraph.graph.message import add_messages class ProperState(BaseModel): messages: List[dict] = [] metadata: dict = {} # 复杂对象 def safe_node(state: ProperState): # 更新messages的标准方式 new_messages = add_messages(state.messages, [{"role": "user", "content": "hi"}]) # 更新metadata的增量方式 new_metadata = {**state.metadata, "last_node": "safe_node"} return {"messages": new_messages, "metadata": new_metadata}

5.2 调试：如何追踪工作流的每一步

当工作流没有按预期运行时，调试可能很困难。以下是几种有效方法：

1. 可视化图形：首先，用app.get_graph().draw_mermaid_png()或.print_ascii()检查你的图结构是否正确，边是否连接无误。
2. 启用详细日志：在invoke时设置debug=True。

   result = app.invoke(initial_state, config={"debug": True})

   这会在返回结果中包含每个节点的执行输入和输出，对于追踪状态变化非常有用。
3. 使用检查点：即使不需要持久化，也可以使用内存检查点器来记录执行历史。

   from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = workflow.compile(checkpointer=checkpointer) result, checkpoint = app.invoke(..., config=config) # 可以通过checkpointer查看历史

4. 单元测试单个节点：将节点函数当作纯函数进行测试，传入模拟状态，断言其输出。

5.3 性能考量与生产部署

• 冷启动延迟：编译图（workflow.compile()）有一定开销。在生产环境中，应在服务启动时预编译好所有工作流，而不是每次请求都编译。
• 状态大小：持久化的状态会存储在数据库中。要避免在状态中存储过大的对象（如原始文件内容、大型数据集）。只存储必要的引用或摘要。
• 并发与锁：当多个请求使用相同的thread_id并发访问时，需要数据库检查点器支持行级锁或乐观锁，以防止状态损坏。SqliteSaver在并发写入时可能有问题，生产环境务必使用PostgresSaver。
• 错误处理：在图定义中，目前没有内置的全局错误处理节点。一个实用的模式是定义一个error_handler节点，并在可能出错的节点后通过条件边连接过去。在error_handler节点中，你可以记录错误、更新状态以提示用户，并决定是重试、跳过还是结束工作流。

5.4 与LangChain的整合：强强联合

LangGraph和LangChain是绝配。你可以轻松地将LangChain的LCEL链、工具、记忆组件作为LangGraph的一个节点。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.prebuilt import ToolExecutor, tools_condition from langgraph.graph import MessagesState # 创建一个LangChain链 prompt = ChatPromptTemplate.from_template("回答关于{topic}的问题。") llm = ChatOpenAI(model="gpt-4") chain = prompt | llm | StrOutputParser() # 将链包装成LangGraph节点 def langchain_node(state: MessagesState): # 从状态中提取输入 last_message = state["messages"][-1].content if state["messages"] else "" # 调用链 response = chain.invoke({"topic": last_message}) # 更新状态 return {"messages": [{"role": "assistant", "content": response}]} # 同样，可以集成LangChain Tools from langchain_community.tools import DuckDuckGoSearchRun search_tool = DuckDuckGoSearchRun() tool_executor = ToolExecutor([search_tool]) def tool_node(state): # 根据消息决定调用哪个工具... result = tool_executor.invoke({"query": "some query"}) return {"messages": [{"role": "tool", "content": result}]}

这种整合让你既能享受LangChain丰富的生态和组件，又能利用LangGraph强大的流程控制能力。

从我自己的使用体验来看，LangGraph最大的价值在于它提供了一种符合直觉的方式来设计和推理复杂的AI应用逻辑。它将混乱的控制流代码，转化为了清晰的、可视化的图结构。虽然学习曲线比直接写脚本要陡一些，但一旦掌握，在开发复杂、可维护的AI应用时，其带来的效率和可靠性的提升是巨大的。尤其是在需要多轮交互、分支决策和状态保持的场景下，它几乎是目前最优雅的解决方案。开始使用的最佳方式，是从一个你之前用链式调用感觉有点别扭的小项目开始重构，亲自体会一下“图”思维带来的不同。