LangGraph实战：构建有状态AI工作流的核心概念与工程实践

1. 项目概述：为什么是LangGraph？

如果你已经用LangChain构建过一些AI应用，可能会遇到一个瓶颈：当业务流程变得复杂，需要处理多轮对话、条件分支、循环或者需要维护一个长期的状态时，单纯用LangChain的链（Chain）和代理（Agent）来组织代码，会感觉有点“力不从心”。代码里开始出现大量的if-else判断，状态在各个函数之间传来传去，逻辑变得难以追踪和调试。这时候，你就需要一个更强大的工具来清晰地定义和控制这些复杂的、有状态的AI工作流。这就是LangGraph诞生的背景。

简单来说，LangGraph是LangChain生态系统中的一个库，专门用于构建有状态的、多参与者的AI应用。它把整个应用流程抽象成一个“图”（Graph），图中的节点代表一个执行步骤（比如调用一次LLM、执行一个工具、查询数据库），边则代表步骤之间的流转逻辑。这种范式特别适合聊天机器人、多步骤任务规划、模拟仿真等场景。它不是要取代LangChain，而是与LangChain深度集成，用图的思想来管理LangChain的各个组件，让复杂流程的编排变得像画流程图一样直观。

我最初接触LangGraph是因为要做一个智能客服的POC，需要处理用户从咨询、比价、下单到售后的一系列连贯操作。用传统的链式写法，代码很快就成了一团乱麻。切换到LangGraph后，整个业务流程被清晰地定义在了一张图里，状态如何流转、在哪里做决策一目了然，开发和维护效率提升了好几个量级。接下来，我就带你从零开始，拆解LangGraph的核心概念和实战用法。

2. LangGraph核心三要素：State、Node、Edge

要理解LangGraph，必须先吃透它的三个核心概念：状态（State）、节点（Node）和边（Edge）。这是构建任何LangGraph应用的基石。

2.1 状态（State）：应用的记忆中枢

在LangGraph中，State是一个贯穿整个图执行过程的、可变的共享数据容器。你可以把它想象成整个工作流的“记忆白板”或者“全局变量字典”。所有节点都从State中读取输入，并将输出写回State。

State通常用一个Pydantic的BaseModel来定义，这能提供良好的类型提示和验证。例如，一个简单的聊天机器人State可能长这样：

from typing import List, Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages import operator class State(TypedDict): # 存储对话历史 messages: Annotated[List, add_messages] # 用户当前查询 query: str # 从知识库检索到的上下文 context: str # LLM生成的最终答案 answer: str # 一个标志位，用于控制流程跳转 needs_human_review: bool

这里有几个关键点：

1. TypedDictvsBaseModel：官方示例常用TypedDict，因为它更轻量，与Python字典兼容性好。但在复杂场景下，使用Pydantic的BaseModel能获得更强大的数据验证和序列化能力。我个人在需要严格数据格式或与外部API交互时，更倾向于用BaseModel。
2. Annotated类型注解：这是LangGraph的一个精髓。Annotated[List, add_messages]不仅仅声明messages是一个列表，还指定了一个归约器（reducer）add_messages。这意味着当多个节点都想修改messages时（比如用户节点添加用户消息，AI节点添加AI回复），add_messages函数会决定如何合并这些修改（通常是追加到列表末尾）。这是实现状态更新的关键机制。
3. 状态设计原则：State应该包含工作流所需的所有数据。设计时要考虑“数据驱动”，即下一个节点执行什么、怎么执行，往往取决于State中的某个字段的值（例如needs_human_review为True时，流转到人工审核节点）。

实操心得：在设计State初期，很容易把所有想到的字段都塞进去，导致State过于臃肿。我的经验是，遵循“最小化”原则，只存放真正需要在节点间共享和传递的数据。临时计算的结果，如果只在单个节点内使用，最好作为节点函数的局部变量。

2.2 节点（Node）：执行单元

节点是图中的一个执行步骤，本质上是一个函数。这个函数接收当前的State（或其一部分）作为输入，执行一些操作（如调用LLM、运行计算、查询API），然后返回一个对State的更新。

节点的定义非常自由：

