LangGraph实战：构建有状态AI工作流与多智能体系统

1. 项目概述：从LangChain到LangGraph的范式演进

如果你在过去一年里深度参与过基于大语言模型的应用开发，那么“LangChain”这个名字对你来说一定如雷贯耳。它几乎成了构建LLM应用的事实标准框架，将提示词管理、文档检索、链式调用等复杂概念封装成简洁的API，让我们能快速搭建起一个可用的智能体或工作流。然而，随着我们构建的应用从简单的“问答机器人”迈向复杂的、多步骤的、具备状态和记忆的“智能业务流程”，单纯使用链（Chain）开始显得力不从心。你可能会遇到这样的困境：一个任务需要根据中间结果动态决定下一步；多个“智能体”需要协作并共享上下文；或者一个长流程需要持久化其执行状态以便中断后恢复。这正是langchain-ai/langgraph诞生的背景。

简单来说，LangGraph是LangChain团队推出的一个用于构建有状态、多参与者工作流的库。它不再将应用视为一条单向流动的“链”，而是将其抽象为一个“图”（Graph）。图中的节点代表一个执行单元（可以是一个LLM调用、一个工具函数，或一段逻辑代码），边则定义了节点之间的流转条件。这种范式允许循环、分支、并行以及状态的全局管理，极大地增强了我们构建复杂、鲁棒的AI应用的能力。它特别适合构建需要长期记忆、工具使用、多智能体协作以及复杂决策逻辑的AI智能体系统。

2. 核心设计理念：为什么是“图”？

在深入代码之前，理解LangGraph背后的设计哲学至关重要。这能帮助你在正确的场景选择它，而不是盲目套用。

2.1 超越“链”的局限性

传统的LangChain“链”本质上是线性的。A -> B -> C，数据按预定顺序流动。虽然可以通过RouterChain等方式实现简单分支，但对于以下场景，线性链就显得非常笨拙：

1. 循环与迭代：例如，一个代码生成智能体，需要先生成代码，再运行测试，如果测试失败，则分析错误并重新生成。这个过程可能循环多次。
2. 动态路由：下一步执行哪个节点，完全取决于上一步的输出内容。比如一个客服机器人，需要根据用户意图（查询、投诉、转人工）调用完全不同的子流程。
3. 多参与者协作：想象一个软件设计团队，有“产品经理”智能体定义需求，“架构师”智能体设计模块，“程序员”智能体编写代码，它们需要在一个共享的“项目状态”上协同工作，互相传递和修改信息。
4. 状态持久化与恢复：一个耗时很长的流程（如分析一份100页的文档），需要支持暂停、保存当前状态，后续再从中断点继续执行。

“图”的模型天然适合描述这些复杂流程。节点是功能单元，边是控制流逻辑。LangGraph让你能够以声明式的方式定义这个图，然后由它的运行时引擎来负责状态管理、节点调度和循环控制。

2.2 两大核心抽象：State与Node

LangGraph的整个架构围绕两个核心概念构建：

State（状态）这是整个工作流的“共享内存”。它是一个字典（或Pydantic模型），定义了图中所有节点都能读取和修改的字段。例如，对于一个写作助手，State里可能有topic（主题）、outline（大纲）、draft（草稿）、feedback（反馈）等字段。State会在整个图的执行过程中被传递和更新。

Node（节点）节点是一个接收当前State、执行某些操作、并返回更新后State的函数。这个函数可以是：

• 一个调用LLM并解析其响应的函数。
• 一个执行计算或调用外部API的工具函数。
• 一个仅仅根据条件修改State中某个标志的逻辑函数。

节点之间通过“边”连接，但下一个节点由谁执行，可以通过一个特殊的“路由逻辑”来决定。

2.3 控制流：让图“活”起来

LangGraph提供了几种强大的控制流原语，这是其灵魂所在：

1. 条件边（Conditional Edge）：根据当前State中的某个值，决定下一步走向哪个节点。这实现了if-else分支。
2. 入口点（Entry Point）：指定图开始执行的第一个节点。
3. 终点（Finish Point）：当图执行到某个节点后，可以标记为结束。
4. 循环（Cycle）：通过将边指向之前的节点，可以形成循环。通常需要配合条件边来设置退出循环的条件，否则会成为死循环。

