1. 项目概述从LangChain到LangGraph的范式演进如果你在过去一年里深度参与过基于大语言模型的应用开发那么“LangChain”这个名字对你来说一定不陌生。它几乎成了构建AI应用的事实标准框架通过其强大的链式编排能力让我们能够将大模型、工具、记忆和外部数据源串联起来创造出功能丰富的智能体。然而随着我们构建的应用越来越复杂从简单的问答机器人到需要多轮决策、状态管理和分支协作的复杂工作流单纯依赖“链”的线性思维开始显得捉襟见肘。这正是LangGraph诞生的背景。LangGraph并非要取代LangChain而是对其核心范式的一次重要升级和补充。你可以把它理解为LangChain生态系统中的“工作流引擎”或“有状态编排层”。如果说LangChain的Chain是定义了一条从A到B的固定流水线那么LangGraph则允许你绘制一张完整的流程图其中包含条件判断、循环、并行执行以及持久化的状态。它引入了“图”的计算模型将应用中的每个步骤节点和步骤之间的流转逻辑边清晰地定义出来特别适合构建具有复杂逻辑、长期记忆和人类参与回路的智能体系统。简单来说当你需要构建一个能处理多轮对话、根据中间结果动态调整执行路径、或者协调多个“子智能体”协作完成任务的AI应用时LangGraph就是你一直在寻找的工具。它让智能体的行为更像一个拥有明确规划和状态感知的“智能体”而不仅仅是一个反应式的工具调用链。2. 核心设计理念与架构拆解2.1 有状态图计算智能体的“大脑”模型LangGraph最核心的设计理念是有状态图计算。这听起来有点学术但理解它对用好这个框架至关重要。在传统的LangChain链中数据或者说“上下文”通常是以字典的形式在链的各个环节间传递。虽然方便但对于需要记住之前发生了什么、并根据历史做决策的复杂场景管理起来就很麻烦。LangGraph将整个系统的运行抽象为一个对“状态”的持续读写和更新过程。这个“状态”是一个定义良好的字典State它包含了应用运行所需的所有信息例如用户的输入、大模型的回复、工具调用的结果、历史对话记录、乃至任何你自定义的变量。图中的每个节点Node都是一个函数它接收当前的完整状态执行一些操作比如调用大模型、运行工具然后返回一个更新后的状态字典。LangGraph的运行时引擎负责将更新后的状态传递给下一个节点。这种设计带来了几个关键优势状态集中管理所有中间数据和历史都存储在同一个状态对象里查询和追溯变得非常清晰。自然的循环与分支基于状态的判断来决定下一步走向哪个节点是实现循环比如“直到满足条件才退出”和条件分支比如“如果工具调用失败则转向人工审核节点”的天然方式。可持久化与可恢复由于整个系统的快照就是当前的状态你可以轻松地将状态保存到数据库并在之后某个时间点精确地恢复执行这对于需要长时间运行或支持异步交互的智能体来说至关重要。2.2 核心组件详解State、Node、Edge要构建一个LangGraph应用你需要理解三个核心组件状态State、节点Node和边Edge。状态State通常通过定义一个TypedDict或Pydantic模型来声明。这相当于为你的智能体定义了一张数据表格的 schema。例如一个客服智能体的状态可能包含user_input,chat_history,agent_scratchpad智能体思考过程,knowledge_base_results等字段。LangGraph使用注解来定义每个字段的更新方式比如是覆盖、追加还是其他自定义合并逻辑。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 用户的最新问题 input: str # 对话历史使用add_messages注解实现消息列表的追加 messages: Annotated[List, add_messages] # 从知识库检索到的文档片段 context: List[str] # 智能体决定要采取的动作 next_action: str节点Node节点是执行具体工作的函数。它接收一个状态字典返回一个更新后的状态字典。一个节点可以做任何事情调用LLM、执行代码、查询数据库、调用API。关键在于节点只关心它需要读写状态的哪些部分。例如一个“检索”节点可能只读取state[“input”]然后调用向量数据库最后将结果写入state[“context”]。def retrieve_node(state: AgentState): # 从状态中获取用户问题 question state[“input”] # 调用检索工具例如向量数据库 docs retriever.invoke(question) # 返回更新后的状态仅更新context字段 return {“context”: docs}边Edge边决定了流程的走向。这是LangGraph实现复杂逻辑的魔法所在。边分为两种条件边Conditional Edge根据当前状态的某个值决定下一个节点是谁。这通常通过一个路由函数实现。普通边Regular Edge无条件地指向下一个节点。通过组合节点和边你就能构建出包含if-else、while循环等复杂逻辑的工作流图。2.3 与LangChain的共生关系理解LangGraph和LangChain的关系能帮助你更好地在项目中定位它们。LangGraph深度集成在langchain包中from langgraph import …它复用并强化了LangChain的许多核心概念LCELLangChain Expression Language你依然可以使用LCEL来优雅地组合链并将这些链作为LangGraph图中的一个节点。这让你能复用大量现有的LangChain生态组件。Runnable 协议LangGraph的节点函数本质上是符合Runnable协议的这意味着它们可以无缝嵌入到现有的LangChain生态中。智能体Agent的重新定义在LangGraph的范式下智能体不再是一个特殊的AgentExecutor对象而是一个由“LLM调用节点”、“工具执行节点”和“路由逻辑边”构成的图。这提供了对智能体决策过程前所未有的透明度和控制力。简而言之LangChain提供了丰富的“乐高积木”模型、工具、检索器而LangGraph提供了搭建复杂动态“建筑结构”的蓝图和粘合剂。3. 从零构建一个LangGraph智能体以研究助手为例理论说得再多不如亲手构建一个。让我们来实现一个相对复杂的“研究助手”智能体。它的工作流程是接收一个研究主题先联网搜索最新信息然后根据搜索内容撰写一份初步报告最后自动检查报告中的事实性陈述并为有疑问的部分添加引用备注。这个过程涉及条件判断和多个步骤的循环。3.1 定义状态与工具首先定义智能体的状态。这个状态需要贯穿整个研究流程。from typing import TypedDict, Annotated, List, Optional from langchain_core.messages import BaseMessage, HumanMessage from langgraph.graph.message import add_messages class ResearchState(TypedDict): # 用户提出的研究主题 topic: str # 对话/指令历史 messages: Annotated[List[BaseMessage], add_messages] # 网络搜索得到的内容 search_results: List[str] # 撰写的报告草稿 draft: str # 需要核实的事实列表事实陈述 来源 facts_to_check: List[tuple[str, str]] # 最终带引用的报告 final_report: Optional[str]接下来装备智能体所需的工具。我们需要一个搜索工具和一个用于检查事实/引用的工具这里用一个大模型调用模拟。from langchain_community.tools import TavilySearchResults from langchain_openai import ChatOpenAI # 初始化工具和模型 search_tool TavilySearchResults(max_results3) llm ChatOpenAI(model“gpt-4-turbo-preview”) def fact_check_tool(claim: str, context: str) - str: “”“模拟一个事实检查工具。在实际应用中这里可以接入专业的Fact-Check API。”“” prompt f”请判断以下陈述在提供的上下文中是否得到完全支持并指出支持它的原文依据如果有的话\n陈述{claim}\n上下文{context}” response llm.invoke(prompt) return response.content3.2 构建图节点搜索、撰写、检查我们将工作流分解为四个主要节点。节点1搜索节点search_node这个节点负责执行网络搜索。def search_node(state: ResearchState): topic state[“topic”] print(f”[搜索节点] 正在搜索主题{topic}”) # 调用搜索工具 results search_tool.invoke({“query”: topic}) # 提取搜索结果的文本内容 search_contents [result[“content”] for result in results] return {“search_results”: search_contents}节点2撰写报告节点draft_node这个节点基于搜索结果为研究主题撰写初稿。def draft_node(state: ResearchState): topic state[“topic”] search_content “\n\n”.join(state[“search_results”]) print(f”[撰写节点] 基于搜索内容撰写‘{topic}’的初稿...”) prompt f”你是一个专业的研究员。请根据以下关于‘{topic}’的网络搜索信息撰写一份结构清晰、内容全面的研究报告草稿。报告应包含引言、核心发现和总结。\n\n搜索信息\n{search_content}” response llm.invoke(prompt) draft response.content # 一个简单的启发式方法从草稿中提取可能的事实陈述以句号分割过滤短句 sentences [s.strip() for s in draft.split(‘.’) if len(s.strip()) 20] # 这里简化处理将前3个长句作为待检查事实并关联其来源用搜索结果的第一个作为示例来源 facts [(sent, state[“search_results”][0]) for sent in sentences[:3]] if state[“search_results”] else [] return {“draft”: draft, “facts_to_check”: facts}节点3事实检查节点fact_check_node这个节点对草稿中的存疑事实进行核查。def fact_check_node(state: ResearchState): facts state[“facts_to_check”] checked_notes [] print(f”[检查节点] 正在核查 {len(facts)} 项事实陈述...”) for claim, source_context in facts: result fact_check_tool(claim, source_context) checked_notes.append(f”- 陈述{claim}\n 检查结果{result}”) notes_str “\n”.join(checked_notes) return {“facts_to_check”: []} # 清空待检查列表表示已处理节点4生成最终报告节点finalize_node这个节点综合草稿和事实检查备注生成最终报告。def finalize_node(state: ResearchState): draft state[“draft”] # 在实际中这里会整合fact_check_node的输出。本例中我们直接添加一个说明部分。 print(“[终稿节点] 生成最终报告...”) final_report f”{draft}\n\n---\n**备注**报告中的部分陈述已通过初步信息源核对建议对关键数据进一步溯源。” return {“final_report”: final_report}3.3 编排图结构与条件路由现在我们用StateGraph把这些节点和逻辑连接起来。from langgraph.graph import StateGraph, END # 1. 创建图并指定状态类型 workflow StateGraph(ResearchState) # 2. 添加节点 workflow.add_node(“search”, search_node) workflow.add_node(“draft”, draft_node) workflow.add_node(“fact_check”, fact_check_node) workflow.add_node(“finalize”, finalize_node) # 3. 设置入口点 workflow.set_entry_point(“search”) # 4. 添加边定义流程 workflow.add_edge(“search”, “draft”) # 搜索完后直接撰写 workflow.add_edge(“finalize”, END) # 生成最终报告后结束 # 5. 关键添加条件边。在撰写草稿后根据是否有待检查的事实来决定下一步。 def should_check_facts(state: ResearchState) - str: “”“路由函数判断是否需要进入事实检查环节。”“” if state[“facts_to_check”] and len(state[“facts_to_check”]) 0: return “fact_check” else: return “finalize” workflow.add_conditional_edges( “draft”, # 源节点 should_check_facts, # 路由判断函数 {“fact_check”: “fact_check”, “finalize”: “finalize”} # 目标映射 ) workflow.add_edge(“fact_check”, “finalize”) # 检查完后去生成终稿 # 6. 编译图 app workflow.compile()3.4 运行与可视化现在我们可以运行这个研究助手了。# 准备初始状态 initial_state {“topic”: “2024年人工智能在医疗诊断领域的最新进展”, “messages”: []} # 运行图 final_state app.invoke(initial_state, config{“recursion_limit”: 50}) print(“\n 研究完成 ”) print(f”最终报告长度{len(final_state[‘final_report’])} 字符”) print(“报告预览”, final_state[‘final_report’][:500], “...”)LangGraph还有一个非常强大的功能就是可以将你定义的图可视化出来这对于调试和理解复杂工作流至关重要。# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法直接显示可以保存到文件 app.get_graph().draw_mermaid_png(output_file_path“research_workflow.png”)生成的流程图会清晰地展示出从search到draft然后根据条件分支到fact_check或finalize的完整路径让你对智能体的决策流程一目了然。4. 高级模式与实战技巧4.1 人工监督与中断Interrupt和Human-in-the-Loop对于高风险或关键业务流程让人类参与审核是刚需。LangGraph通过interrupt机制优雅地支持了这一点。你可以在图中任何一个节点之后设置中断将执行暂停等待外部输入比如用户在UI上点击“批准”或修改某些内容然后再继续执行。from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import ToolNode, tools_condition # 使用检查点存储器这是支持中断和恢复的基础 memory MemorySaver() workflow StateGraph(…, checkpointermemory) # … 添加节点和边 … # 在某个节点后配置中断 workflow.add_node(“human_review”, human_review_node) # 假设这是一个等待人工输入的节点 workflow.add_edge(“some_node”, “human_review”) workflow.add_edge(“human_review”, “next_node”) # 当调用到human_review节点时图的执行会暂停并返回一个包含waiting_for_input标记的结果。 # 你的前端应用可以据此展示一个审核界面。当人工操作完成后你再通过app.update_state()传入新的状态并继续执行。注意中断功能强烈依赖于检查点Checkpointer。检查点不仅记录了当前状态还记录了图的执行位置使得暂停和恢复成为可能。在生产环境中你需要将MemorySaver替换为如PostgresSaver或RedisSaver这样的持久化存储。4.2 多智能体协作MessagesState与子图LangGraph非常适合构建多智能体系统其中不同的智能体扮演不同角色通过消息传递进行协作。MessagesState是一个预定义的状态类型专门为基于消息的对话场景设计。你可以为每个智能体定义一个子图Subgraph然后将这些子图作为主图的节点。主图负责在不同智能体之间路由消息。from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import create_react_agent # 定义状态使用预建的MessagesState它自动处理消息列表的追加 class MultiAgentState(MessagesState): current_agent: str # 记录当前该谁发言 # 创建两个不同专长的智能体 analyst_agent create_react_agent(llm, tools[data_analysis_tool]) writer_agent create_react_agent(llm, tools[writing_assistant_tool]) # 创建主图 workflow StateGraph(MultiAgentState) # 将智能体子图添加为主图的节点 workflow.add_node(“analyst”, analyst_agent) workflow.add_node(“writer”, writer_agent) # 定义一个路由函数根据消息内容或规则决定下一个发言的智能体 def router(state: MultiAgentState) - str: last_message state[“messages”][-1] if “数据分析” in last_message.content: return “analyst” else: return “writer” workflow.add_conditional_edges(“analyst”, router) workflow.add_conditional_edges(“writer”, router) workflow.set_entry_point(“analyst”) # 从分析师开始 app workflow.compile()在这个模式下analyst和writer节点各自都是一个完整的、拥有推理和工具调用能力的智能体子图。主图像一个调度员根据对话内容决定将任务交给谁实现了智能体间的协同工作。4.3 性能优化与检查点策略当你的图变得复杂或需要长时间运行时性能和资源管理就变得重要。选择性状态更新在节点函数中只返回你真正修改了的状态字段。这可以减少序列化和传递的数据量。LangGraph的状态合并机制非常高效但明确返回{“updated_field”: new_value}仍然是最佳实践。检查点配置检查点虽然强大但频繁保存会影响性能。你可以在编译图时配置检查点策略。from langgraph.checkpoint import MemorySaver checkpointer MemorySaver( serde“json”, # 序列化方式 # at… 可以配置在哪些节点之后自动保存检查点 ) app workflow.compile(checkpointercheckpointer)对于非常长的运行链可以考虑只在关键的决策点或外部调用如工具调用之后保存检查点而不是每一步都保存。异步支持LangGraph原生支持异步节点函数。如果你的节点涉及大量I/O操作如网络请求、数据库查询使用async def定义节点函数并用ainvoke调用图可以显著提升吞吐量。async def async_search_node(state: State): result await async_http_client.get(…) return {“data”: result} # 异步调用 final_state await app.ainvoke(initial_state)5. 常见问题与调试指南在实际使用LangGraph的过程中你可能会遇到一些典型问题。以下是一些排查思路和解决方案。5.1 状态更新不符合预期问题节点修改了状态但下游节点读取到的值还是旧的。排查首先确认你的状态字段注解是否正确。例如对于列表追加必须使用Annotated[List, add_messages]或operator.add。如果错误地使用了默认的覆盖方式更新就会丢失。在节点函数中多使用print或日志输出当前状态和返回的状态确认更新逻辑。检查是否有多个节点在并发修改同一个字段虽然LangGraph默认是顺序执行但在复杂路由下需理清顺序。5.2 图陷入无限循环问题智能体在一个循环里出不来直到达到递归限制。排查检查条件边Conditional Edge的路由逻辑这是最常见的循环来源。确保你的路由函数在所有可能的状态下都能最终导向END或一个不指向自身的节点。可以打印路由函数的输入和返回值来调试。设置递归限制在app.invoke()时传入config{“recursion_limit”: N}这是一个安全网防止无限循环耗尽资源。可视化你的图使用app.get_graph().draw_mermaid()生成图表直观检查是否存在循环路径。确保每个循环都有退出条件。5.3 工具调用失败或LLM响应格式错误问题在智能体节点中工具调用抛出异常或者LLM返回的内容无法被解析为工具调用。排查隔离测试工具首先确保你的工具函数在LangGraph上下文之外能正常工作。包装工具调用在节点函数内部使用try…except包裹工具调用和LLM调用并将错误信息捕获后写入状态以便后续节点进行错误处理或重试。优化Prompt对于ReAct模式的智能体LLM输出格式错误通常与Prompt指令不清晰有关。确保你的系统消息明确要求以Action:和Action Input:的格式输出。使用LangChain的ToolsAgent或create_react_agent等预建组件可以减少这类问题。5.4 检查点恢复失败问题保存的检查点无法加载或者恢复后状态不对。排查序列化兼容性确保状态中所有对象都是可序列化的如JSON或Pickle。避免在状态中存储数据库连接、文件句柄等不可序列化的对象。复杂对象应存储其引用ID在节点中重新初始化。检查点键冲突每个运行图实例的线程IDthread_id必须是唯一的。如果你在恢复时使用了重复的thread_id可能会加载到错误的状态。确保你的thread_id管理逻辑正确。存储后端一致性如果你使用了外部数据库如Postgres作为检查点存储确保图编译时使用的检查点序列化/反序列化配置与存储时一致。5.5 调试与日志记录实战技巧使用print进行快速调试在每个节点的开始和结束打印状态的关键字段这是最直接的方法。利用langsmith进行追踪如果你使用LangSmithLangGraph的每一步执行都会自动生成详细的追踪记录包括每个节点的输入输出、耗时、工具调用详情等这是生产环境调试的利器。状态快照在关键节点后手动将状态字典保存到文件或日志系统便于事后分析。import json def some_node(state): # … 业务逻辑 … with open(f”state_snapshot_{node_name}.json”, “w”) as f: json.dump(state, f, indent2, defaultstr) # defaultstr处理不可序列化对象 return new_state图结构验证在编译图之前仔细检查边的连接特别是条件边的目标映射字典确保所有引用的节点名都已正确添加。LangGraph将AI应用的开发从“链式思维”提升到了“图式思维”为你处理复杂、有状态、多分支的工作流提供了强大而优雅的抽象。初学时会觉得概念稍多但一旦理解了State、Node、Edge和Checkpoint这几个核心支柱构建复杂智能体就会变得像画流程图一样直观。从简单的线性链升级到LangGraph就像是给你的AI应用装上了规划和决策的“大脑”让它能真正胜任那些需要记忆、思考和判断的复杂任务。