def retrieve_node(state: State): """检索节点：根据用户查询，从向量库获取相关上下文""" query = state[“query”] # 假设我们有一个检索函数 retrieved_docs = my_retriever.invoke(query) # 返回一个字典，这个字典会被用来更新State return {“context”: “\n”.join([doc.page_content for doc in retrieved_docs])} def llm_node(state: State): """LLM节点：根据上下文和查询生成回答""" from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_template(“”” 基于以下上下文回答问题： 上下文：{context} 问题：{query} 请给出专业、准确的回答。 “””) chain = prompt | ChatOpenAI(model=“gpt-4”) response = chain.invoke({“context”: state[“context”], “query”: state[“query”]}) return {“answer”: response.content}

关键特性：

• 输入输出：节点函数接收一个State（或它的子集），返回一个字典。返回字典的键值对会被“合并”到全局State中。
• 副作用：节点可以执行任何有副作用的操作，比如写入数据库、发送邮件。
• 纯函数理想：虽然可以有副作用，但尽量将节点设计为“纯函数”，即输出只由输入State决定。这有利于测试、调试和实现节点复用。

2.3 边（Edge）：流程控制器

边决定了执行完一个节点后，下一步该去哪里。这是LangGraph实现条件逻辑、循环和并行执行的核心。边通常由一个条件函数（Conditional Edge）或一个固定映射来定义。

1. 固定边（Fixed Edge）最简单的情况，直接指定下一个节点。

from langgraph.graph import StateGraph builder = StateGraph(State) builder.add_node(“retrieve”, retrieve_node) builder.add_node(“generate”, llm_node) # 添加一条从 “retrieve” 到 “generate” 的边 builder.add_edge(“retrieve”, “generate”)

2. 条件边（Conditional Edge）这是LangGraph最强大的特性之一。根据State的内容，动态决定下一个节点。

def route_after_generate(state: State): """根据生成答案的质量，决定下一步""" answer = state[“answer”] # 假设我们有一个函数判断答案是否需要人工复核 if needs_human_review(answer): return “human_review_node” else: return “end” # 可以指向一个结束节点，或者用”__end__“特殊标识 # 将条件函数添加为边 builder.add_conditional_edges( “generate”, # 起始节点 route_after_generate, # 路由函数 [“human_review_node”, “end”] # 路由函数可能返回的目的地节点列表（供框架验证） )

3. 入口与出口每个图需要指定一个入口节点（set_entry_point）和一个或多个出口。出口通常用字符串常量”__end__“表示。

builder.set_entry_point(“retrieve”) # 在条件边中，可以将 “__end__“ 作为一个目的地，表示流程结束。

注意事项：条件函数route_after_generate必须返回一个字符串，且该字符串必须在提供的可选目的地列表中，或者是”__end__“。否则运行时会出现InvalidUpdateError。在开发时，务必仔细检查所有可能的分支返回值。

3. 构建你的第一个LangGraph应用：智能问答助手

理论讲得再多，不如动手做一个。我们来构建一个增强检索生成（RAG）流程的智能问答助手。这个流程包含：用户输入、检索、生成、安全检查四个节点，并根据安全检查结果决定是直接输出还是进入人工审核。

3.1 定义State与工具准备

首先，定义我们的State。这次我们使用PydanticBaseModel，体验一下更严格的类型控制。

from pydantic import BaseModel, Field from typing import List, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langgraph.graph import add_messages class AgentState(BaseModel): """智能体状态""" # 对话历史，使用归约器自动处理消息追加 messages: Annotated[List[BaseMessage], add_messages] = Field(default_factory=list) # 用户当前问题 question: str # 检索到的文档列表 documents: Optional[List[str]] = Field(default_factory=list) # 生成的初始答案 draft_answer: Optional[str] = None # 安全检查结果 safety_check_passed: Optional[bool] = None # 最终答复 final_answer: Optional[str] = None # 初始化一个空的向量存储检索器（这里用FAISS示例，你需要准备自己的数据） from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings # 假设我们已经有一个构建好的vectorstore # vectorstore = FAISS.load_local(“your_index”, embeddings, allow_dangerous_deserialization=True) # retriever = vectorstore.as_retriever(search_kwargs={“k”: 3})

3.2 实现各个节点函数

接下来，我们实现四个节点函数。

节点1：检索节点这个节点从State中取出用户问题，调用检索器获取相关文档。

def retrieve_node(state: AgentState) -> dict: """检索相关文档""" print(f“[Retrieve Node] 正在检索问题: {state.question}”) # 这里是模拟检索，实际应接入你的检索器 # retrieved_docs = retriever.invoke(state.question) # 模拟数据 simulated_docs = [ “文档A：LangGraph是一个用于构建有状态多智能体应用的框架。”, “文档B：它使用图结构来定义工作流，节点是函数，边是控制流。”, “文档C：State是贯穿整个图执行的共享内存。” ] return {“documents”: simulated_docs}