通过组合这些元素，你可以构建出状态机、循环工作流、多智能体协作系统等复杂结构。

3. 从零构建你的第一个LangGraph应用

理论说得再多，不如动手实践。让我们构建一个经典的“写作助手”智能体，它能够根据用户主题，先生成大纲，再撰写草稿，最后根据模拟的反馈进行修订。这个过程包含线性步骤和条件循环。

3.1 环境搭建与基础定义

首先，安装必要的库。这里我们使用OpenAI的模型。

pip install langgraph openai langchain-openai

接下来，我们定义整个工作流共享的State。使用TypedDict可以让类型提示更清晰。

from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 1. 定义State：描述整个工作流的数据结构 class WritingState(TypedDict): topic: str # 用户输入的主题 outline: List[str] # 生成的大纲（列表） draft: str # 撰写的草稿 feedback: str # 收到的反馈 revised_draft: str # 修订后的草稿 revision_count: int # 修订次数，用于控制循环 needs_revision: bool # 是否还需要继续修订

我们定义了一个包含文章写作各个阶段信息的State。Annotated和operator.add是LangGraph用于定义“如何更新State”的语法，这里我们先使用简单的赋值更新。

3.2 实现核心功能节点

现在，我们创建三个核心节点函数：generate_outline,write_draft,review_and_revise。

# 初始化LLM llm = ChatOpenAI(model="gpt-4-turbo-preview") # 2. 节点A：生成大纲 def generate_outline(state: WritingState) -> WritingState: """根据主题生成文章大纲""" prompt = f""" 请为以下主题生成一份详细的文章大纲，以列表形式返回核心要点： 主题：{state['topic']} """ response = llm.invoke(prompt) # 假设LLM返回的是文本，我们按行分割成列表。实际应用中可能需要更复杂的解析。 outline_list = response.content.strip().split('\n') # 过滤空行和列表标记 outline_list = [point.strip(' -•') for point in outline_list if point.strip()] # 更新State state['outline'] = outline_list print(f"[节点：生成大纲] 完成。生成{len(outline_list)}个要点。") return state # 3. 节点B：根据大纲撰写草稿 def write_draft(state: WritingState) -> WritingState: """根据大纲撰写完整草稿""" outline_str = '\n'.join(state['outline']) prompt = f""" 请根据以下主题和大纲，撰写一篇完整的文章草稿。 主题：{state['topic']} 大纲： {outline_str} 请输出完整的文章内容。 """ response = llm.invoke(prompt) state['draft'] = response.content # 初始化修订相关状态 state['revision_count'] = 0 state['needs_revision'] = True # 默认进入修订环节 print(f"[节点：撰写草稿] 完成。草稿长度：{len(state['draft'])}字符。") return state

前两个节点是线性的。关键在第三个节点，它模拟了“获取反馈并决定是否修订”的循环过程。

# 4. 节点C：评审与修订（这是一个可能被多次调用的节点） def review_and_revise(state: WritingState) -> WritingState: """模拟评审，生成反馈并决定是否修订""" revision_num = state['revision_count'] + 1 state['revision_count'] = revision_num print(f"\n=== 第 {revision_num} 轮修订 ===") # 模拟一个“评审者”生成反馈（实际可能是用户反馈或另一个LLM） review_prompt = f""" 请评审以下文章草稿，提供一段简洁的修改反馈。如果文章已经很好，请说“无需修改”。 文章主题：{state['topic']} 文章草稿： {state['draft'] if revision_num == 1 else state['revised_draft']} """ feedback_response = llm.invoke(review_prompt) state['feedback'] = feedback_response.content print(f"[模拟评审反馈]: {state['feedback'][:150]}...") # 关键逻辑：根据反馈内容决定是否需要修订 # 这里用一个简单规则：如果反馈包含“无需修改”，则停止；或者修订超过3次也强制停止。 if "无需修改" in state['feedback'] or revision_num >= 3: state['needs_revision'] = False print(f"决定终止修订。原因：{'反馈满意' if '无需修改' in state['feedback'] else '达到最大修订次数'}。") if revision_num == 1: # 如果第一轮就无需修改，则将draft复制到revised_draft state['revised_draft'] = state['draft'] else: # 需要进行修订 revise_prompt = f""" 根据以下反馈，对文章进行修订。 原始文章： {state['draft'] if revision_num == 1 else state['revised_draft']} 反馈意见： {state['feedback']} 请输出修订后的完整文章。 """ revise_response = llm.invoke(revise_prompt) state['revised_draft'] = revise_response.content state['needs_revision'] = True # 保持True，等待下一轮评审 print(f"[节点：执行修订] 完成。修订后长度：{len(state['revised_draft'])}字符。") return state

