1. 项目概述从LangChain到LangGraph的范式跃迁如果你在过去一年里深度参与过AI应用开发尤其是基于大语言模型LLM的智能体Agent构建那么“LangChain”这个名字对你来说一定如雷贯耳。它几乎成了连接LLM与外部工具、数据源的“标准胶水”。然而随着我们试图构建的智能体逻辑越来越复杂——从简单的问答机器人到需要多步骤决策、具备长期记忆、甚至能并行执行多个任务的“数字员工”——我们开始感到力不从心。用LangChain的链Chain和代理Agent来编排这些复杂、有状态、带循环的工作流代码会迅速变得臃肿且难以维护状态管理更是成为一场噩梦。这正是langchain-ai/langgraph诞生的背景。它不是LangChain的替代品而是一个强有力的补充和进化。简单来说LangGraph是一个用于构建有状态、多参与者Agent应用的库。它将复杂的工作流建模为图Graph其中节点代表处理单元可以是LLM调用、工具执行或自定义函数边则定义了控制流下一步该执行哪个节点。这种范式让我们能够清晰地描述“如果这一步成功则进入审核节点如果失败则跳转到修复节点并循环重试”这样的逻辑并且以非常直观、可调试的方式实现。对我而言从传统的链式思维切换到图式思维就像从编写线性脚本升级到了设计电路板。它解决的核心痛点有三个复杂流程的可视化与可维护性、应用状态的显式管理、以及对人类反馈Human-in-the-loop等异步事件的原生支持。无论你是想构建一个能自动处理客服工单的智能体集群还是一个需要反复推敲写作大纲的创作助手LangGraph都提供了那个缺失的、工业级的编排框架。2. 核心概念拆解图、状态与循环在深入代码之前我们必须先吃透LangGraph的几个核心抽象。这能让你在后续设计工作流时思路清晰事半功倍。2.1 图Graph工作流的骨架在LangGraph中一切皆图。一个图由两部分组成节点Nodes这是实际“干活”的地方。一个节点可以是一个简单的函数一个LangChain链一个工具调用或者另一个子图。关键是其输入和输出必须符合整个图所定义的状态State结构。边Edges它决定了工作流的走向。边分为两种起始边Start Edge指定哪个节点最先执行。条件边Conditional Edges这是LangGraph的灵魂。它允许你根据前一个节点的输出结果动态决定下一个要执行的节点。这实现了if-else和switch的分支逻辑。这种图的模型特别适合描述那些非线性的业务流程。例如一个内容审核工作流生成内容节点 - 根据内容安全评分条件边决定是流向直接发布节点还是人工审核节点或是拒绝并告警节点。2.2 状态State应用的记忆体这是LangGraph与简单链最根本的区别。在传统的链式调用中信息通常以字典形式在链间传递缺乏统一的、类型化的管理。LangGraph引入了状态State的概念。你可以把状态想象成一个共享的白板或者一个结构化的数据库。它在图执行开始时被初始化并随着每个节点的执行而被更新。每个节点都读取这个状态的一部分并可能修改它。状态的结构由一个类似Pydantic的模型定义明确了每个字段的类型和用途。例如一个写作助手的状态可能定义为from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages class State(TypedDict): # 对话历史由LangGraph内置的add_messages操作符管理 messages: Annotated[List, add_messages] # 当前生成的文章大纲 outline: str # 已完成的章节列表 completed_sections: List[str] # 当前迭代次数用于防止无限循环 iteration: intAnnotated和add_messages是LangGraph提供的语法糖用于简化对消息列表这种常见结构的追加操作。状态管理让数据流变得清晰、可预测并且易于调试。2.3 循环Cycles与持久化让智能体“活”起来图结构天然支持循环——一个节点的边可以指回之前的节点。这使得构建循环代理ReAct, Reflexion等模式变得极其简单。例如一个代码修复智能体可以循环执行分析错误-尝试修复-运行测试如果测试失败则带着新的错误信息再次进入分析错误节点直到成功或达到最大重试次数。更强大的是LangGraph支持检查点Checkpointing功能。图的执行状态可以被持久化到数据库如SQLite、Postgres。这意味着你可以暂停一个长时间运行的工作流比如一个需要等待人工审批的流程几天后再从完全相同的状态恢复执行。这对于构建需要与人类异步交互的、长时间运行的业务流程至关重要。3. 实战构建一个带自我审查的写作智能体理论说得再多不如动手一试。我们来构建一个相对复杂的智能体一个能根据主题生成文章大纲并对自己生成的每个章节进行批判性审查和重写的写作助手。这个工作流将充分展示条件边、状态管理和循环的魅力。3.1 环境准备与状态定义首先安装必要的包并定义我们的应用状态。pip install langgraph langchain-openai假设我们使用OpenAI的模型。import os from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages # 定义状态结构 class WritingState(TypedDict): 写作智能体的共享状态 # 用户输入的主题 topic: str # 对话历史用于记录LLM调用 messages: Annotated[List, add_messages] # 生成的文章大纲字符串列表如 [引言, 背景, 论点1, 结论] outline: List[str] # 当前正在处理的章节索引 current_section_index: int # 已完成的章节内容 completed_sections: List[str] # 审查意见 review_feedback: str # 重写次数计数器 rewrite_count: int这个状态定义了工作流中需要流转的所有信息。add_messages操作符会自动帮我们处理消息列表的追加非常方便。3.2 构建工作流节点节点就是普通的Python函数但它第一个参数必须是状态返回值必须是一个字典用于更新状态而不是替换整个状态。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4-turbo-preview) def generate_outline(state: WritingState) - dict: 节点1根据主题生成大纲 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field class Outline(BaseModel): sections: List[str] Field(description文章大纲的章节标题列表) parser JsonOutputParser(pydantic_objectOutline) prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的写作助手。请根据用户主题生成一份逻辑清晰的文章大纲。只输出JSON。), (human, 文章主题{topic}) ]) chain prompt | llm | parser # 调用链获取结构化大纲 result chain.invoke({topic: state[topic]}) # 更新状态存入大纲并初始化当前章节索引为0从第一个章节开始写 return { outline: result[sections], current_section_index: 0, messages: [{role: assistant, content: f已生成大纲{result[sections]}}] } def write_section(state: WritingState) - dict: 节点2撰写当前章节内容 current_index state[current_section_index] section_title state[outline][current_index] prompt ChatPromptTemplate.from_messages([ (system, 你正在撰写文章的一部分。请根据章节标题和上下文撰写详细、连贯的内容。), (human, 文章主题{topic} 当前章节标题{section_title} 已完成的上一章节内容供参考{previous_section} 请开始撰写本章节内容。 ) ]) previous_section state[completed_sections][-1] if state[completed_sections] else 无 chain prompt | llm response chain.invoke({ topic: state[topic], section_title: section_title, previous_section: previous_section }) # 将写好的章节存入一个临时变量供下一个审查节点使用。 # 注意我们并不直接更新completed_sections因为可能需要重写。 new_section response.content return { draft_section: new_section, # 这是一个临时字段未在State中定义但可以在节点间传递。 messages: [{role: assistant, content: f已起草章节 {section_title} 的内容。}] }注意draft_section这个字段并没有在我们定义的WritingState中。在LangGraph中节点返回的更新字典可以包含新的键这些键会被合并到当前状态中。这是一种灵活的临时状态传递方式。但为了清晰最好还是在State中明确定义所有可能的字段。3.3 实现条件路由与审查循环接下来是核心逻辑写好的章节需要经过自我审查。我们设计一个审查节点并根据审查结果决定是“通过”进入下一章还是“重写”当前章。def review_section(state: WritingState) - dict: 节点3审查当前章节草稿 section_title state[outline][state[current_section_index]] draft state.get(draft_section, ) prompt ChatPromptTemplate.from_messages([ (system, 你是一位严格的编辑。请审查以下文章章节草稿指出其在逻辑、事实、文笔或结构上的具体问题。如果问题严重请要求重写。), (human, 章节标题{title} 章节草稿{draft} 请给出审查意见并以‘裁决通过’或‘裁决重写’作为结尾。 ) ]) chain prompt | llm response chain.invoke({title: section_title, draft: draft}) feedback response.content # 解析裁决结果 if 裁决通过 in feedback: decision approve elif 裁决重写 in feedback: decision rewrite else: # 如果LLM没有按格式输出默认要求重写 decision rewrite feedback \n注由于未明确裁决默认要求重写。 return { review_feedback: feedback, review_decision: decision, # 临时字段用于路由 rewrite_count: state.get(rewrite_count, 0) (1 if decision rewrite else 0) } def finalize_section(state: WritingState) - dict: 节点4章节审查通过将其正式加入已完成列表 current_index state[current_section_index] draft state[draft_section] completed state[completed_sections].copy() completed.append(draft) # 准备进入下一章 next_index current_index 1 is_finished next_index len(state[outline]) updates { completed_sections: completed, current_section_index: next_index, # 清理临时字段 draft_section: None, review_feedback: , review_decision: None } if is_finished: updates[status] finished return updates现在我们需要定义条件边函数它根据状态中的某个值这里是review_decision来决定下一个节点。def route_after_review(state: WritingState) - str: 根据审查决定路由到不同节点 decision state.get(review_decision) if decision approve: return finalize_section # 去往节点4定稿 elif decision rewrite: # 检查重写次数避免无限循环 if state.get(rewrite_count, 0) 3: return finalize_section # 超过3次强制通过 else: return write_section # 返回节点2重写 else: # 默认情况也去重写 return write_section3.4 组装图并执行所有部件准备完毕现在用StateGraph把它们组装起来。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态结构 workflow StateGraph(WritingState) # 2. 添加节点 workflow.add_node(generate_outline, generate_outline) workflow.add_node(write_section, write_section) workflow.add_node(review_section, review_section) workflow.add_node(finalize_section, finalize_section) # 3. 设置起始节点 workflow.set_entry_point(generate_outline) # 4. 添加边 workflow.add_edge(generate_outline, write_section) # 生成大纲后开始写第一章 workflow.add_edge(write_section, review_section) # 写完一章立即审查 # 从审查节点出发是条件边 workflow.add_conditional_edges( review_section, route_after_review, # 条件路由函数 { finalize_section: finalize_section, write_section: write_section } ) # 从定稿节点出发判断是否所有章节都写完了 workflow.add_conditional_edges( finalize_section, # 这个函数检查状态决定是继续写下一章还是结束 lambda state: write_section if state[current_section_index] len(state[outline]) else END, { write_section: write_section, END: END } ) # 5. 编译图 app workflow.compile()现在我们可以运行这个智能体了。# 初始化状态 initial_state {topic: 人工智能对未来工作的影响, completed_sections: [], messages: []} # 运行图 final_state app.invoke(initial_state, config{recursion_limit: 50}) # 设置递归上限防止死循环 print(f文章主题{final_state[topic]}) print(f生成大纲{final_state[outline]}) for i, (title, content) in enumerate(zip(final_state[outline], final_state[completed_sections])): print(f\n--- 章节 {i1}: {title} ---) print(content[:200] ...) # 打印前200字符 print(f\n总重写次数{final_state.get(rewrite_count, 0)})执行这段代码你会看到智能体按照“生成大纲 - 写第1章 - 审查 - (通过/重写) - 写第2章 - ...”的流程自动运行直到所有章节完成。控制台输出会清晰地展示这个过程。4. 高级特性与生产级考量上面的例子展示了LangGraph的核心能力。但在生产环境中我们还需要考虑更多。4.1 持久化与人类介入LangGraph的检查点Checkpoint系统允许我们将图的状态保存到数据库。结合interrupt机制可以在特定节点暂停等待外部输入如人工审核。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 创建SQLite存储 conn sqlite3.connect(checkpoints.db) checkpointer SqliteSaver(conn) # 在编译图时加入检查点管理器 app workflow.compile(checkpointercheckpointer) # 现在调用invoke时需要指定一个线程ID用于标识这个特定的工作流实例 config {configurable: {thread_id: user_123_article_1}} initial_state {topic: AI伦理探讨, completed_sections: [], messages: []} # 执行到第一个review_section节点时我们可能想暂停 # 假设我们在review_section节点配置了interrupt_before # 那么执行会暂停并返回当前状态和下一个节点信息。 result app.invoke(initial_state, configconfig) # 假设在review_section处中断了state中会包含review_feedback # 人工审核后可以更新状态中的反馈然后继续执行 updated_state result.state updated_state[review_feedback] 人工审核意见第二部分论据不足请补充案例。 updated_state[review_decision] rewrite # 人工决定重写 # 从中断处继续执行 next_result app.invoke(updated_state, configconfig)这个特性对于构建需要人工审批环节的企业级流程如内容发布、贷款审批极其有用。4.2 并行与子图对于可以并行执行的任务LangGraph支持多线程节点。你可以将一个节点标记为并行执行当所有并行节点完成后再汇聚到下一个节点。这通过add_node和特殊的边配置来实现。更复杂的模块化可以通过子图实现。你可以将一个复杂的子工作流如图片生成流水线编译成一个独立的图然后将其作为主图的一个节点。这大大提升了代码的复用性和可管理性。# 假设我们有一个已编译的图片处理子图image_processing_graph workflow.add_node(“generate_image”, image_processing_graph)在image_processing_graph内部可能又包含了“生成提示词”、“调用DALL-E”、“后期处理”等多个节点。在主图中它就像一个黑盒节点一样被调用。4.3 可视化与调试LangGraph提供了出色的可视化工具这对于调试复杂工作流不可或缺。from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本格式的图结构 print(app.get_graph().print_ascii())生成的Mermaid图会清晰展示所有节点和边包括条件分支。当工作流没有按预期运行时这张图是定位问题的第一份参考资料。此外通过检查每个节点输入/输出的状态快照可以像看日志一样追溯整个执行过程。5. 避坑指南与性能优化在实际项目中踩过一些坑后我总结出以下经验能帮你节省大量调试时间。5.1 状态设计是重中之重问题状态结构设计不合理导致节点间数据传递混乱或状态过于庞大影响性能。建议最小化原则状态只存储工作流真正需要共享和持久化的数据。临时计算的结果尽量放在节点函数的局部变量中或通过返回值中的临时字段传递。显式优于隐式尽量在State的TypedDict中明确定义所有字段。虽然可以动态添加但明确定义有助于类型检查、文档化和团队协作。分离可变与不可变对于列表、字典等可变对象在节点中更新时最好先进行拷贝如list.copy()再修改并返回。这能避免一些难以追踪的副作用。5.2 妥善处理LLM输出的不确定性问题条件边依赖LLM的输出如“裁决通过”但LLM可能不按预定格式回答导致路由失败。解决方案强化提示词工程在系统提示词中严格要求输出格式并使用JsonOutputParser等解析器强制结构化输出。设置默认路由在条件路由函数中对无法解析的情况设置一个安全的默认路径例如无法判断时导向“人工审核”或“重试”节点。添加验证节点在关键决策点前增加一个“输出格式验证与清洗”节点确保下游节点接收到的数据是干净的。5.3 控制循环与超时问题智能体陷入死循环不断重写某个章节或等待永远不来的人工反馈。解决方案设置硬性限制在状态中设置iteration或rewrite_count字段并在条件边函数中检查。超过阈值则强制跳出循环导向失败处理或人工接管节点。利用recursion_limit在app.invoke()的config中设置recursion_limit这是LangGraph内置的防止图无限递归的保险丝。实现超时机制对于等待外部事件的节点如人类反馈可以在调用时设置一个超时时间超时后触发一个超时处理子图。5.4 性能与成本优化问题工作流中LLM调用次数过多导致响应慢、成本高。优化策略缓存LLM响应对具有确定性的LLM调用例如基于相同大纲生成章节标题可以使用LangChain的缓存组件如InMemoryCache,SQLiteCache避免重复计算。并行化独立节点仔细分析工作流找出可以并行执行的节点例如文章不同章节的图片生成。使用LangGraph的并发特性来缩短整体运行时间。流式输出对于需要实时向用户展示进度的场景如一个字一个字地生成文章可以利用LLM的流式响应和LangGraph的状态流更新实现更流畅的用户体验。从链Chain到图Graph的转变不仅仅是换了一个工具更是思维模式的升级。LangGraph迫使你更早地思考应用的状态流和控制流而这通常能催生出更健壮、更易维护的架构。它可能不像直接调用ChatGPT API那样“快糙猛”但对于任何严肃的、计划投入生产的AI智能体应用来说这份在设计和编排上投入的精力最终都会在可靠性、可扩展性和可调试性上带来丰厚的回报。开始用图的视角来设计你的下一个智能体吧你会发现复杂的协作逻辑 suddenly makes sense。