节点2：生成节点这个节点将用户问题和检索到的文档组合成提示词，调用LLM生成初步答案。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model=“gpt-3.5-turbo”, temperature=0) def generate_node(state: AgentState) -> dict: """基于问题和文档生成答案草稿""" print(f“[Generate Node] 基于 {len(state.documents)} 个文档生成答案。”) context = “\n\n”.join(state.documents) prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个专业的AI助手，请严格根据提供的上下文来回答问题。如果上下文不包含答案，请明确说‘根据已有信息无法回答’。”), (“human”, “上下文：{context}\n\n问题：{question}”) ]) chain = prompt | llm response = chain.invoke({“context”: context, “question”: state.question}) draft = response.content # 同时，将AI的回复也添加到消息历史中，保持对话完整性 new_messages = [AIMessage(content=draft)] return {“draft_answer”: draft, “messages”: new_messages}

节点3：安全检查节点这是一个非常重要的节点，用于对生成的内容进行过滤，防止输出有害或不合规信息。

def safety_check_node(state: AgentState) -> dict: """对生成的答案草稿进行安全检查""" print(f“[Safety Check Node] 正在检查答案安全性。”) draft = state.draft_answer # 这里可以实现你的安全检查逻辑，例如调用另一个LLM进行分类，或使用关键词过滤 # 本例使用一个简单的模拟检查 forbidden_keywords = [“有害内容”, “机密信息”, “政治敏感词示例”] check_passed = True for keyword in forbidden_keywords: if keyword in draft: check_passed = False break # 也可以基于规则或模型进行更复杂的判断 if len(draft) > 500: # 例如，答案太长可能意味着模型在胡编乱造 check_passed = False return {“safety_check_passed”: check_passed}

节点4：人工审核节点（模拟）如果安全检查未通过，流程将进入此节点。在实际应用中，这里可以连接到一个工单系统、通知接口，或者只是一个标记。

def human_review_node(state: AgentState) -> dict: """模拟人工审核环节""" print(f“[Human Review Node] 答案需要人工审核。草稿：{state.draft_answer[:100]}...”) # 在实际系统中，这里可能： # 1. 将任务放入审核队列 # 2. 发送邮件/钉钉/飞书通知 # 3. 调用一个内部审核API # 本例中，我们模拟审核后，手动修改答案 reviewed_answer = “[经人工审核] “ + state.draft_answer + “ (已确认安全)” return {“final_answer”: reviewed_answer}

节点5：完成节点如果安全检查通过，直接使用草稿作为最终答案。

def finalize_node(state: AgentState) -> dict: """完成节点，准备最终输出""" print(“[Finalize Node] 安全检查通过，生成最终答案。”) return {“final_answer”: state.draft_answer}

3.3 组装图并设置流程逻辑

现在，我们用StateGraph把这些节点和边组装起来。

from langgraph.graph import StateGraph, END # 1. 创建图构建器，指定State类型 workflow = StateGraph(AgentState) # 2. 添加所有节点 workflow.add_node(“retrieve”, retrieve_node) workflow.add_node(“generate”, generate_node) workflow.add_node(“safety_check”, safety_check_node) workflow.add_node(“human_review”, human_review_node) workflow.add_node(“finalize”, finalize_node) # 3. 设置入口点：从检索开始 workflow.set_entry_point(“retrieve”) # 4. 添加固定边：retrieve -> generate -> safety_check workflow.add_edge(“retrieve”, “generate”) workflow.add_edge(“generate”, “safety_check”) # 5. 添加条件边：根据安全检查结果路由 def route_by_safety(state: AgentState): if state.safety_check_passed: return “finalize” # 通过，去完成节点 else: return “human_review” # 未通过，去人工审核 workflow.add_conditional_edges( “safety_check”, route_by_safety, [“finalize”, “human_review”] # 可能的目的地 ) # 6. 从 human_review 和 finalize 节点连接到结束 workflow.add_edge(“human_review”, END) workflow.add_edge(“finalize”, END) # 7. 编译图，得到可执行对象 app = workflow.compile()

3.4 运行与可视化

现在，我们可以运行这个图了。

