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这里有几个关键点TypedDictvsBaseModel官方示例常用TypedDict因为它更轻量与Python字典兼容性好。但在复杂场景下使用Pydantic的BaseModel能获得更强大的数据验证和序列化能力。我个人在需要严格数据格式或与外部API交互时更倾向于用BaseModel。Annotated类型注解这是LangGraph的一个精髓。Annotated[List, add_messages]不仅仅声明messages是一个列表还指定了一个归约器reduceradd_messages。这意味着当多个节点都想修改messages时比如用户节点添加用户消息AI节点添加AI回复add_messages函数会决定如何合并这些修改通常是追加到列表末尾。这是实现状态更新的关键机制。状态设计原则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_factorylist) # 用户当前问题 question: str # 检索到的文档列表 documents: Optional[List[str]] Field(default_factorylist) # 生成的初始答案 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_deserializationTrue) # 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 [ “文档ALangGraph是一个用于构建有状态多智能体应用的框架。”, “文档B它使用图结构来定义工作流节点是函数边是控制流。”, “文档CState是贯穿整个图执行的共享内存。” ] return {“documents”: simulated_docs}节点2生成节点这个节点将用户问题和检索到的文档组合成提示词调用LLM生成初步答案。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) 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(contentdraft)] 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的一个杀手级特性是状态持久化。这意味着你可以中断一个长流程比如一个持续多天的客户服务对话稍后从断点恢复。这对于构建复杂的、有状态的会话代理至关重要。实现持久化通常需要两个步骤定义检查点Checkpointer这是一个存储和加载State的组件。LangGraph支持内存、文件系统、数据库如Redis、SQLite等多种后端。在编译图时传入Checkpointer。from langgraph.checkpoint import MemorySaver # 创建一个基于内存的检查点管理器生产环境建议用Redis等 checkpointer MemorySaver() # 编译图时传入checkpointer app_with_memory workflow.compile(checkpointercheckpointer) # 使用一个唯一的线程ID来标识这个会话 config {“configurable”: {“thread_id”: “user_123_session_1”}} # 第一次调用状态会被保存 initial_state AgentState(question“什么是状态图”) result1 app_with_memory.invoke(initial_state, configconfig) 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, configconfig) 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_factorylist) 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(questionstate.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_exceptionsTrue) # 处理结果合并文档 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_modelChatResponse) 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, configconfig) # 构建响应 return ChatResponse( answerresult.get(“final_answer”, “”), thread_idrequest.thread_id, needs_reviewnot result.get(“safety_check_passed”, True) ) except Exception as e: raise HTTPException(status_code500, detailf“处理请求时出错: {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中的字段没有被更新或者被错误地覆盖。根因与解决归约器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] # 追加返回字典的键错误节点返回的字典其键必须与State中定义的字段名完全一致。Python是大小写敏感的{“Answer”: …}无法更新answer字段。解决使用IDE的自动补全或仔细核对字段名。建议从State类的定义中直接复制字段名。5.2 条件边Conditional Edge路由错误问题现象流程没有按预期的分支走或者抛出InvalidUpdateError。排查步骤打印调试在条件函数route_by_safety中打印state确保你用来做判断的字段如state.safety_check_passed在此时已经被正确赋值。检查返回值类型条件函数必须返回一个字符串。这个字符串必须是add_conditional_edges方法中path参数列表里的一个或者是END。检查所有分支确保条件函数的每一个可能的分支都有返回值。最稳妥的方法是最后加一个else子句。def route_function(state): if state[‘value’] 10: return “path_a” elif state[‘value’] 5: return “path_b” else: # 确保所有情况都被覆盖 return “path_c” # 或 return END5.3 图编译或执行时报错问题现象在调用app.invoke()时出现KeyError,AttributeError或各种Pydantic验证错误。排查清单错误类型可能原因解决方案KeyError节点函数尝试访问State中不存在的键。1. 检查节点函数输入参数名是否与State字段名匹配。2. 确保上游节点已经将所需数据写入State。ValidationError(Pydantic)节点返回的数据类型与State字段定义的类型不匹配。1. 检查节点返回值的类型。例如字段定义为List[str]就不能返回单个str。2. 对于可选字段返回None是允许的。InvalidUpdateErrorLangGraph无法将节点的更新应用到State。1. 最常见于归约器冲突见5.1。2. 检查条件边返回了无效的目的地。节点函数内部异常你的节点代码本身有bug如调用失败的API。1. 用try…except包裹节点函数内部可能出错的部分并返回一个错误状态。2. 在节点函数内增加详细的日志。调试建议在开发阶段可以打开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 result5.4 性能优化点LLM调用异步化如果图中多个节点需要调用LLM且它们之间没有严格的先后依赖考虑使用async节点函数和异步LLM客户端如AsyncOpenAI并在主循环中使用asyncio.gather并行调用可以显著减少I/O等待时间。状态精简持久化检查点会保存整个State。避免在State中存储过大的数据如图片二进制流。可以只存储数据的引用如URL或数据库ID。图的编译workflow.compile()会进行优化。对于生产环境编译一次并复用app对象不要每次请求都重新编译。超时与重试对于调用外部API的节点务必设置超时和重试机制避免单个节点挂起导致整个工作流卡死。可以使用tenacity等重试库。6. LangGraph vs LangChain如何选择这是被问得最多的问题之一。它们不是替代关系而是互补关系。特性LangChainLangGraph核心定位AI应用开发框架提供与各种LLM、工具、检索器交互的标准化组件LCEL。复杂工作流编排框架专注于管理有状态的、多步骤的、带条件分支和循环的AI应用流程。编程范式链式Chain和代理Agent。链是线性的代理通过LLM决定下一步动作。基于图Graph。显式地定义所有节点和边控制流是确定的或由条件函数决定。状态管理状态通常通过链的输入/输出传递或在自定义的Runnable中管理相对隐式。状态是头等公民。有明确的、类型化的State对象贯穿整个图的生命周期。适用场景快速构建简单的问答、文本处理、单次工具调用等场景。Agent适合开放性的、步骤不确定的任务。构建复杂的、有明确业务流程的AI应用。如多轮审批、游戏模拟、带故障恢复的自动化流程、需要长期记忆的对话机器人。可预测性Agent的行为由LLM决定有一定随机性调试难度较高。流程由开发者定义的图决定可预测性强易于调试和测试。学习曲线相对平缓从简单的PromptLLM开始即可。如何选择从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工作流并获取结果。