这个节点是工作流的核心。它每次被调用都会增加修订计数，模拟获取反馈，并根据反馈内容（或修订次数）动态设置state[‘needs_revision’]的值。这个值将决定工作流的下一步走向。

3.3 组装图并定义控制流

有了节点，我们需要把它们组装成一个有向图，并定义执行路径。

# 5. 创建图构建器 workflow = StateGraph(WritingState) # 6. 添加节点 workflow.add_node(“generate_outline”, generate_outline) workflow.add_node(“write_draft”, write_draft) workflow.add_node(“review_and_revise”, review_and_revise) # 7. 定义边的连接关系 # 7.1 设置入口点：从“生成大纲”开始 workflow.set_entry_point(“generate_outline”) # 7.2 线性连接：大纲 -> 草稿 workflow.add_edge(“generate_outline”, “write_draft”) # 7.3 从草稿到评审修订 workflow.add_edge(“write_draft”, “review_and_revise”) # 8. 定义条件边：从“评审修订”节点出发，根据状态决定下一步 def decide_after_review(state: WritingState) -> str: """路由函数：判断是否需要继续修订""" if state[‘needs_revision’]: # 如果需要修订，则返回节点名”review_and_revise”，形成循环 return “review_and_revise” else: # 如果不需要修订，则结束整个工作流 return END workflow.add_conditional_edges( “review_and_revise”, # 从这个节点出发 decide_after_review, # 使用这个函数决定下一个节点 {“review_and_revise”: “review_and_revise”, END: END} # 可能的目的地映射 ) # 9. 编译图，得到可执行对象 app = workflow.compile()

这段代码构建了一个完整的图：generate_outline -> write_draft -> review_and_revise。从review_and_revise节点出来，有一条条件边，它调用decide_after_review函数检查State中的needs_revision标志。如果为True，则路由回review_and_revise节点自身，形成循环；如果为False，则路由到预定义的END，表示工作流结束。

3.4 执行与可视化

现在，我们可以运行这个工作流，并查看其内部结构。

# 10. 初始化输入状态 initial_state: WritingState = { “topic”: “人工智能在医疗诊断中的应用与挑战”, “outline”: [], “draft”: “”, “feedback”: “”, “revised_draft”: “”, “revision_count”: 0, “needs_revision”: False, } # 11. 执行图 print(“开始执行写作助手工作流…\n”) final_state = app.invoke(initial_state) print(“\n=== 工作流执行结束 ===”) print(f“最终修订次数：{final_state[‘revision_count’]}”) print(f“最终草稿预览：{final_state.get(‘revised_draft’, final_state.get(‘draft’, ‘’))[:200]}…”)

为了理解我们构建的图，LangGraph提供了出色的可视化工具。

# 12. 可视化图结构（需要安装graphviz） try: from IPython.display import Image, display # 将图导出为PNG图片 image_data = app.get_graph().draw_mermaid_png() display(Image(image_data)) except ImportError: # 如果不方便显示图片，可以打印文本表示 print(app.get_graph().draw_mermaid())

生成的图会清晰地显示三个节点，以及从review_and_revise节点出发的菱形决策框和两条边，直观展示了循环逻辑。