# 定义初始状态 initial_state = AgentState( messages=[], # 初始对话历史为空 question=“LangGraph是什么？它的核心概念有哪些？” ) # 执行图 final_state = app.invoke(initial_state) print(“\n=== 执行完成 ===”) print(f“最终答案：{final_state[‘final_answer’]}”) print(f“对话历史消息数：{len(final_state[‘messages’])}”)

为了更直观地理解流程，LangGraph提供了可视化功能（需要安装pygraphviz，或者使用get_graph().draw_mermaid()输出Mermaid文本）。

# 打印图的结构（文本形式） print(app.get_graph().draw_ascii()) # 或者，如果你想以图片形式保存（需要Graphviz） try: from langchain_core.runnables.graph import MermaidDrawer graph_image = MermaidDrawer(app.get_graph()).draw() # 可以将graph_image保存为文件或显示在notebook中 except ImportError: print(“如需生成图片，请安装 pygraphviz 和 pillow”)

4. 高级特性与实战技巧

掌握了基础构建后，我们来看看LangGraph的一些高级特性，这些特性能帮你处理更复杂的场景。

4.1 长期记忆（Persisted State）与检查点

LangGraph的一个杀手级特性是状态持久化。这意味着你可以中断一个长流程（比如一个持续多天的客户服务对话），稍后从断点恢复。这对于构建复杂的、有状态的会话代理至关重要。

实现持久化通常需要两个步骤：

1. 定义检查点（Checkpointer）：这是一个存储和加载State的组件。LangGraph支持内存、文件系统、数据库（如Redis、SQLite）等多种后端。
2. 在编译图时传入Checkpointer。

from langgraph.checkpoint import MemorySaver # 创建一个基于内存的检查点管理器（生产环境建议用Redis等） checkpointer = MemorySaver() # 编译图时传入checkpointer app_with_memory = workflow.compile(checkpointer=checkpointer) # 使用一个唯一的线程ID来标识这个会话 config = {“configurable”: {“thread_id”: “user_123_session_1”}} # 第一次调用，状态会被保存 initial_state = AgentState(question=“什么是状态图？”) result1 = app_with_memory.invoke(initial_state, config=config) print(f“第一次回答: {result1[‘final_answer’][:50]}...”) # 模拟一段时间后，用户提出后续问题 # 我们基于同一个thread_id调用，LangGraph会自动加载之前的状态 # 注意：我们需要更新State中的question，但messages等历史会被保留 new_state_input = {“question”: “它和有限状态机有什么区别？”} # 使用 `update_state` 方式调用，只更新question字段，其他状态继承 result2 = app_with_memory.invoke(new_state_input, config=config) print(f“第二次回答: {result2[‘final_answer’][:50]}...”) # 此时，result2的messages里包含了第一次和第二次的对话历史

实操心得：thread_id的设计非常关键。它可以是用户ID_会话ID的组合。对于Web应用，通常将thread_id存储在用户的会话（Session）或数据库记录中。这样，即使用户关闭浏览器再回来，也能恢复之前的对话上下文。

4.2 子图（Subgraph）与模块化

当你的应用变得非常庞大时，将整个图放在一个文件里会难以管理。子图允许你将一部分节点和边打包成一个独立的、可复用的单元。这极大地提升了代码的模块化和可维护性。

假设我们把“检索-生成”这个核心RAG流程打包成一个子图。

from langgraph.graph import StateGraph, END # 1. 首先，定义一个子图专用的、更精简的State class RAGState(BaseModel): question: str documents: List[str] = Field(default_factory=list) answer: str = “” # 2. 定义子图内部的节点（与之前类似，但使用RAGState） def sub_retrieve(state: RAGState): # … 检索逻辑 … return {“documents”: [“模拟文档1”, “模拟文档2”]} def sub_generate(state: RAGState): # … 生成逻辑 … return {“answer”: “基于文档生成的模拟答案”} # 3. 构建子图 sub_builder = StateGraph(RAGState) sub_builder.add_node(“retrieve”, sub_retrieve) sub_builder.add_node(“generate”, sub_generate) sub_builder.set_entry_point(“retrieve”) sub_builder.add_edge(“retrieve”, “generate”) sub_builder.add_edge(“generate”, END) # 子图有自己的结束点 # 编译子图 rag_subgraph = sub_builder.compile() # 4. 在主图中，将子图作为一个“超级节点”来使用 def rag_super_node(state: AgentState): """主图中调用子图的节点函数""" # 准备子图的输入 sub_input = RAGState(question=state.question) # 运行子图 sub_result = rag_subgraph.invoke(sub_input) # 将子图的结果整合到主图State中 return { “documents”: sub_result[“documents”], “draft_answer”: sub_result[“answer”] } # 在主图构建器中，像添加普通节点一样添加这个超级节点 main_builder = StateGraph(AgentState) main_builder.add_node(“rag_module”, rag_super_node) # 这里添加的是包装函数 # … 添加其他节点和边 …

