拓冰网站优化 
1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一年里折腾过LLM应用开发那么“LangChain”这个名字对你来说肯定不陌生。它像是一套强大的乐高积木提供了连接大语言模型、工具、数据源的各种组件。我们用Chains把任务串起来用Agents赋予模型使用工具的能力。但说实话当业务流程变得稍微复杂一点——比如需要循环判断、状态持久化或者多角色协作时仅用Chain和Agent来搭建代码很快就会变得像一团纠缠的意大利面状态管理混乱调试起来更是噩梦。这就是LangGraph出现的背景。它不是要取代LangChain而是站在它的肩膀上解决更复杂的“编排”问题。你可以把LangChain看作是提供了标准化的砖块和接口而LangGraph则提供了设计图纸和施工流程专门用来搭建那些有状态、带循环、多分支的工作流。它引入了“图”的概念将应用中的每个步骤节点和步骤之间的流转逻辑边清晰地定义出来。这对于构建智能体、复杂的对话系统、多步骤决策引擎来说是游戏规则的改变者。简单来说当你需要模型“记住”之前的对话、根据中间结果决定下一步是调用工具还是直接回答、或者让多个AI角色像团队一样协作时LangGraph就是你该深入研究的工具。它让复杂逻辑变得可视化、可调试、可维护。2. LangGraph核心三要素State, Node, Edge理解LangGraph最关键的就是吃透它的三个核心概念状态State、节点Node和边Edge。这构成了任何LangGraph应用的骨架。2.1 State工作流的记忆中枢在LangGraph中State是一个贯穿整个工作流执行过程的共享数据容器。它通常是一个Pydantic模型或是一个简单的字典定义了工作流需要关心和传递的所有数据。为什么需要State想象一个客服机器人。用户问“北京的天气怎么样” 机器人回答后用户接着问“那上海呢” 一个没有状态的系统会把第二个问题当作独立问题处理而一个有状态的系统会记得上一个问题是关于“天气”的从而更准确地理解“那上海呢”指的是“上海的天气”。State就是用来保存这类上下文、中间计算结果、历史消息等一切需要跨节点共享的信息。定义一个State非常简单通常使用TypedDict或Pydantic BaseModel。例如一个简单的对话状态可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史add_messages是一个归并函数确保消息列表正确追加 messages: Annotated[List[dict], add_messages] # 用户的最新问题 user_query: str # 从知识库检索到的相关文档 retrieved_docs: List[str] # 模型生成的最终答案 final_answer: str这里的关键是Annotated的使用。Annotated[List[dict], add_messages]不仅声明了messages字段的类型是List[dict]还通过add_messages指定了当多个节点试图修改这个字段时应该如何合并reduce这些修改。add_messages是LangGraph提供的一个内置归并函数专门用于合并消息列表确保对话历史有序且不重复。对于其他字段如user_query如果没有特殊归并需求可以使用operator.add或直接覆盖。实操心得State设计是第一步也是最容易出错的一步。一开始就要想清楚工作流需要哪些数据哪些数据是只由一个节点产生并使用的可以定义为局部变量哪些数据需要在多个节点间共享和传递必须放入State过度设计State会导致图复杂度剧增而设计不足又无法实现功能。我的经验是先从最小可行状态开始随着节点增加再逐步扩展。2.2 Node执行具体任务的单元Node节点是图中的一个功能单元它接收当前的State执行一些操作如调用LLM、查询数据库、运行代码然后返回一个更新后的State字典。这个字典包含了它希望修改的State部分。节点本质上就是一个函数。它的签名是固定的function_name(state: State) - PartialState。PartialState是一个字典其键是State中定义的字段名值是你想为该字段设置的新值。LangGraph会用这个返回的字典去更新全局的State。让我们看一个调用大语言模型的节点示例from langchain_openai import ChatOpenAI # 初始化模型 llm ChatOpenAI(model“gpt-4o”) def call_llm_node(state: State): 节点调用LLM生成回答 # 1. 从State中获取需要的信息 conversation_history state[“messages”] query state[“user_query”] # 2. 构建发送给LLM的提示词 system_prompt “你是一个有帮助的助手。” human_message query # 3. 调用LLM response llm.invoke([ {“role”: “system”, “content”: system_prompt}, *conversation_history, {“role”: “user”, “content”: human_message} ]) # 4. 将LLM的回复封装成消息格式 ai_message {“role”: “assistant”, “content”: response.content} # 5. 返回要更新的State部分 # 注意我们只返回需要修改的字段。这里更新了消息历史。 return {“messages”: [ai_message]}这个节点做了几件事读取状态、准备数据、执行核心逻辑调用API、处理结果、最后返回状态更新。关键点在于节点函数只返回它想要修改的那部分State而不是完整的State。LangGraph的运行时负责将这些部分更新智能地合并到全局State中。2.3 Edge控制流程的方向盘如果Node是车站那么Edge边就是连接车站的轨道它决定了工作流的下一步该去哪里。边定义了基于当前State的条件逻辑。边主要分为两种普通边Fixed Edges无条件地将一个节点的输出连接到另一个节点的输入。这用于定义固定的、线性的执行顺序。条件边Conditional Edges根据State中的某个值或某个判断函数的返回值动态地决定下一个要执行的节点。这是实现循环、分支判断的关键。条件边是LangGraph强大灵活性的源泉。它允许你创建诸如“如果模型回答中包含不确定词汇则跳转到‘检索文档’节点否则直接结束”这样的逻辑。定义一个条件边通常需要创建一个路由函数。这个函数接收State返回一个字符串这个字符串就是下一个要执行的节点的名字。def should_retrieve_docs(state: State) - str: 路由函数判断是否需要检索文档 last_message state[“messages”][-1] llm_response last_message[“content”] # 简单的关键词判断逻辑实际应用中会更复杂 uncertainty_keywords [“我不确定”, “我不知道”, “根据我的知识”] if any(keyword in llm_response for keyword in uncertainty_keywords): return “retrieve_node” # 跳转到检索节点 else: return “end” # 结束流程在构建图时你会将这个路由函数与特定的边关联起来从而创建动态的工作流路径。3. 构建你的第一个LangGraph应用一个简单的对话代理理论说得再多不如动手搭一个。我们来构建一个最简单的、带条件判断的对话代理。它的逻辑是用户提问 - LLM尝试回答 - 如果LLM表示不确定则去检索知识库 - 基于检索结果重新生成答案。3.1 环境准备与依赖安装首先确保你的Python环境建议3.10以上并安装必要库。我们将使用langgraph、langchain-openai用于调用GPT以及langchain-community假设我们用一个简单的向量库模拟检索。pip install langgraph langchain-openai langchain-community为了模拟检索我们用一个内存中的FAISS向量存储和OpenAIEmbeddings。你需要在环境变量中设置你的OPENAI_API_KEY。import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.documents import Document # 设置API Key (请替换为你的key) os.environ[“OPENAI_API_KEY”] “your-api-key-here” # 初始化LLM和Embedding模型 llm ChatOpenAI(model“gpt-4o”, temperature0) embeddings OpenAIEmbeddings() # 创建一个模拟的知识库 docs [ Document(page_content“LangGraph是一个用于构建有状态、多参与者LLM应用的库。”), Document(page_content“LangChain提供了连接LLM与外部工具和数据的组件。”), Document(page_content“状态State在LangGraph中用于在节点间传递信息。”), ] vectorstore FAISS.from_documents(docs, embeddings) retriever vectorstore.as_retriever()3.2 定义State与节点函数根据我们的流程State需要包含消息历史、用户查询、检索到的文档和最终答案。from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AgentState(TypedDict): 定义我们代理的工作流状态 messages: Annotated[List[dict], add_messages] user_query: str retrieved_docs: Optional[List[Document]] final_answer: Optional[str]接下来定义三个节点函数call_llm_node: 直接调用LLM回答问题。retrieve_docs_node: 从知识库检索相关文档。generate_with_context_node: 基于检索到的文档让LLM生成最终答案。def call_llm_node(state: AgentState): 节点1直接调用LLM生成初始回答 query state[“user_query”] history state[“messages”] # 构建提示词 prompt f“请回答以下问题{query}。如果你非常确定答案请直接给出。如果你不确定请在回答中包含‘我不确定’或类似表述。” response llm.invoke([ *history, {“role”: “user”, “content”: prompt} ]) ai_msg {“role”: “assistant”, “content”: response.content} return {“messages”: [ai_msg]} def retrieve_docs_node(state: AgentState): 节点2根据用户查询检索相关文档 query state[“user_query”] docs retriever.invoke(query) # 检索相关文档 return {“retrieved_docs”: docs} def generate_with_context_node(state: AgentState): 节点3结合检索到的文档生成最终答案 query state[“user_query”] docs state[“retrieved_docs”] context “\n\n”.join([doc.page_content for doc in docs]) prompt f“请基于以下背景知识回答问题。\n背景知识{context}\n\n问题{query}” response llm.invoke([{“role”: “user”, “content”: prompt}]) ai_msg {“role”: “assistant”, “content”: response.content} return {“messages”: [ai_msg], “final_answer”: response.content}3.3 定义条件路由逻辑我们需要一个路由函数来检查LLM的初始回答是否包含不确定性从而决定是直接结束还是去检索文档。def route_after_initial_answer(state: AgentState) - str: 路由函数分析LLM初始回答决定下一步 last_message state[“messages”][-1] answer last_message[“content”].lower() # 判断是否包含不确定性表述 uncertainty_phrases [“i’m not sure”, “i don’t know”, “不确定”, “不了解”, “根据我的知识”] if any(phrase in answer for phrase in uncertainty_phrases): return “retrieve_docs” # 需要检索 else: # LLM看起来很确定将其回答作为最终答案 state[“final_answer”] last_message[“content”] return “end” # 直接结束3.4 组装图并运行现在把所有的节点和边组装成一张完整的图。from langgraph.graph import StateGraph, END # 1. 创建一个图并指定它使用的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“call_llm”, call_llm_node) workflow.add_node(“retrieve_docs”, retrieve_docs_node) workflow.add_node(“generate_final_answer”, generate_with_context_node) # 3. 设置入口点 workflow.set_entry_point(“call_llm”) # 4. 添加边包括条件边 # 从 call_llm 节点出来后根据路由函数决定去向 workflow.add_conditional_edges( “call_llm”, route_after_initial_answer, { “retrieve_docs”: “retrieve_docs”, # 如果返回“retrieve_docs”则跳转到对应节点 “end”: END # 如果返回“end”则直接结束工作流 } ) # 5. 添加固定边 # 从 retrieve_docs 节点无条件地连接到 generate_final_answer 节点 workflow.add_edge(“retrieve_docs”, “generate_final_answer”) # 从 generate_final_answer 节点无条件地连接到结束 workflow.add_edge(“generate_final_answer”, END) # 6. 编译图 app workflow.compile()至此你的第一个LangGraph应用就构建完成了你可以通过app.get_graph().draw_mermaid()来生成一个Mermaid图在支持的环境下可视化你的工作流。现在来运行它# 定义初始状态 initial_state: AgentState { “messages”: [], “user_query”: “LangGraph和LangChain有什么区别”, “retrieved_docs”: None, “final_answer”: None } # 运行工作流 final_state app.invoke(initial_state) print(“最终答案”, final_state[“final_answer”]) print(“完整消息历史”, final_state[“messages”])当你运行这段代码时工作流会启动。LLM会先尝试直接回答。由于这个问题需要精确对比LLM很可能会在回答中包含“根据我的知识”这类表述从而触发路由函数走向retrieve_docs节点。检索节点会从我们模拟的知识库中找到相关文档然后generate_final_answer节点会结合这些文档生成一个更准确的答案。最终这个答案会被保存在final_state[“final_answer”]中。4. 深入LangGraph高级特性子图、持久化与消息管理当你掌握了基础构建后LangGraph的一些高级特性能让你的应用变得更强大、更易管理。4.1 使用子图Subgraphs模块化复杂逻辑当一个工作流变得非常庞大时将所有节点都放在主图中会难以维护。子图允许你将一组相关的节点和边打包成一个独立的、可复用的单元。这就像编程中的函数让主图的结构变得清晰。假设我们上面例子中的“检索-生成”流程retrieve_docsgenerate_final_answer在很多地方都会用到我们可以将其封装成一个子图。from langgraph.graph import StateGraph def create_rag_subgraph(): 创建一个RAG检索增强生成子图 rag_graph StateGraph(AgentState) # 添加子图内部的节点函数定义同上 rag_graph.add_node(“retrieve”, retrieve_docs_node) rag_graph.add_node(“generate”, generate_with_context_node) # 定义子图内部的边 rag_graph.add_edge(“retrieve”, “generate”) rag_graph.set_entry_point(“retrieve”) rag_graph.set_finish_point(“generate”) # 设置子图的出口节点 return rag_graph.compile() # 在主图中使用子图 main_workflow StateGraph(AgentState) main_workflow.add_node(“call_llm”, call_llm_node) # 将子图作为一个“节点”添加到主图 rag_app create_rag_subgraph() main_workflow.add_node(“rag_process”, rag_app) # rag_app本身就是一个可调用的图 # ... 后续设置主图的边将“call_llm”连接到“rag_process”或“END”这样主图只需要关心“调用LLM”和“执行RAG流程”这两个高级步骤细节被隐藏在了子图中大大提升了代码的可读性和可复用性。4.2 状态持久化Persistence与长期记忆对于聊天机器人或多轮对话应用我们经常需要将会话状态保存下来以便用户下次回来时能继续对话。LangGraph内置了强大的持久化支持可以轻松地将State保存到数据库如SQLite、PostgreSQL或内存中。持久化的核心是Persistence接口。以下是一个使用SqliteSaver将状态保存到本地SQLite数据库的示例from langgraph.checkpoint.sqlite import SqliteSaver # 初始化一个SQLite检查点存储器 persister SqliteSaver.from_conn_string(“:memory:”) # 使用内存数据库也可用文件路径如“checkpoints.db” # 在编译图时传入persister app workflow.compile(checkpointerpersister) # 运行工作流时需要指定一个线程IDthread_id来标识唯一的会话 config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {“messages”: [], “user_query”: “你好”} # invoke会保存状态 result1 app.invoke(initial_state, configconfig) print(“第一次回答后状态已保存。”) # 模拟用户接着提问 new_state {“messages”: result1[“messages”], “user_query”: “我刚才问了什么”} # 再次调用LangGraph会从检查点恢复上一次的完整状态并在此基础上更新 result2 app.invoke(new_state, configconfig) print(“第二次回答它记得之前的对话”, result2[“messages”][-1][“content”])通过thread_id你可以为不同的用户或会话创建独立的、可恢复的状态流。这对于构建需要“长期记忆”的智能体至关重要。4.3 细粒度消息管理在复杂的对话流中你可能需要更精细地控制消息列表。langgraph.graph.message模块提供了一些工具函数除了我们之前用到的add_messages还有trim_messages: 根据Token数量、消息条数或自定义函数来修剪历史消息防止上下文窗口溢出。AnyMessage,HumanMessage,AIMessage等类型化的消息类提供更好的类型提示。例如实现一个只保留最近5轮对话的消息管理策略from langgraph.graph.message import trim_messages class StateWithTrimmedMessages(TypedDict): messages: Annotated[List[dict], add_messages, trim_messages(max_messages10)] # 保留最多10条消息 # ... 其他字段这样每次更新messages字段时trim_messages归并器会自动确保列表长度不超过10条无需在节点函数中手动处理。5. 常见问题排查与性能优化实战在实际开发中你肯定会遇到各种问题。下面是我在项目中踩过的一些坑和总结的解决方案。5.1 问题一State更新不符合预期症状节点返回了更新字典但最终State中的值没有被正确修改或者被意外覆盖。根因与排查归并器Reducer冲突这是最常见的原因。State中每个字段的Annotated第二个参数就是它的归并器。如果你为List字段设置了add_messages但节点返回的更新值不是一个List[dict]或者格式不对归并就会失败。对于简单字段如str,int默认归并器是operator.setitem后写入的覆盖先前的。如果你希望是累加需要自定义归并器。节点返回了完整的State节点函数应该只返回它想修改的字段字典。如果你错误地返回了完整的State对象可能会干扰其他字段。解决方案打印调试在每个节点函数的开头和结尾打印state和return的值。检查归并器确保你理解每个字段的归并逻辑。对于复杂合并可以自定义归并函数。使用类型提示严格定义State的TypedDict或Pydantic模型IDE和mypy能帮你提前发现许多类型错误。5.2 问题二条件边Conditional Edge不生效症状工作流没有按照预想的分支执行总是走固定路径或报错。根因与排查路由函数返回值与映射不匹配add_conditional_edges的第三个参数是一个映射字典。你的路由函数返回的字符串必须是这个字典的一个键。如果返回了字典中不存在的字符串图会找不到下一个节点而报错。路由函数逻辑错误路由函数的判断条件可能过于严格或宽松导致预期外的分支选择。仔细检查路由函数中对State的读取和判断逻辑。State中依赖的数据未就绪路由函数在Node A之后执行它读取的State应该是Node A更新后的。确保Node A已经将路由判断所需的数据正确地写入了State。解决方案在路由函数内增加详细的日志打印判断条件和返回值。使用app.get_graph().draw_mermaid()可视化你的图确认条件边的指向是否正确。简化测试先让路由函数返回一个固定值测试分支是否能正确跳转再逐步完善判断逻辑。5.3 问题三图编译或执行速度慢症状对于简单的逻辑app.invoke()执行时间过长。根因与排查节点中的阻塞操作最常见的瓶颈是节点内部同步调用了耗时的I/O操作如网络请求LLM API调用、数据库查询、大文件读写等。在同步代码中这些操作会阻塞整个线程。State过于庞大如果State中存储了非常大的对象如整个文档内容、大型列表在节点间序列化/反序列化传递时会消耗额外时间。未利用并发如果图中有多个可以并行执行的节点例如同时调用两个不同的API使用默认的同步执行方式会串行运行浪费了时间。解决方案异步化推荐将节点函数定义为async def并在其中使用await调用异步客户端。LangGraph完全支持异步执行。from langchain_openai import AsyncChatOpenAI async_llm AsyncChatOpenAI(model“gpt-4o”) async def async_call_llm_node(state: State): response await async_llm.ainvoke(...) return {“messages”: [response]}然后用app.ainvoke()来异步调用整个图效率会大幅提升。精简State只把必要的数据放在State中。对于大型数据可以考虑只存储引用如ID、路径在节点内部按需加载。分析性能使用Python的cProfile或line_profiler工具定位具体的耗时函数。5.4 调试与可视化技巧使用stream进行逐步调试app.invoke()是一次性执行到底。使用app.stream()可以让你以流式方式获取每个节点执行前后的状态非常适合调试。for step in app.stream(initial_state, configconfig): node_name step[0] # 执行的节点名 node_output step[1] # 节点执行后的状态 print(f“节点 [{node_name}] 执行完毕。当前State: {node_output}”)Mermaid可视化在Jupyter Notebook或支持Mermaid的Markdown查看器中app.get_graph().draw_mermaid()可以生成工作流的可视化图一目了然地看清所有节点和边的结构。打印编译后的图结构print(app.get_graph())或print(app.get_graph().to_json())可以输出图的JSON表示用于深入分析。6. LangGraph与LangChain、Dify的定位辨析社区里经常有人混淆LangGraph、LangChain以及Dify这类平台的关系。理解它们的定位差异能帮你更好地选择工具。LangChain可以看作是**“基础设施层”或“组件库”**。它提供了与各种LLM、嵌入模型、向量数据库、工具等进行交互的标准接口LLMRetrieverTool。它的核心价值在于标准化和集成让你不用为每个模型或数据库写不同的适配代码。Chain和Agent是其上层的两种编排模式但相对简单。LangGraph是**“复杂编排层”。它专注于解决当多个步骤之间存在复杂依赖、循环、状态共享时的流程控制问题。它利用“图”这一计算机科学中的经典模型提供了声明式的方式来定义工作流。它底层可以使用LangChain的组件如ChatOpenAI,FAISS但它的核心价值是编排逻辑本身**。Dify/AutoGen等平台属于**“应用层”或“低代码平台”**。它们提供了图形化界面让你可以通过拖拽等方式构建AI工作流并集成了用户管理、知识库管理、API部署、监控等开箱即用的功能。它们可能底层使用了LangChain或LangGraph的技术但目标用户是更希望关注业务逻辑而非代码的开发者或产品经理。关系类比如果你想快速搭建一个包含前端、后端、数据库的完整AI应用选Dify这类平台。如果你想用代码灵活地连接各种AI模型和数据源构建一个中等复杂度的流程LangChain是很好的起点。当你用LangChain构建的Agent逻辑变得无比复杂、难以维护或者你需要精细控制带有状态循环和分支的复杂工作流时就该引入LangGraph了。它们不是互斥的而是可以协同工作。一个典型的模式是用LangChain的组件处理与外部服务的交互用LangGraph的图来编排这些组件之间的复杂执行逻辑。 LangGraph实战：构建有状态、可编排的复杂AI工作流
1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一年里折腾过LLM应用开发那么“LangChain”这个名字对你来说肯定不陌生。它像是一套强大的乐高积木提供了连接大语言模型、工具、数据源的各种组件。我们用Chains把任务串起来用Agents赋予模型使用工具的能力。但说实话当业务流程变得稍微复杂一点——比如需要循环判断、状态持久化或者多角色协作时仅用Chain和Agent来搭建代码很快就会变得像一团纠缠的意大利面状态管理混乱调试起来更是噩梦。这就是LangGraph出现的背景。它不是要取代LangChain而是站在它的肩膀上解决更复杂的“编排”问题。你可以把LangChain看作是提供了标准化的砖块和接口而LangGraph则提供了设计图纸和施工流程专门用来搭建那些有状态、带循环、多分支的工作流。它引入了“图”的概念将应用中的每个步骤节点和步骤之间的流转逻辑边清晰地定义出来。这对于构建智能体、复杂的对话系统、多步骤决策引擎来说是游戏规则的改变者。简单来说当你需要模型“记住”之前的对话、根据中间结果决定下一步是调用工具还是直接回答、或者让多个AI角色像团队一样协作时LangGraph就是你该深入研究的工具。它让复杂逻辑变得可视化、可调试、可维护。2. LangGraph核心三要素State, Node, Edge理解LangGraph最关键的就是吃透它的三个核心概念状态State、节点Node和边Edge。这构成了任何LangGraph应用的骨架。2.1 State工作流的记忆中枢在LangGraph中State是一个贯穿整个工作流执行过程的共享数据容器。它通常是一个Pydantic模型或是一个简单的字典定义了工作流需要关心和传递的所有数据。为什么需要State想象一个客服机器人。用户问“北京的天气怎么样” 机器人回答后用户接着问“那上海呢” 一个没有状态的系统会把第二个问题当作独立问题处理而一个有状态的系统会记得上一个问题是关于“天气”的从而更准确地理解“那上海呢”指的是“上海的天气”。State就是用来保存这类上下文、中间计算结果、历史消息等一切需要跨节点共享的信息。定义一个State非常简单通常使用TypedDict或Pydantic BaseModel。例如一个简单的对话状态可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史add_messages是一个归并函数确保消息列表正确追加 messages: Annotated[List[dict], add_messages] # 用户的最新问题 user_query: str # 从知识库检索到的相关文档 retrieved_docs: List[str] # 模型生成的最终答案 final_answer: str这里的关键是Annotated的使用。Annotated[List[dict], add_messages]不仅声明了messages字段的类型是List[dict]还通过add_messages指定了当多个节点试图修改这个字段时应该如何合并reduce这些修改。add_messages是LangGraph提供的一个内置归并函数专门用于合并消息列表确保对话历史有序且不重复。对于其他字段如user_query如果没有特殊归并需求可以使用operator.add或直接覆盖。实操心得State设计是第一步也是最容易出错的一步。一开始就要想清楚工作流需要哪些数据哪些数据是只由一个节点产生并使用的可以定义为局部变量哪些数据需要在多个节点间共享和传递必须放入State过度设计State会导致图复杂度剧增而设计不足又无法实现功能。我的经验是先从最小可行状态开始随着节点增加再逐步扩展。2.2 Node执行具体任务的单元Node节点是图中的一个功能单元它接收当前的State执行一些操作如调用LLM、查询数据库、运行代码然后返回一个更新后的State字典。这个字典包含了它希望修改的State部分。节点本质上就是一个函数。它的签名是固定的function_name(state: State) - PartialState。PartialState是一个字典其键是State中定义的字段名值是你想为该字段设置的新值。LangGraph会用这个返回的字典去更新全局的State。让我们看一个调用大语言模型的节点示例from langchain_openai import ChatOpenAI # 初始化模型 llm ChatOpenAI(model“gpt-4o”) def call_llm_node(state: State): 节点调用LLM生成回答 # 1. 从State中获取需要的信息 conversation_history state[“messages”] query state[“user_query”] # 2. 构建发送给LLM的提示词 system_prompt “你是一个有帮助的助手。” human_message query # 3. 调用LLM response llm.invoke([ {“role”: “system”, “content”: system_prompt}, *conversation_history, {“role”: “user”, “content”: human_message} ]) # 4. 将LLM的回复封装成消息格式 ai_message {“role”: “assistant”, “content”: response.content} # 5. 返回要更新的State部分 # 注意我们只返回需要修改的字段。这里更新了消息历史。 return {“messages”: [ai_message]}这个节点做了几件事读取状态、准备数据、执行核心逻辑调用API、处理结果、最后返回状态更新。关键点在于节点函数只返回它想要修改的那部分State而不是完整的State。LangGraph的运行时负责将这些部分更新智能地合并到全局State中。2.3 Edge控制流程的方向盘如果Node是车站那么Edge边就是连接车站的轨道它决定了工作流的下一步该去哪里。边定义了基于当前State的条件逻辑。边主要分为两种普通边Fixed Edges无条件地将一个节点的输出连接到另一个节点的输入。这用于定义固定的、线性的执行顺序。条件边Conditional Edges根据State中的某个值或某个判断函数的返回值动态地决定下一个要执行的节点。这是实现循环、分支判断的关键。条件边是LangGraph强大灵活性的源泉。它允许你创建诸如“如果模型回答中包含不确定词汇则跳转到‘检索文档’节点否则直接结束”这样的逻辑。定义一个条件边通常需要创建一个路由函数。这个函数接收State返回一个字符串这个字符串就是下一个要执行的节点的名字。def should_retrieve_docs(state: State) - str: 路由函数判断是否需要检索文档 last_message state[“messages”][-1] llm_response last_message[“content”] # 简单的关键词判断逻辑实际应用中会更复杂 uncertainty_keywords [“我不确定”, “我不知道”, “根据我的知识”] if any(keyword in llm_response for keyword in uncertainty_keywords): return “retrieve_node” # 跳转到检索节点 else: return “end” # 结束流程在构建图时你会将这个路由函数与特定的边关联起来从而创建动态的工作流路径。3. 构建你的第一个LangGraph应用一个简单的对话代理理论说得再多不如动手搭一个。我们来构建一个最简单的、带条件判断的对话代理。它的逻辑是用户提问 - LLM尝试回答 - 如果LLM表示不确定则去检索知识库 - 基于检索结果重新生成答案。3.1 环境准备与依赖安装首先确保你的Python环境建议3.10以上并安装必要库。我们将使用langgraph、langchain-openai用于调用GPT以及langchain-community假设我们用一个简单的向量库模拟检索。pip install langgraph langchain-openai langchain-community为了模拟检索我们用一个内存中的FAISS向量存储和OpenAIEmbeddings。你需要在环境变量中设置你的OPENAI_API_KEY。import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.documents import Document # 设置API Key (请替换为你的key) os.environ[“OPENAI_API_KEY”] “your-api-key-here” # 初始化LLM和Embedding模型 llm ChatOpenAI(model“gpt-4o”, temperature0) embeddings OpenAIEmbeddings() # 创建一个模拟的知识库 docs [ Document(page_content“LangGraph是一个用于构建有状态、多参与者LLM应用的库。”), Document(page_content“LangChain提供了连接LLM与外部工具和数据的组件。”), Document(page_content“状态State在LangGraph中用于在节点间传递信息。”), ] vectorstore FAISS.from_documents(docs, embeddings) retriever vectorstore.as_retriever()3.2 定义State与节点函数根据我们的流程State需要包含消息历史、用户查询、检索到的文档和最终答案。from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AgentState(TypedDict): 定义我们代理的工作流状态 messages: Annotated[List[dict], add_messages] user_query: str retrieved_docs: Optional[List[Document]] final_answer: Optional[str]接下来定义三个节点函数call_llm_node: 直接调用LLM回答问题。retrieve_docs_node: 从知识库检索相关文档。generate_with_context_node: 基于检索到的文档让LLM生成最终答案。def call_llm_node(state: AgentState): 节点1直接调用LLM生成初始回答 query state[“user_query”] history state[“messages”] # 构建提示词 prompt f“请回答以下问题{query}。如果你非常确定答案请直接给出。如果你不确定请在回答中包含‘我不确定’或类似表述。” response llm.invoke([ *history, {“role”: “user”, “content”: prompt} ]) ai_msg {“role”: “assistant”, “content”: response.content} return {“messages”: [ai_msg]} def retrieve_docs_node(state: AgentState): 节点2根据用户查询检索相关文档 query state[“user_query”] docs retriever.invoke(query) # 检索相关文档 return {“retrieved_docs”: docs} def generate_with_context_node(state: AgentState): 节点3结合检索到的文档生成最终答案 query state[“user_query”] docs state[“retrieved_docs”] context “\n\n”.join([doc.page_content for doc in docs]) prompt f“请基于以下背景知识回答问题。\n背景知识{context}\n\n问题{query}” response llm.invoke([{“role”: “user”, “content”: prompt}]) ai_msg {“role”: “assistant”, “content”: response.content} return {“messages”: [ai_msg], “final_answer”: response.content}3.3 定义条件路由逻辑我们需要一个路由函数来检查LLM的初始回答是否包含不确定性从而决定是直接结束还是去检索文档。def route_after_initial_answer(state: AgentState) - str: 路由函数分析LLM初始回答决定下一步 last_message state[“messages”][-1] answer last_message[“content”].lower() # 判断是否包含不确定性表述 uncertainty_phrases [“i’m not sure”, “i don’t know”, “不确定”, “不了解”, “根据我的知识”] if any(phrase in answer for phrase in uncertainty_phrases): return “retrieve_docs” # 需要检索 else: # LLM看起来很确定将其回答作为最终答案 state[“final_answer”] last_message[“content”] return “end” # 直接结束3.4 组装图并运行现在把所有的节点和边组装成一张完整的图。from langgraph.graph import StateGraph, END # 1. 创建一个图并指定它使用的State类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“call_llm”, call_llm_node) workflow.add_node(“retrieve_docs”, retrieve_docs_node) workflow.add_node(“generate_final_answer”, generate_with_context_node) # 3. 设置入口点 workflow.set_entry_point(“call_llm”) # 4. 添加边包括条件边 # 从 call_llm 节点出来后根据路由函数决定去向 workflow.add_conditional_edges( “call_llm”, route_after_initial_answer, { “retrieve_docs”: “retrieve_docs”, # 如果返回“retrieve_docs”则跳转到对应节点 “end”: END # 如果返回“end”则直接结束工作流 } ) # 5. 添加固定边 # 从 retrieve_docs 节点无条件地连接到 generate_final_answer 节点 workflow.add_edge(“retrieve_docs”, “generate_final_answer”) # 从 generate_final_answer 节点无条件地连接到结束 workflow.add_edge(“generate_final_answer”, END) # 6. 编译图 app workflow.compile()至此你的第一个LangGraph应用就构建完成了你可以通过app.get_graph().draw_mermaid()来生成一个Mermaid图在支持的环境下可视化你的工作流。现在来运行它# 定义初始状态 initial_state: AgentState { “messages”: [], “user_query”: “LangGraph和LangChain有什么区别”, “retrieved_docs”: None, “final_answer”: None } # 运行工作流 final_state app.invoke(initial_state) print(“最终答案”, final_state[“final_answer”]) print(“完整消息历史”, final_state[“messages”])当你运行这段代码时工作流会启动。LLM会先尝试直接回答。由于这个问题需要精确对比LLM很可能会在回答中包含“根据我的知识”这类表述从而触发路由函数走向retrieve_docs节点。检索节点会从我们模拟的知识库中找到相关文档然后generate_final_answer节点会结合这些文档生成一个更准确的答案。最终这个答案会被保存在final_state[“final_answer”]中。4. 深入LangGraph高级特性子图、持久化与消息管理当你掌握了基础构建后LangGraph的一些高级特性能让你的应用变得更强大、更易管理。4.1 使用子图Subgraphs模块化复杂逻辑当一个工作流变得非常庞大时将所有节点都放在主图中会难以维护。子图允许你将一组相关的节点和边打包成一个独立的、可复用的单元。这就像编程中的函数让主图的结构变得清晰。假设我们上面例子中的“检索-生成”流程retrieve_docsgenerate_final_answer在很多地方都会用到我们可以将其封装成一个子图。from langgraph.graph import StateGraph def create_rag_subgraph(): 创建一个RAG检索增强生成子图 rag_graph StateGraph(AgentState) # 添加子图内部的节点函数定义同上 rag_graph.add_node(“retrieve”, retrieve_docs_node) rag_graph.add_node(“generate”, generate_with_context_node) # 定义子图内部的边 rag_graph.add_edge(“retrieve”, “generate”) rag_graph.set_entry_point(“retrieve”) rag_graph.set_finish_point(“generate”) # 设置子图的出口节点 return rag_graph.compile() # 在主图中使用子图 main_workflow StateGraph(AgentState) main_workflow.add_node(“call_llm”, call_llm_node) # 将子图作为一个“节点”添加到主图 rag_app create_rag_subgraph() main_workflow.add_node(“rag_process”, rag_app) # rag_app本身就是一个可调用的图 # ... 后续设置主图的边将“call_llm”连接到“rag_process”或“END”这样主图只需要关心“调用LLM”和“执行RAG流程”这两个高级步骤细节被隐藏在了子图中大大提升了代码的可读性和可复用性。4.2 状态持久化Persistence与长期记忆对于聊天机器人或多轮对话应用我们经常需要将会话状态保存下来以便用户下次回来时能继续对话。LangGraph内置了强大的持久化支持可以轻松地将State保存到数据库如SQLite、PostgreSQL或内存中。持久化的核心是Persistence接口。以下是一个使用SqliteSaver将状态保存到本地SQLite数据库的示例from langgraph.checkpoint.sqlite import SqliteSaver # 初始化一个SQLite检查点存储器 persister SqliteSaver.from_conn_string(“:memory:”) # 使用内存数据库也可用文件路径如“checkpoints.db” # 在编译图时传入persister app workflow.compile(checkpointerpersister) # 运行工作流时需要指定一个线程IDthread_id来标识唯一的会话 config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {“messages”: [], “user_query”: “你好”} # invoke会保存状态 result1 app.invoke(initial_state, configconfig) print(“第一次回答后状态已保存。”) # 模拟用户接着提问 new_state {“messages”: result1[“messages”], “user_query”: “我刚才问了什么”} # 再次调用LangGraph会从检查点恢复上一次的完整状态并在此基础上更新 result2 app.invoke(new_state, configconfig) print(“第二次回答它记得之前的对话”, result2[“messages”][-1][“content”])通过thread_id你可以为不同的用户或会话创建独立的、可恢复的状态流。这对于构建需要“长期记忆”的智能体至关重要。4.3 细粒度消息管理在复杂的对话流中你可能需要更精细地控制消息列表。langgraph.graph.message模块提供了一些工具函数除了我们之前用到的add_messages还有trim_messages: 根据Token数量、消息条数或自定义函数来修剪历史消息防止上下文窗口溢出。AnyMessage,HumanMessage,AIMessage等类型化的消息类提供更好的类型提示。例如实现一个只保留最近5轮对话的消息管理策略from langgraph.graph.message import trim_messages class StateWithTrimmedMessages(TypedDict): messages: Annotated[List[dict], add_messages, trim_messages(max_messages10)] # 保留最多10条消息 # ... 其他字段这样每次更新messages字段时trim_messages归并器会自动确保列表长度不超过10条无需在节点函数中手动处理。5. 常见问题排查与性能优化实战在实际开发中你肯定会遇到各种问题。下面是我在项目中踩过的一些坑和总结的解决方案。5.1 问题一State更新不符合预期症状节点返回了更新字典但最终State中的值没有被正确修改或者被意外覆盖。根因与排查归并器Reducer冲突这是最常见的原因。State中每个字段的Annotated第二个参数就是它的归并器。如果你为List字段设置了add_messages但节点返回的更新值不是一个List[dict]或者格式不对归并就会失败。对于简单字段如str,int默认归并器是operator.setitem后写入的覆盖先前的。如果你希望是累加需要自定义归并器。节点返回了完整的State节点函数应该只返回它想修改的字段字典。如果你错误地返回了完整的State对象可能会干扰其他字段。解决方案打印调试在每个节点函数的开头和结尾打印state和return的值。检查归并器确保你理解每个字段的归并逻辑。对于复杂合并可以自定义归并函数。使用类型提示严格定义State的TypedDict或Pydantic模型IDE和mypy能帮你提前发现许多类型错误。5.2 问题二条件边Conditional Edge不生效症状工作流没有按照预想的分支执行总是走固定路径或报错。根因与排查路由函数返回值与映射不匹配add_conditional_edges的第三个参数是一个映射字典。你的路由函数返回的字符串必须是这个字典的一个键。如果返回了字典中不存在的字符串图会找不到下一个节点而报错。路由函数逻辑错误路由函数的判断条件可能过于严格或宽松导致预期外的分支选择。仔细检查路由函数中对State的读取和判断逻辑。State中依赖的数据未就绪路由函数在Node A之后执行它读取的State应该是Node A更新后的。确保Node A已经将路由判断所需的数据正确地写入了State。解决方案在路由函数内增加详细的日志打印判断条件和返回值。使用app.get_graph().draw_mermaid()可视化你的图确认条件边的指向是否正确。简化测试先让路由函数返回一个固定值测试分支是否能正确跳转再逐步完善判断逻辑。5.3 问题三图编译或执行速度慢症状对于简单的逻辑app.invoke()执行时间过长。根因与排查节点中的阻塞操作最常见的瓶颈是节点内部同步调用了耗时的I/O操作如网络请求LLM API调用、数据库查询、大文件读写等。在同步代码中这些操作会阻塞整个线程。State过于庞大如果State中存储了非常大的对象如整个文档内容、大型列表在节点间序列化/反序列化传递时会消耗额外时间。未利用并发如果图中有多个可以并行执行的节点例如同时调用两个不同的API使用默认的同步执行方式会串行运行浪费了时间。解决方案异步化推荐将节点函数定义为async def并在其中使用await调用异步客户端。LangGraph完全支持异步执行。from langchain_openai import AsyncChatOpenAI async_llm AsyncChatOpenAI(model“gpt-4o”) async def async_call_llm_node(state: State): response await async_llm.ainvoke(...) return {“messages”: [response]}然后用app.ainvoke()来异步调用整个图效率会大幅提升。精简State只把必要的数据放在State中。对于大型数据可以考虑只存储引用如ID、路径在节点内部按需加载。分析性能使用Python的cProfile或line_profiler工具定位具体的耗时函数。5.4 调试与可视化技巧使用stream进行逐步调试app.invoke()是一次性执行到底。使用app.stream()可以让你以流式方式获取每个节点执行前后的状态非常适合调试。for step in app.stream(initial_state, configconfig): node_name step[0] # 执行的节点名 node_output step[1] # 节点执行后的状态 print(f“节点 [{node_name}] 执行完毕。当前State: {node_output}”)Mermaid可视化在Jupyter Notebook或支持Mermaid的Markdown查看器中app.get_graph().draw_mermaid()可以生成工作流的可视化图一目了然地看清所有节点和边的结构。打印编译后的图结构print(app.get_graph())或print(app.get_graph().to_json())可以输出图的JSON表示用于深入分析。6. LangGraph与LangChain、Dify的定位辨析社区里经常有人混淆LangGraph、LangChain以及Dify这类平台的关系。理解它们的定位差异能帮你更好地选择工具。LangChain可以看作是**“基础设施层”或“组件库”**。它提供了与各种LLM、嵌入模型、向量数据库、工具等进行交互的标准接口LLMRetrieverTool。它的核心价值在于标准化和集成让你不用为每个模型或数据库写不同的适配代码。Chain和Agent是其上层的两种编排模式但相对简单。LangGraph是**“复杂编排层”。它专注于解决当多个步骤之间存在复杂依赖、循环、状态共享时的流程控制问题。它利用“图”这一计算机科学中的经典模型提供了声明式的方式来定义工作流。它底层可以使用LangChain的组件如ChatOpenAI,FAISS但它的核心价值是编排逻辑本身**。Dify/AutoGen等平台属于**“应用层”或“低代码平台”**。它们提供了图形化界面让你可以通过拖拽等方式构建AI工作流并集成了用户管理、知识库管理、API部署、监控等开箱即用的功能。它们可能底层使用了LangChain或LangGraph的技术但目标用户是更希望关注业务逻辑而非代码的开发者或产品经理。关系类比如果你想快速搭建一个包含前端、后端、数据库的完整AI应用选Dify这类平台。如果你想用代码灵活地连接各种AI模型和数据源构建一个中等复杂度的流程LangChain是很好的起点。当你用LangChain构建的Agent逻辑变得无比复杂、难以维护或者你需要精细控制带有状态循环和分支的复杂工作流时就该引入LangGraph了。它们不是互斥的而是可以协同工作。一个典型的模式是用LangChain的组件处理与外部服务的交互用LangGraph的图来编排这些组件之间的复杂执行逻辑。