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 # 方式一使用TypedDictLangGraph经典方式 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_factorylist) # 专用于消息累加 question: str None intermediate_steps: List[dict] Field(default_factorylist) # 记录工具调用结果 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(modelgpt-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最强大也最灵活的部分。边分为两种普通边Fixed Edges无条件地从节点A指向节点B。条件边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_factorylist) # 模型生成的答案草稿 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_factorylist)接下来我们定义四个核心节点# 节点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(modelgpt-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(modelgpt-3.5-turbo, temperature0) 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(modelgpt-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, subgraphsTrue): # subgraphsTrue可展示子图细节 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_factorylist) 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_questionstate.user_query, current_questionstate.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( valuesrag_final_state, # config... 可以传递配置 ) } main_workflow.add_node(rag_processor, call_rag_subgraph_node) # ... 主图的其他节点和边4.2 集成长期记忆Long-term MemoryLangGraph本身不直接提供开箱即用的长期记忆存储但它与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_factorylist) 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你需要考虑以下几点线程安全与并发app.invoke()通常是线程安全的因为执行过程是纯函数对状态的转换。但要确保你提供的工具如数据库连接、第三方API客户端本身是线程安全的或为每个请求创建新实例。状态序列化如果你使用Pydantic定义State它天然支持model_dump()和model_validate()进行JSON序列化与反序列化。这对于将检查点checkpoint保存到数据库、实现异步或持续的工作流至关重要。检查点CheckpointsLangGraph支持检查点机制允许你在图执行到某个节点后暂停并将完整状态保存下来后续可以从此状态恢复执行。这对于处理耗时极长的任务或等待外部人工输入的任务非常有用。# 这是一个高级特性涉及配置和持久化层 from langgraph.checkpoint import MemorySaver checkpoint_memory MemorySaver() app workflow.compile(checkpointercheckpoint_memory) # 第一次执行生成一个线程ID config {configurable: {thread_id: user_123_session_1}} result1 app.invoke(initial_state, configconfig) # 模拟一段时间后从上次中断的地方继续执行 # LangGraph会基于thread_id自动加载最新的检查点状态 result2 app.invoke(new_input_state, configconfig)日志与追踪利用app.stream()输出每一步的状态变化这是调试的利器。在生产环境中可以将这些日志结构化后发送到如OpenTelemetry、LangSmith等可观测性平台方便监控每个节点的耗时、输入输出和异常。从LangChain到LangGraph本质上是从“编排组件”到“编排智能”的思维升级。LangChain帮你把工具准备好而LangGraph帮你设计智能体如何使用这些工具去完成复杂的、非线性的任务。它带来的最大好处是代码即架构——你的应用流程图几乎就是你的代码结构这使得复杂系统的设计、理解和维护成本大大降低。刚开始接触时你可能会觉得多了一层抽象有点绕但一旦适应了这种“图思维”在构建具备复杂决策和循环能力的AI应用时你会发现自己再也回不去了。 
长河君 · 2026/8/15 12:45:33 · 阅读 1.2k 从LangChain到LangGraph：构建复杂AI工作流的图计算范式
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 # 方式一使用TypedDictLangGraph经典方式 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_factorylist) # 专用于消息累加 question: str None intermediate_steps: List[dict] Field(default_factorylist) # 记录工具调用结果 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(modelgpt-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最强大也最灵活的部分。边分为两种普通边Fixed Edges无条件地从节点A指向节点B。条件边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_factorylist) # 模型生成的答案草稿 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_factorylist)接下来我们定义四个核心节点# 节点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(modelgpt-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(modelgpt-3.5-turbo, temperature0) 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(modelgpt-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, subgraphsTrue): # subgraphsTrue可展示子图细节 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_factorylist) 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_questionstate.user_query, current_questionstate.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( valuesrag_final_state, # config... 可以传递配置 ) } main_workflow.add_node(rag_processor, call_rag_subgraph_node) # ... 主图的其他节点和边4.2 集成长期记忆Long-term MemoryLangGraph本身不直接提供开箱即用的长期记忆存储但它与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_factorylist) 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你需要考虑以下几点线程安全与并发app.invoke()通常是线程安全的因为执行过程是纯函数对状态的转换。但要确保你提供的工具如数据库连接、第三方API客户端本身是线程安全的或为每个请求创建新实例。状态序列化如果你使用Pydantic定义State它天然支持model_dump()和model_validate()进行JSON序列化与反序列化。这对于将检查点checkpoint保存到数据库、实现异步或持续的工作流至关重要。检查点CheckpointsLangGraph支持检查点机制允许你在图执行到某个节点后暂停并将完整状态保存下来后续可以从此状态恢复执行。这对于处理耗时极长的任务或等待外部人工输入的任务非常有用。# 这是一个高级特性涉及配置和持久化层 from langgraph.checkpoint import MemorySaver checkpoint_memory MemorySaver() app workflow.compile(checkpointercheckpoint_memory) # 第一次执行生成一个线程ID config {configurable: {thread_id: user_123_session_1}} result1 app.invoke(initial_state, configconfig) # 模拟一段时间后从上次中断的地方继续执行 # LangGraph会基于thread_id自动加载最新的检查点状态 result2 app.invoke(new_input_state, configconfig)日志与追踪利用app.stream()输出每一步的状态变化这是调试的利器。在生产环境中可以将这些日志结构化后发送到如OpenTelemetry、LangSmith等可观测性平台方便监控每个节点的耗时、输入输出和异常。从LangChain到LangGraph本质上是从“编排组件”到“编排智能”的思维升级。LangChain帮你把工具准备好而LangGraph帮你设计智能体如何使用这些工具去完成复杂的、非线性的任务。它带来的最大好处是代码即架构——你的应用流程图几乎就是你的代码结构这使得复杂系统的设计、理解和维护成本大大降低。刚开始接触时你可能会觉得多了一层抽象有点绕但一旦适应了这种“图思维”在构建具备复杂决策和循环能力的AI应用时你会发现自己再也回不去了。