子图的优势：

• 封装复杂性：将复杂流程隐藏在一个简单的接口后面。
• 复用性：同一个子图可以在主图的不同位置被多次调用。
• 独立测试：子图可以单独进行测试和调试。

4.3 并行执行与异步支持

LangGraph的节点默认是顺序执行的。但某些场景下，多个独立的任务可以并行执行以提升效率。虽然LangGraph核心API没有直接的“并行节点”概念，但可以通过模式来实现。

模式一：在单个节点内并行这是最常用的方式。在一个节点函数内部，使用asyncio或并发库来并行执行多个IO密集型任务（如同时调用多个不同的API或检索器）。

import asyncio from langchain_community.retrievers import WikipediaRetriever, ArxivRetriever async def parallel_retrieve_node(state: AgentState): """并行从多个数据源检索""" query = state.question # 创建多个检索任务 wiki_retriever = WikipediaRetriever() # ArxivRetriever可能需要配置 # arxiv_retriever = ArxivRetriever() tasks = [ asyncio.to_thread(wiki_retriever.invoke, query), # asyncio.to_thread(arxiv_retriever.invoke, query), asyncio.to_thread(simple_web_search, query), # 假设的另一个检索函数 ] # 并行等待所有任务完成 results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果，合并文档 all_docs = [] for r in results: if isinstance(r, Exception): print(f“某个检索源失败: {r}”) continue all_docs.extend(r) return {“documents”: all_docs}

模式二：使用StateGraph的add_edge的潜在并发在定义图时，如果从节点A有多条边分别指向节点B和节点C，并且没有条件限制，理论上B和C可以并发执行。但标准compile()后的图是顺序执行的。要实现真正的图级并发，需要探索LangGraph的多智能体（Multi-Agent）或消息传递（Message Passing）模式，这通常涉及更复杂的设计，如让多个“智能体”节点监听共享状态并独立运行。

注意事项：并行化会带来状态竞争的复杂性。确保并行节点修改的是State中不同的字段，或者使用线程安全的数据结构。对于简单的并行IO，推荐在节点函数内部实现。

4.4 与FastAPI等Web框架集成

LangGraph应用最终需要以API的形式提供服务。与FastAPI集成非常直接。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .your_langgraph_module import app, AgentState, checkpointer # 导入你编译好的图和组件 api_app = FastAPI(title=“LangGraph智能助手API”) class ChatRequest(BaseModel): thread_id: str # 客户端提供的会话ID question: str class ChatResponse(BaseModel): answer: str thread_id: str needs_review: bool = False @api_app.post(“/chat”, response_model=ChatResponse) async def chat_endpoint(request: ChatRequest): try: # 准备调用配置，包含持久化的thread_id config = {“configurable”: {“thread_id”: request.thread_id}} # 调用LangGraph应用。输入是一个字典，会更新到State的对应字段。 inputs = {“question”: request.question} result = app.invoke(inputs, config=config) # 构建响应 return ChatResponse( answer=result.get(“final_answer”, “”), thread_id=request.thread_id, needs_review=not result.get(“safety_check_passed”, True) ) except Exception as e: raise HTTPException(status_code=500, detail=f“处理请求时出错: {str(e)}”) # 可以再添加一个端点来初始化或清理会话 @api_app.delete(“/session/{thread_id}”) async def clear_session(thread_id: str): # 这里需要调用checkpointer的删除方法（如果支持） # 例如：checkpointer.delete(config) return {“message”: f”Session {thread_id} cleared (模拟)”}

这样，你就拥有了一个具备长期记忆能力的AI助手后端。前端只需要维护一个thread_id（可以是用户ID+时间戳），就能实现连续对话。

5. 常见问题排查与调试技巧实录

在实际开发中，你肯定会遇到各种问题。下面是我踩过的一些坑和总结的排查技巧。

5.1 状态更新不符合预期

问题现象：节点返回了数据，但State中的字段没有被更新，或者被错误地覆盖。

根因与解决：