注意：在实际项目中，LLM调用可能失败、网络可能不稳定。一个重要的实操心得是，务必在关键节点函数中添加完善的异常处理和状态回滚逻辑。例如，在review_and_revise节点中，如果LLM调用失败，你应该捕获异常，将needs_revision设置为False，并在State中记录错误信息，而不是让整个图崩溃。这能极大提升生产环境工作流的鲁棒性。

4. 进阶模式：多智能体协作与工具调用

基础的循环工作流只是LangGraph的冰山一角。它的真正威力体现在构建多智能体系统和集成工具调用上。让我们设计一个更复杂的场景：一个“研究助手”团队，包含一个“规划员”、一个“研究员”和一个“写作者”。

4.1 设计多智能体State

首先，我们需要一个更复杂的State来支持团队协作。

from typing import Literal from datetime import datetime class ResearchState(TypedDict): query: str # 用户的研究问题 plan: str # 规划员制定的研究计划 research_materials: List[str] # 研究员收集的资料摘要列表 report: str # 写作者生成的报告 current_agent: Literal[“planner”, “researcher”, “writer”, “end”] # 当前该谁行动 max_turns: int # 最大协作轮次 current_turn: int # 当前轮次 last_activity: str # 上一个活动的记录

这里我们引入了current_agent字段，明确指示当前应该由哪个“智能体”节点来行动。这是一种清晰的状态驱动路由方式。

4.2 实现智能体节点与工具使用

每个智能体都是一个独立的节点，并且“研究员”智能体可以使用工具（如网络搜索）。

# 假设我们有一个搜索工具（这里用模拟函数代替SerpAPI等真实工具） def web_search(query: str) -> str: """模拟网络搜索工具""" # 真实场景下，这里会集成SerpAPI、Tavily等搜索API print(f”[工具调用] 搜索：{query}“) # 返回模拟结果 return f”关于‘{query}’的模拟搜索结果摘要。当前时间：{datetime.now()}。” # 规划员智能体节点 def planner_agent(state: ResearchState) -> ResearchState: print(f”\n[回合{state[‘current_turn’]}] 规划员开始工作...“) prompt = f”用户的研究问题是：{state[‘query’]}。请制定一个分三步的研究计划。" response = llm.invoke(prompt) state[‘plan’] = response.content state[‘current_agent’] = “researcher” # 指定下一个执行者 state[‘last_activity’] = “规划员制定了计划” return state # 研究员智能体节点（使用工具） def researcher_agent(state: ResearchState) -> ResearchState: print(f”[回合{state[‘current_turn’]}] 研究员开始工作...“) # 根据计划中的关键词进行搜索（这里简化处理，搜索整个问题） search_result = web_search(state[‘query’]) # 让LLM总结搜索结果 summary_prompt = f”请将以下搜索材料进行归纳总结，形成一条简洁的研究笔记。\n材料：{search_result}“ summary_response = llm.invoke(summary_prompt) # 将笔记添加到材料库 if ‘research_materials’ not in state or state[‘research_materials’] is None: state[‘research_materials’] = [] state[‘research_materials’].append(summary_response.content) # 判断是否收集了足够材料（这里简单判断数量） if len(state[‘research_materials’]) >= 2: # 假设收集2条笔记后转给写作者 state[‘current_agent’] = “writer” else: # 材料不足，继续研究（但增加轮次计数以防死循环） state[‘current_agent’] = “researcher” state[‘last_activity’] = f”研究员收集了{len(state[‘research_materials’])}条材料” return state # 写作者智能体节点 def writer_agent(state: ResearchState) -> ResearchState: print(f”[回合{state[‘current_turn’]}] 写作者开始工作...“) materials = ‘\n---\n’.join(state[‘research_materials’]) prompt = f”根据以下研究计划和收集的材料，撰写一份详细的研究报告。\n研究计划：{state[‘plan’]}\n研究材料：{materials}“ response = llm.invoke(prompt) state[‘report’] = response.content state[‘current_agent’] = “end” # 工作完成 state[‘last_activity’] = “写作者完成了报告” return state

4.3 构建基于状态路由的协作图

这个图的边不是固定的，而是完全由State中的current_agent字段决定。

