1. 从“链”到“图”：为什么我们需要LangGraph？

如果你在过去一年里折腾过基于大语言模型的应用开发，那么“LangChain”这个名字对你来说一定不陌生。它几乎成了LLM应用开发的代名词，通过其强大的“链”（Chain）抽象，把提示词模板、模型调用、工具使用、记忆管理这些零散的组件，像搭积木一样串联起来，形成一个可执行的、有明确方向的工作流。我最初用LangChain做RAG（检索增强生成）和简单的Agent时，感觉非常顺手——你只需要定义好每个环节的输入输出，LangChain就能帮你把数据流安排得明明白白。

但很快，我就遇到了瓶颈。当我试图构建一个稍微复杂一点的智能体，比如一个需要根据用户问题动态决定是查询数据库、调用工具还是进行多轮对话的客服机器人时，传统的“链”就显得力不从心了。链的本质是线性的、确定性的流程，就像一条单行道。而现实中的复杂任务，尤其是涉及决策、循环和状态管理的任务，更像是一个交通网络，有岔路、有环岛、有需要掉头重来的路段。用“链”来模拟这种网络，代码会变得异常臃肿，充斥着大量的if-else判断和手动状态维护，可读性和可维护性急剧下降。

这恰恰是LangGraph诞生的背景。它不是要取代LangChain，而是LangChain生态的自然演进和有力补充。你可以把LangChain看作是提供了丰富的“砖块”（组件）和“水泥”（连接逻辑），而LangGraph则提供了一张“设计蓝图”和“施工框架”，专门用于建造那些结构非线性的、有状态的复杂应用。它引入了“图”（Graph）的计算范式，将应用中的每个步骤定义为节点（Node），步骤之间的流转关系定义为边（Edge），从而能够优雅地处理循环、条件分支和并行执行。OpenAI团队内部就曾分享过，他们利用类似图计算的方法，在5个月内零手写代码产出了百万行级别的复杂系统，这充分说明了图抽象在构建复杂AI工作流时的巨大潜力。

所以，这篇内容不是简单的API罗列，而是想和你深入聊聊，当我们从LangChain的“链式思维”切换到LangGraph的“图式思维”时，我们究竟在解决什么问题？在实际项目中，这种转变会带来哪些设计上的革新和效率上的提升？我会结合自己的踩坑经验，带你理解LangGraph的核心三要素，并手把手解析如何用它构建一个真正具备复杂决策能力的智能体。

2. LangGraph核心三要素：State、Node、Edge深度拆解

理解LangGraph，最关键的是吃透它的三个核心概念：状态（State）、节点（Node）和边（Edge）。这构成了LangGraph应用的骨架。很多初学者觉得LangGraph复杂，往往是因为没有厘清这三者之间的关系。

2.1 State：应用记忆的单一可信源

在LangChain的链中，数据通常以字典（dict）的形式在组件间传递，你可能会在链的某个环节往字典里塞一个临时变量，在另一个环节再取出来。这种方式在简单链中可行，但在复杂图中，数据流向多变，这种“隐式”的状态管理很快就会失控。

LangGraph的State是一个显式的、强类型的状态容器。它定义了在整个图执行过程中，所有节点共享和修改的数据结构。这就像是给整个应用建立了一个唯一的、结构化的“记忆白板”。

如何定义一个State？通常，我们会使用TypedDict或者Pydantic模型来定义。我强烈推荐使用Pydantic，因为它能提供类型验证和更清晰的文档。

from typing import Annotated, List from typing_extensions import TypedDict from pydantic import BaseModel import operator # 方式一：使用TypedDict（LangGraph经典方式） class AgentState(TypedDict): # 用户输入的问题 input: str # 模型生成的思考过程或中间答案 scratchpad: Annotated[List[str], operator.add] # 关键！这表示该字段会被节点追加，而非覆盖 # 最终返回给用户的结果 output: str # 决定下一个节点的路由逻辑 next: str # 方式二：使用Pydantic BaseModel（更现代，推荐） from pydantic import BaseModel, Field from langgraph.graph.message import add_messages class AgentStatePydantic(BaseModel): messages: Annotated[list, add_messages] = Field(default_factory=list) # 专用于消息累加 question: str = None intermediate_steps: List[dict] = Field(default_factory=list) # 记录工具调用结果 final_answer: str = None