1. 归约器（Reducer）冲突：这是最常见的原因。回想一下State定义中的Annotated[List, add_messages]。add_messages就是一个归约器。如果你定义了一个字段history: Annotated[List, add_messages]，但在节点中返回{“history”: [“new item”]}，归约器add_messages会尝试用它的逻辑（通常是追加）来合并这个新值。如果你期望的是完全替换，那就会出错。
   • 解决：仔细检查State中每个Annotated字段指定的归约器是否与你的更新意图匹配。对于非列表型、需要直接替换的字段，不要使用Annotated，或者使用operator.setitem作为归约器（表示直接设置）。

   from typing import Annotated import operator class MyState(TypedDict): direct_replace_field: Annotated[str, operator.setitem] # 直接替换 append_only_list: Annotated[List, add_messages] # 追加

2. 返回字典的键错误：节点返回的字典，其键必须与State中定义的字段名完全一致。Python是大小写敏感的，{“Answer”: …}无法更新answer字段。
   • 解决：使用IDE的自动补全或仔细核对字段名。建议从State类的定义中直接复制字段名。

5.2 条件边（Conditional Edge）路由错误

问题现象：流程没有按预期的分支走，或者抛出InvalidUpdateError。

排查步骤：

1. 打印调试：在条件函数route_by_safety中打印state，确保你用来做判断的字段（如state.safety_check_passed）在此时已经被正确赋值。
2. 检查返回值类型：条件函数必须返回一个字符串。这个字符串必须是add_conditional_edges方法中path参数列表里的一个，或者是END。
3. 检查所有分支：确保条件函数的每一个可能的分支都有返回值。最稳妥的方法是最后加一个else子句。

   def route_function(state): if state[‘value’] > 10: return “path_a” elif state[‘value’] > 5: return “path_b” else: # 确保所有情况都被覆盖 return “path_c” # 或 return END

5.3 图编译或执行时报错

问题现象：在调用app.invoke()时出现KeyError,AttributeError或各种Pydantic验证错误。

排查清单：

调试建议：在开发阶段，可以打开LangGraph的调试输出，它会显示每个节点的输入和输出。

# 一种简单的调试方式：在每个节点函数开始和结束打印日志 def my_node(state): print(f”>>> Entering my_node. State keys: {state.keys()}”) # … your logic … result = {“key”: “value”} print(f”<<< Exiting my_node. Returning: {result}”) return result

5.4 性能优化点

1. LLM调用异步化：如果图中多个节点需要调用LLM，且它们之间没有严格的先后依赖，考虑使用async节点函数和异步LLM客户端（如AsyncOpenAI），并在主循环中使用asyncio.gather并行调用，可以显著减少I/O等待时间。
2. 状态精简：持久化检查点会保存整个State。避免在State中存储过大的数据（如图片二进制流）。可以只存储数据的引用（如URL或数据库ID）。
3. 图的编译：workflow.compile()会进行优化。对于生产环境，编译一次并复用app对象，不要每次请求都重新编译。
4. 超时与重试：对于调用外部API的节点，务必设置超时和重试机制，避免单个节点挂起导致整个工作流卡死。可以使用tenacity等重试库。

6. LangGraph vs LangChain：如何选择？

这是被问得最多的问题之一。它们不是替代关系，而是互补关系。

如何选择？

• 从LangChain开始：如果你的需求是“调用一个LLM，然后根据结果再调用一个工具，然后结束”，用LangChain的Chain或简单Agent就够了。
• 升级到LangGraph：当你发现你的Chain里嵌套了太多if-else；当你需要管理跨越多次调用的对话状态；当你需要实现一个清晰的、有多个阶段和决策点的业务流程时，就是引入LangGraph的最佳时机。

它们可以一起用吗？当然可以！最常见的模式是：用LangGraph作为顶层的流程控制器，用LangChain的LCEL来构建每个节点内部的强大功能。例如，一个“研究助理”图中，可能有一个“搜索节点”，这个节点内部就是用LangChain的RetrievalChain实现的。

最后，关于开头热词中的“Dify和LangGraph可以一起用吗？”，答案是肯定的。Dify作为一个AI应用平台，可以负责前端界面、用户管理、知识库管理、模型编排等。而LangGraph可以作为Dify后台的一个“自定义工具”或“高级工作流引擎”，来处理Dify内置工作流引擎无法表达的、极其复杂的业务逻辑。你可以通过Dify的API触发一个LangGraph工作流，并获取结果。