# 构建图 research_workflow = StateGraph(ResearchState) # 添加节点 research_workflow.add_node(“planner”, planner_agent) research_workflow.add_node(“researcher”, researcher_agent) research_workflow.add_node(“writer”, writer_agent) # 设置入口点 research_workflow.set_entry_point(“planner”) # 定义路由函数：这是多智能体协作的核心 def route_by_agent(state: ResearchState) -> str: """根据state中的current_agent字段决定下一个节点""" # 增加轮次计数 state[‘current_turn’] = state.get(‘current_turn’, 0) + 1 # 检查是否超过最大轮次，防止无限循环 if state[‘current_turn’] > state.get(‘max_turns’, 5): print(“达到最大协作轮次，强制结束。”) return “writer” # 强制跳转到写作者进行收尾 next_agent = state[‘current_agent’] print(f”路由决策：下一个执行者是 {next_agent}“) return next_agent # 返回下一个节点的名称 # 添加条件边：每个节点执行完后，都通过同一个路由函数决定下一步 research_workflow.add_conditional_edges( “planner”, route_by_agent, {“planner”: “planner”, “researcher”: “researcher”, “writer”: “writer”, “end”: END} ) research_workflow.add_conditional_edges( “researcher”, route_by_agent, {“planner”: “planner”, “researcher”: “researcher”, “writer”: “writer”, “end”: END} ) research_workflow.add_conditional_edges( “writer”, route_by_agent, {“planner”: “planner”, “researcher”: “researcher”, “writer”: “writer”, “end”: END} ) # 编译图 research_app = research_workflow.compile()

在这个设计中，route_by_agent是核心调度器。每个智能体完成工作后，都会修改state[‘current_agent’]，然后由这个路由函数读取该值，决定下一个激活的节点。这实现了动态的、基于状态的工作流控制，非常灵活。

实操心得：状态设计是灵魂。在多智能体或复杂工作流中，State的设计至关重要。你需要仔细思考哪些数据是共享的，哪些是节点私有的（不应放入State）。一个好的经验是：State应该只包含影响工作流控制流和最终输出的核心数据。临时变量或中间计算过程最好放在节点函数的局部作用域内。过度设计State会让图难以理解和调试。

5. 生产级考量：检查点、持久化与并发

对于简单的演示流程，上述代码已经足够。但一旦部署到生产环境，你必须考虑三个关键问题：持久化（如何保存和恢复长耗时任务的状态）、检查点（如何在任意步骤暂停和继续）以及并发（如何同时处理多个请求）。LangGraph通过Checkpointer抽象和StateGraph的compile(checkpointer=…)参数来支持这些特性。

5.1 使用检查点实现状态持久化

检查点机制允许你将图执行过程中的任意状态快照保存到数据库（如Redis、PostgreSQL），并在之后通过线程ID或进程ID恢复执行。这对于需要长时间运行或可能中断的工作流至关重要。

from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 创建一个临时SQLite数据库来存储检查点（生产环境应使用更稳定的存储） _, db_path = tempfile.mkstemp(suffix=”.db“) print(f”检查点数据库位于：{db_path}“) # 创建SQLite检查点存储器 checkpointer = SqliteSaver.from_conn_string(f”sqlite:///{db_path}“) # 在编译图时传入检查点存储器 persistent_app = workflow.compile(checkpointer=checkpointer) # 现在，调用invoke时需要指定一个config，其中包含thread_id config = {“configurable”: {“thread_id”: “user_123_session_1”}} # thread_id唯一标识一个工作流实例 initial_state = {“topic”: “可持续能源的未来”} # 第一次执行 print(“第一次执行（可能只执行了部分步骤）...”) try: # 模拟执行到一半被中断（例如，我们在这里手动抛出异常） result1 = persistent_app.invoke(initial_state, config=config) except Exception as e: print(f”模拟执行中断：{e}“) # 状态已经被自动保存到检查点 # 第二次执行，从检查点恢复 print(“\n从检查点恢复执行...”) result2 = persistent_app.invoke(None, config=config) # 输入状态为None，表示从检查点恢复 print(f”恢复后最终结果：{result2[‘topic’]}“)