这里有一个极易踩坑的关键点：Annotated的使用。在TypedDict中，Annotated[List[str], operator.add]这个注解是LangGraph的魔法所在。它告诉框架，对于scratchpad这个字段，当多个节点都尝试修改它时，应该使用operator.add（即列表的+操作）来合并更新，而不是后一个节点直接覆盖前一个节点的值。这对于记录日志、累积思考过程至关重要。如果这里定义错了，状态更新逻辑就会混乱。

我的实操心得：在项目初期就花时间设计好State的结构。思考清楚哪些数据是贯穿始终的（如对话历史messages），哪些是临时中间产物（如scratchpad），哪些是控制流的关键（如next）。一个好的State设计能让后续的节点和边定义清晰十倍。

2.2 Node：功能单一的原子操作单元

节点是图的基本执行单元。在LangGraph中，一个节点就是一个函数（或可调用对象），它接收当前的State，执行一些操作（如调用LLM、查询工具、处理数据），然后返回一个包含对State更新内容的字典。

节点的黄金法则：单一职责一个节点只做一件事。例如：

• call_model节点：只负责调用大模型并获取回复。
• search_web节点：只负责执行网络搜索。
• evaluate_condition节点：只负责根据当前状态判断下一步去哪。

from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4-turbo-preview") def call_llm_node(state: AgentStatePydantic) -> dict: """节点：调用大模型生成思考或回答""" # 1. 从状态中构建提示词 messages = state.messages # 假设我们最后一条消息是用户问题 user_question = state.question # 2. 可以在此处添加系统提示词或上下文 system_message = {"role": "system", "content": "你是一个有帮助的助手，请逐步思考。"} full_messages = [system_message] + messages[-5:] # 只保留最近5条消息作为上下文 # 3. 调用模型 response = model.invoke(full_messages) # 4. 返回状态更新。注意：我们更新的是`messages`字段，由于定义了add_messages，这会追加。 return {"messages": [response]}

踩坑提醒：节点函数返回的字典，其键必须与State中定义的字段名对应。返回{"new_key": value}是无效的，除非new_key已在State中定义。更新逻辑（覆盖、追加等）则由State定义时的Annotated注解决定。

2.3 Edge：决定流程走向的智能路由

边定义了节点之间的流转逻辑。这是LangGraph最强大也最灵活的部分。边分为两种：

1. 普通边（Fixed Edges）：无条件地从节点A指向节点B。
2. 条件边（Conditional Edges）：根据State中的某个条件值，动态决定下一个节点。

条件边是实现复杂工作流（如循环、分支）的核心。它通常由一个路由函数（routing function）来实现。

from langgraph.graph import END def should_continue(state: AgentStatePydantic) -> str: """路由函数：根据模型回复或状态，决定下一步""" last_message = state.messages[-1] # 场景1：如果模型回复中包含“FINAL_ANSWER”关键词，则结束 if "FINAL_ANSWER" in last_message.content: return "end" # 指向一个名为“end”的节点，或直接返回`END`常量 # 场景2：如果模型回复要求调用工具（例如包含“TOOL:”前缀） elif "TOOL:" in last_message.content: return "call_tool" # 场景3：默认情况，继续让模型思考 else: return "continue_think" # 在构建图时，我们会这样使用条件边 import operator from langgraph.graph import StateGraph, START workflow = StateGraph(AgentStatePydantic) # 添加节点 workflow.add_node("agent", call_llm_node) workflow.add_node("tool", call_tool_node) workflow.add_node("end", lambda state: state) # 结束节点可以什么都不做 # 设置起点 workflow.set_entry_point("agent") # 添加条件边 workflow.add_conditional_edges( "agent", # 源节点 should_continue, # 路由函数 { "continue_think": "agent", # 条件值 -> 下一个节点名 "call_tool": "tool", "end": END # 使用内置的END常量直接结束 } ) # 添加固定边：工具执行完后，永远返回agent节点继续思考 workflow.add_edge("tool", "agent")

这里有一个至关重要的设计模式：agent -> (条件判断) -> tool -> agent构成了一个经典的“思考-行动”循环（ReAct模式）。模型（agent）决定是否需要行动（调用tool），行动的结果被写回State，然后State又被送入模型进行下一轮思考，直到模型认为可以给出最终答案。这个循环用LangChain的链来实现会非常别扭，但用LangGraph的图来描述则无比自然。

3. 实战构建：一个具备自我修正能力的RAG问答图

理解了核心三要素，我们通过一个更贴近实际需求的例子来串联它们：构建一个具备自我修正能力的RAG问答系统。普通RAG是“检索->生成”的单向链，而我们的系统将具备“生成->评估检索结果->必要时重新检索”的循环能力。

3.1 定义复杂状态与节点

我们的State需要记录更多信息。

from pydantic import BaseModel, Field from typing import List, Optional, Literal from langchain_core.messages import BaseMessage class RAGState(BaseModel): """RAG图的状态""" # 原始问题 original_question: str # 当前轮次的问题（可能被重写） current_question: str # 检索到的文档块 retrieved_docs: List[str] = Field(default_factory=list) # 模型生成的答案草稿 draft_answer: Optional[str] = None # 对答案置信度的自我评估（高/中/低） confidence: Optional[Literal["high", "medium", "low"]] = None # 重写后的问题（如果需要重新检索） rewritten_question: Optional[str] = None # 控制流标志 next_step: Literal["retrieve", "generate", "evaluate", "rewrite", "end"] = "retrieve" # 消息历史（用于与模型对话） messages: List[BaseMessage] = Field(default_factory=list)

接下来，我们定义四个核心节点：

# 节点1：检索 def retrieve_node(state: RAGState): """从向量库检索相关文档""" from your_vector_store import retriever # 假设你已初始化检索器 docs = retriever.invoke(state.current_question) return { "retrieved_docs": [doc.page_content for doc in docs], "next_step": "generate" } # 节点2：生成答案 def generate_answer_node(state: RAGState): """基于检索到的文档生成答案""" context = "\n\n".join(state.retrieved_docs[:5]) # 取前5个相关片段 prompt = f""" 基于以下上下文，回答问题。如果上下文不足以回答问题，请如实说明。 上下文：{context} 问题：{state.current_question} 答案： """ # 调用模型 llm = ChatOpenAI(model="gpt-3.5-turbo") answer = llm.invoke(prompt).content return { "draft_answer": answer, "next_step": "evaluate" } # 节点3：自我评估置信度 def evaluate_confidence_node(state: RAGState): """让模型自我评估答案的置信度""" prompt = f""" 你刚刚基于一些参考文档回答了以下问题。 问题：{state.current_question} 你生成的答案：{state.draft_answer} 请严格评估这个答案的质量： 1. 答案是否直接、充分地回答了问题？ 2. 答案是否完全基于提供的参考文档？（指出未基于文档的部分） 3. 如果用户对答案提出质疑，你是否有足够的依据进行辩护？ 请仅输出一个单词：HIGH（高置信度）、MEDIUM（中置信度）或LOW（低置信度）。 """ llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) evaluation = llm.invoke(prompt).content.strip().upper() confidence_map = {"HIGH": "high", "MEDIUM": "medium", "LOW": "low"} confidence = confidence_map.get(evaluation[:4], "medium") # 简单映射 next_step = "end" if confidence == "high" else "rewrite" return { "confidence": confidence, "next_step": next_step } # 节点4：重写问题 def rewrite_question_node(state: RAGState): """当置信度低时，重写问题以更好地检索""" prompt = f""" 原始问题是：“{state.original_question}” 我们之前用“{state.current_question}”进行检索，但生成的答案置信度不高。 请分析可能的原因，并重写一个更利于从知识库中检索到关键信息的问题。 输出格式：重写后的问题：[你的问题] """ llm = ChatOpenAI(model="gpt-4-turbo-preview") response = llm.invoke(prompt).content # 简单提取重写后的问题 import re match = re.search(r"重写后的问题：(.*)", response) new_question = match.group(1).strip() if match else state.current_question + " (详细版)" return { "rewritten_question": new_question, "current_question": new_question, # 更新当前问题 "next_step": "retrieve" # 跳回检索节点，开始新一轮循环 }

3.2 构建并编译图

现在，我们将节点和边组装起来。

from langgraph.graph import StateGraph, START, END # 初始化图 workflow = StateGraph(RAGState) # 添加所有节点 workflow.add_node("retrieve", retrieve_node) workflow.add_node("generate", generate_answer_node) workflow.add_node("evaluate", evaluate_confidence_node) workflow.add_node("rewrite", rewrite_question_node) # 定义结束节点（一个空操作，用于汇聚流程） def end_node(state): return state workflow.add_node("end", end_node) # 设置入口：从检索开始 workflow.set_entry_point("retrieve") # 添加固定边（线性部分） workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", "evaluate") # 添加条件边（实现循环的关键） def route_after_evaluate(state: RAGState) -> str: """根据评估结果路由""" return state.next_step # 直接返回状态中已计算好的下一步标志 workflow.add_conditional_edges( "evaluate", route_after_evaluate, { "rewrite": "rewrite", "end": "end" } ) # 重写问题后，固定跳回检索 workflow.add_edge("rewrite", "retrieve") # 结束节点指向END workflow.add_edge("end", END) # 编译图 app = workflow.compile()

3.3 运行与调试：可视化与流式输出

图编译好后，就可以运行了。LangGraph提供了强大的可视化工具，对于调试复杂流程至关重要。

# 1. 运行图 initial_state = {"original_question": "LangGraph和LangChain的主要区别是什么？", "current_question": "LangGraph和LangChain的主要区别是什么？"} final_state = app.invoke(initial_state) print(final_state["draft_answer"]) print(f"最终置信度：{final_state['confidence']}") # 2. 流式输出（观察执行步骤） for step in app.stream(initial_state, subgraphs=True): # subgraphs=True可展示子图细节 node_name = list(step.keys())[0] print(f"--- 执行节点: {node_name} ---") print(f"状态更新: {step[node_name]}") print() # 3. 可视化（需要安装pygraphviz） try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: # 或者生成Mermaid文本，粘贴到Mermaid Live Editor查看 print(app.get_graph().draw_mermaid())

踩坑实录：compiled_state_graph.stream()如何终止？这是一个高频问题。在上面的evaluate_confidence_node中，我们通过将next_step设置为"end"来引导流程走向结束节点。结束节点再通过workflow.add_edge("end", END)连接到LangGraph内置的END符号，从而优雅地终止整个图的执行。关键在于，终止的逻辑应该由你的节点逻辑和路由函数来控制，而不是从外部强行打断stream。如果你需要实现超时或外部中断，可能需要结合异步任务和信号机制，这超出了LangGraph图本身的范围。

4. 进阶技巧：子图、长期记忆与生产级考量

当应用变得极其复杂时，将所有逻辑塞进一个大图里会难以管理。这时就需要用到子图（Subgraph）。子图允许你将一个功能模块（例如，完整的问答流程）封装起来，作为一个“超级节点”嵌入到主图中。这极大地提升了代码的模块化和复用性。

4.1 使用子图封装复杂逻辑

假设我们上面构建的自我修正RAG流程已经很稳定，我们想把它作为一个组件，用在更大的客服工作流中。

from langgraph.graph import StateGraph as BaseStateGraph from langgraph.graph import START as BASE_START from langgraph.graph import END as BASE_END # 1. 首先，将之前的RAG图定义为一个可编译的子图函数 def create_rag_subgraph(): rag_workflow = BaseStateGraph(RAGState) # ... (重复上面第3节的添加节点和边的代码) rag_workflow.add_edge("end", BASE_END) return rag_workflow.compile() # 2. 在主图中，我们将子图作为一个节点 from langgraph.graph import StateGraph, SubgraphState # 主图的状态可能更简单 class MainAppState(BaseModel): user_query: str conversation_history: List[dict] = Field(default_factory=list) rag_result: Optional[str] = None # 使用SubgraphState来“包裹”子图的状态 rag_subgraph_state: Optional[SubgraphState] = None # 创建主图 main_workflow = StateGraph(MainAppState) # 定义一个节点，这个节点的作用是调用RAG子图 def call_rag_subgraph_node(state: MainAppState): # 初始化子图输入状态 rag_input_state = RAGState( original_question=state.user_query, current_question=state.user_query ) # 获取子图app rag_app = create_rag_subgraph() # 运行子图 rag_final_state = rag_app.invoke(rag_input_state) # 将子图的结果和状态带回主图状态 return { "rag_result": rag_final_state["draft_answer"], # 保存子图的完整状态，便于追踪或后续使用 "rag_subgraph_state": SubgraphState( values=rag_final_state, # config=... 可以传递配置 ) } main_workflow.add_node("rag_processor", call_rag_subgraph_node) # ... 主图的其他节点和边

4.2 集成长期记忆（Long-term Memory）

LangGraph本身不直接提供开箱即用的长期记忆存储，但它与LangChain的ChatMessageHistory等组件无缝集成。关键在于将记忆作为State的一部分，并在节点中正确地读取和更新它。

更佳实践：使用add_messages注解前面我们在State中已经见过了Annotated[list, add_messages]。这是LangGraph为对话记忆设计的“语法糖”。add_messages是一个归约器（reducer），它知道如何将新旧消息列表智能地合并（例如，合并同角色的连续消息），而不仅仅是追加。

from langgraph.graph.message import add_messages class ConversationState(BaseModel): # 核心：使用add_messages管理对话历史 messages: Annotated[list, add_messages] = Field(default_factory=list) user_input: str def chat_node(state: ConversationState): # 直接使用state.messages，它已经是整理好的对话历史 llm = ChatOpenAI() # 模型调用会自动看到完整的对话上下文 response = llm.invoke(state.messages) # 返回更新，add_messages会处理合并逻辑 return {"messages": [response]}

对于需要持久化的长期记忆（如保存到数据库），你可以在图中添加一个专门的“记忆持久化节点”，在对话轮次结束时或达到某个条件时触发，将state.messages保存起来。下次会话开始时，再从数据库加载并初始化到state.messages中。

4.3 生产环境部署与监控

将LangGraph应用部署到生产环境（如使用FastAPI），你需要考虑以下几点：

1. 线程安全与并发：app.invoke()通常是线程安全的，因为执行过程是纯函数对状态的转换。但要确保你提供的工具（如数据库连接、第三方API客户端）本身是线程安全的或为每个请求创建新实例。
2. 状态序列化：如果你使用Pydantic定义State，它天然支持model_dump()和model_validate()进行JSON序列化与反序列化。这对于将检查点（checkpoint）保存到数据库、实现异步或持续的工作流至关重要。
3. 检查点（Checkpoints）：LangGraph支持检查点机制，允许你在图执行到某个节点后暂停，并将完整状态保存下来，后续可以从此状态恢复执行。这对于处理耗时极长的任务或等待外部人工输入的任务非常有用。

   # 这是一个高级特性，涉及配置和持久化层 from langgraph.checkpoint import MemorySaver checkpoint_memory = MemorySaver() app = workflow.compile(checkpointer=checkpoint_memory) # 第一次执行，生成一个线程ID config = {"configurable": {"thread_id": "user_123_session_1"}} result1 = app.invoke(initial_state, config=config) # 模拟一段时间后，从上次中断的地方继续执行 # LangGraph会基于thread_id自动加载最新的检查点状态 result2 = app.invoke(new_input_state, config=config)

4. 日志与追踪：利用app.stream()输出每一步的状态变化，这是调试的利器。在生产环境中，可以将这些日志结构化后发送到如OpenTelemetry、LangSmith等可观测性平台，方便监控每个节点的耗时、输入输出和异常。

从LangChain到LangGraph，本质上是从“编排组件”到“编排智能”的思维升级。LangChain帮你把工具准备好，而LangGraph帮你设计智能体如何使用这些工具去完成复杂的、非线性的任务。它带来的最大好处是代码即架构——你的应用流程图几乎就是你的代码结构，这使得复杂系统的设计、理解和维护成本大大降低。刚开始接触时，你可能会觉得多了一层抽象有点绕，但一旦适应了这种“图思维”，在构建具备复杂决策和循环能力的AI应用时，你会发现自己再也回不去了。