通过使用checkpointer，LangGraph会在每个节点执行后自动保存State。当使用相同的thread_id再次调用invoke时，它会从上次中断的节点继续执行，而不是从头开始。这对于处理网络超时、服务重启或用户主动暂停的场景非常有用。

5.2 配置与并发处理

在生产中，一个LangGraph应用会同时处理成千上万个独立的请求（例如，每个用户对话都是一个独立的工作流实例）。config参数就是用来区分这些实例的。

• thread_id：这是最重要的配置项，用于唯一标识一个工作流会话。通常可以用用户ID_会话ID的组合。
• configurable字典：你还可以在这里传递其他动态参数，比如不同用户的API密钥、模型选择等。这些参数可以在节点函数中通过context对象访问。

def node_with_config(state: WritingState, context): # 可以通过context.configurable获取配置 user_model = context.configurable.get(“model”, “gpt-4”) llm = ChatOpenAI(model=user_model) # … 其余逻辑

一个常见的坑是忘记管理thread_id的生命周期。如果你为每个新请求重复使用同一个thread_id，那么新请求会从上一个请求的检查点恢复，导致状态混乱。务必确保每个独立的工作流实例都有唯一的thread_id，并在会话结束时（或定期）清理旧的检查点数据，以免存储膨胀。

6. 调试、监控与性能优化

当你的LangGraph应用变得复杂时，调试和监控就成为挑战。以下是几个实战技巧：

1. 可视化与日志记录如前所述，app.get_graph().draw_mermaid()是无价的调试工具。此外，在每个节点的开始和结束处打印详细的日志（包括State的关键字段），可以帮助你跟踪执行流。考虑使用结构化日志库（如structlog），并将thread_id包含在每条日志中，便于追踪。

2. 使用interrupt_before/after进行调试LangGraph允许你在特定节点的执行前或执行后设置“中断点”，以便手动检查或修改State。

from langgraph.graph import StateGraph workflow = StateGraph(WritingState) # … 添加节点和边 # 在编译前，可以标记在哪个节点前后中断 app = workflow.compile(interrupt_before=[“review_and_revise”]) # 调用时，执行会在指定的节点前暂停，返回一个可继续执行的对象 result = app.invoke(initial_state, config={“configurable”: {“thread_id”: “debug_1”}}) print(“在‘review_and_revise’节点前中断。当前State:”, result) # 你可以手动修改result，然后继续执行 # app.invoke(result, …)

3. 性能优化：异步与批处理如果节点涉及大量IO（如调用多个外部API或数据库查询），强烈建议使用异步节点。

import asyncio async def async_research_agent(state: ResearchState): # 可以在这里并发发起多个网络请求 tasks = [async_web_search(q) for q in queries] results = await asyncio.gather(*tasks) # … 处理结果 return state

在添加节点时，LangGraph能正确处理异步函数。对于需要调用多个同类工具的场景（如同时检索多个知识库），在单个节点内进行批处理，比分拆成多个串行节点效率高得多。

4. 超时与重试机制在节点函数中，对于可能失败的LLM调用或API请求，务必实现重试逻辑和超时控制。可以使用tenacity等重试库。同时，在State中设计一个errors字段来记录非致命错误，避免因单次调用失败导致整个工作流崩溃。

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def call_llm_with_retry(prompt): # 带有指数退避的重试逻辑 return llm.invoke(prompt)

构建基于LangGraph的复杂AI工作流，就像在编写一个智能程序的剧本。State是你的舞台背景，Node是演员，而由条件边和路由函数定义的图结构，就是导演手中的剧本。它赋予了应用动态响应、持续演进和协作决策的能力。从我个人的实践经验来看，从“链式思维”切换到“图式思维”需要一个适应过程，但一旦掌握，你将能设计出此前难以想象的复杂AI应用。最关键的是始终保持State的简洁与清晰，它是整个系统可维护性的基石。开始你的第一个图构建吧，从自动化一个你日常工作中重复的多步骤任务开始，你会迅速体会到它的强大之处。