企业官网建设流程全解析

1. 从“线性脚本”到“状态驱动”：为什么需要进阶的工作流？

如果你用过 LangChain 或者自己写过一些简单的 AI 应用脚本，大概会经历这样一个过程：写一个函数，调用大模型 API，处理返回结果，再根据结果决定下一步。这就像写一个线性的剧本，角色（Agent）按部就班地说台词。但当剧情变得复杂，角色增多，甚至需要根据观众（用户）的实时反馈来调整剧情时，这种“剧本式”的编程就捉襟见肘了。

这就是LangGraph要解决的核心问题。它不是一个替代 LangChain 的工具，而是一个基于 LangChain 构建的、专门用于编排复杂、有状态、多步骤 AI 工作流的框架。你可以把它想象成一个可视化、可编程的流程图引擎，专门为 AI 智能体（Agent）设计。我们之前可能已经了解了如何用 LangGraph 构建一个简单的对话链或工具调用流程，但那只是入门。当我们要构建真正可用于生产环境的系统时，三个进阶能力至关重要：状态持久化（检查点）、人工介入（Human-in-the-loop）和多智能体协作。

想象一下这些场景：一个自动化的客户支持系统，在处理到一半时需要转接给人工客服确认敏感信息；一个多步骤的数据分析流水线，其中某个步骤失败后，我们希望能从失败点继续，而不是重头开始；一个由“研究员”、“写手”、“审阅员”多个 AI 角色组成的团队，如何高效协作完成一份报告？这些需求，正是 LangGraph 进阶功能大显身手的地方。本文将深入这三个核心进阶主题，结合代码示例和设计思路，帮你把 LangGraph 从“玩具”升级为“生产级工具”。

2. 状态持久化：理解与应用检查点（Checkpoint）

在简单的工作流中，状态（State）是内存中的临时对象，流程跑完就消失了。但对于一个可能运行数小时、包含昂贵模型调用（比如 GPT-4）的复杂工作流来说，这是不可接受的。我们需要检查点（Checkpoint）机制：将工作流在任意节点的完整状态保存下来，允许系统在中断（如程序崩溃、服务器重启）后从中断点恢复，也支持暂停、继续等操作。

2.1 检查点的核心概念与价值

LangGraph 中的检查点不仅仅是保存一个变量值。它保存的是整个State对象的序列化快照，包括：

1. 当前节点（Node）：工作流执行到了哪个步骤。
2. 状态值（Values）：State对象中所有键的值。
3. 下一步（Next）：根据当前状态和节点逻辑，计算出的下一个要执行的节点（或节点列表）。
4. 元数据（Metadata）：可选，可以包含时间戳、版本、创建者等信息。

其核心价值在于：

• 容错与恢复：进程崩溃或部署更新后，可以从最新的检查点继续，避免重复执行已完成的昂贵步骤（如大模型调用）。
• 调试与审计：可以回溯工作流的历史状态，查看每一步的输入输出，便于定位问题。
• 支持异步与长时运行：工作流可以暂停，等待外部事件（如人工审核、第三方API回调），事件到来后从检查点继续。
• 实现“续跑”功能：用户关闭了网页，下次打开可以继续之前的会话。

2.2 如何实现检查点：配置持久化存储

LangGraph 通过CheckpointSaver这一抽象来实现持久化。它不是一个开箱即用的数据库，而是一个接口，你需要为其提供一个具体的存储后端。社区中已有一些实现，你也可以自己实现。

一个常见的搭配是使用SqliteSaver，它轻量且易于集成。下面是一个完整的示例，展示如何创建一个带检查点的工作流。

首先，定义我们的状态。假设我们有一个文档总结和提问的工作流。

from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langgraph.checkpoint.sqlite import SqliteSaver import operator # 1. 定义状态结构 class State(TypedDict): # 原始文档 original_document: str # 生成的摘要 summary: str # 根据摘要生成的问题列表 questions: List[str] # 用户对问题的答案 answers: Annotated[List[str], operator.add] # 使用operator.add来追加列表 # 当前步骤 step: str # 2. 创建持久化存储后端 # 这里使用SQLite，数据会保存在 `checkpoints.db` 文件中 memory = SqliteSaver.from_conn_string("checkpoints.db") # 3. 构建图，并传入 `checkpointer` builder = StateGraph(State, config_schema=dict) # config_schema 用于区分不同工作流实例 builder.add_node("summarize", summarize_node) builder.add_node("generate_questions", generate_questions_node) builder.add_node("collect_answers", collect_answers_node) builder.set_entry_point("summarize") builder.add_edge("summarize", "generate_questions") builder.add_edge("generate_questions", "collect_answers") builder.add_edge("collect_answers", END) # 关键步骤：将 memory 作为 checkpointer 传入 graph = builder.compile(checkpointer=memory)

现在，当我们运行这个图时，需要提供一个config参数，其中包含configurable字段来指定本次运行的thread_id。thread_id是唯一标识一个工作流会话的键，所有检查点都会通过它来存储和读取。

# 运行工作流，并指定 thread_id config = {"configurable": {"thread_id": "user_session_12345"}} initial_state = {"original_document": "这里是一篇很长的技术文章..."} # 第一次运行 result = graph.invoke(initial_state, config=config) print(result["summary"]) # 输出摘要 # 模拟中断：程序在这里结束 # 第二次运行（恢复）：我们只需要相同的 config，并从空状态或部分状态开始调用。 # LangGraph 会自动加载最新的检查点，并从上次中断的节点继续执行。 new_result = graph.invoke({}, config=config) # 注意这里传入空状态或仅更新部分状态 print(new_result["questions"]) # 会输出之前 generate_questions 节点产生的问题

注意：在恢复执行时，invoke传入的初始状态会与从检查点加载的状态进行合并。通常，如果你只是想继续，传入空字典{}即可。如果你想用新数据覆盖状态的某些部分（例如更新了文档），你可以传入对应的键值对。

2.3 检查点的高级用法与避坑指南

1. 并发安全与锁机制： 当多个进程或线程同时尝试恢复和更新同一个thread_id的检查点时，会发生竞争条件。SqliteSaver利用数据库事务提供了一定的并发安全。但对于高并发场景，你可能需要在应用层或使用支持乐观锁/悲观锁的存储后端（如 Redis、PostgreSQL）来实现更精细的控制。

2. 状态序列化与版本控制： 如果你的State结构（即TypedDict的字段）发生了变更（比如新增或删除了一个字段），旧的检查点在反序列化时可能会失败。在生产环境中，你需要考虑状态模式的版本迁移策略。一个简单的方法是在State中保留一个version字段，并在加载旧检查点时编写迁移逻辑。

3. 存储成本与清理： 检查点会不断创建。对于一个长期运行、频繁触发的工作流，存储空间可能快速增长。你需要定期清理旧的检查点。SqliteSaver本身不提供自动清理，你需要自己执行 SQL 语句（如DELETE FROM checkpoints WHERE thread_id = ? AND timestamp < ?）或使用其他存储后端的 TTL（生存时间）功能。

4. 配置化（Configurable）的威力：config参数中的configurable字段非常强大。除了thread_id，你还可以存储其他会话级元数据，比如user_id、project_id等。这些信息也会被保存在检查点中，并在恢复时可用。这允许你基于更丰富的上下文来恢复工作流。

config = { "configurable": { "thread_id": "analysis_789", "user_id": "user_abc", "priority": "high" } }

3. 人工介入：实现 Human-in-the-Loop 工作流

完全自动化的 AI 工作流虽然高效，但在处理关键决策、创造性任务或涉及伦理、安全的场景时，引入人工判断是必要且明智的。LangGraph 通过“暂停”并等待外部输入的机制来优雅地支持这一点。

3.1 核心机制：interrupt_before与interrupt_after

LangGraph 允许你在指定的节点之前或之后设置中断点。当工作流执行到这些点时，它会自动暂停，将控制权交还给调用者，并返回一个特殊的结果，表明它正在等待某个特定节点的输入。

• interrupt_before[“node_name”]: 在进入node_name节点之前暂停。这通常用于在AI执行前，由人来提供输入或指令。
• interrupt_after[“node_name”]: 在离开node_name节点之后暂停。这通常用于在AI产生输出后，由人来审核、修改或确认结果。

3.2 实战：构建一个带人工审核的文档发布流程

让我们构建一个简单的博客发布工作流：AI 生成初稿 -> 人工审核 -> AI 根据反馈修改 -> 最终发布。

from typing import Literal from langgraph.graph import StateGraph, END, MessagesState from langgraph.prebuilt import ToolNode from langgraph.checkpoint.sqlite import SqliteSaver # 使用 MessagesState 简化对话状态管理 class State(MessagesState): draft: str = "" human_feedback: str = "" final_content: str = "" status: Literal["drafting", "awaiting_review", "revising", "published"] = "drafting" # 定义节点函数 def write_draft_node(state: State): # 模拟AI写作 new_draft = f"基于最新趋势，这是一篇关于LangGraph的博文草稿。当前状态: {state['status']}" return {"draft": new_draft, "status": "awaiting_review"} def revise_draft_node(state: State): # 模拟AI根据反馈修改 feedback = state.get("human_feedback", "无反馈") revised = f"{state['draft']}\n\n---\n已根据反馈 '{feedback}' 进行修改。" return {"draft": revised, "status": "revising"} def publish_node(state: State): return {"final_content": state["draft"], "status": "published"} # 构建图 builder = StateGraph(State) builder.add_node("write_draft", write_draft_node) builder.add_node("revise_draft", revise_draft_node) builder.add_node("publish", publish_node) builder.set_entry_point("write_draft") # 关键：在 write_draft 之后设置中断，等待人工审核 builder.add_edge("write_draft", "revise_draft") builder.add_edge("revise_draft", "publish") builder.add_edge("publish", END) memory = SqliteSaver.from_conn_string(":memory:") # 配置中断：在 ‘revise_draft’ 节点之前中断，因为我们需要在AI修改前拿到人工反馈。 graph = builder.compile( checkpointer=memory, interrupt_before=["revise_draft"] # 这里设置中断点 )

运行这个工作流：

config = {"configurable": {"thread_id": "blog_post_1"}} # 1. 首次调用，AI写草稿，然后在 revise_draft 前中断 result = graph.invoke({"status": "drafting"}, config=config) print(result) # 输出可能包含一个特殊标记，或者你可以检查状态。实际上，`invoke` 会抛出一个 `Interruption` 异常或返回特定结构。 # 在 LangGraph 的流式或事件处理中，更常见的模式是使用 `stream` 并监听事件。 # 为了清晰，我们换一种方式演示：使用 `stream` 并处理 `Send` 事件。 from langgraph.graph import Send # 假设我们有一个处理函数 def run_workflow_with_human(): thread_config = {"configurable": {"thread_id": "blog_post_2"}} # 初始化 for event in graph.stream({"status": "drafting"}, thread_config, stream_mode="values"): if isinstance(event, dict): print(f"状态更新: {event}") # 在实际应用中，这里会有一个事件循环，当检测到需要人工介入时，就跳出循环或发送通知。 # 更直观的 API 使用：`get_state` 和 `update_state` # 首先，运行到中断点 try: graph.invoke({"status": "drafting"}, config=config) except Exception as e: # 这里会捕获到中断，在实际框架中，可能有更优雅的交互方式。 # 例如，Dify、Coze 等平台封装了此过程，在UI上生成一个“待办事项”。 pass # 此时，工作流已暂停。我们可以获取当前状态。 current_state = graph.get_state(config) print(f"当前草稿: {current_state.values['draft']}") print(f"当前状态: {current_state.values['status']}") # 应该是 awaiting_review # 现在，模拟人工审核，提供反馈 human_feedback_input = "开头不够吸引人，请加入一个具体的案例。" # 将反馈更新到状态中，并告诉图继续执行（从中断点 revise_draft 开始） updated_state = {"human_feedback": human_feedback_input} graph.update_state(config, updated_state) # 继续执行（从 revise_draft 节点开始） result = graph.invoke({}, config=config) # 传入空状态或仅包含更新的状态 print(f"最终内容: {result['final_content']}") print(f"发布状态: {result['status']}")

实操心得：在真实的后端服务中，你通常不会用try...except来捕获中断。更好的模式是：

1. 使用graph.stream()异步执行工作流。
2. 监听事件类型，当遇到Send事件（其name字段指向一个中断的节点）时，将工作流state和config存入数据库，并生成一个任务ID。
3. 暴露一个 REST API（如POST /task/{task_id}/feedback）供前端或人工审核界面调用。
4. 当API接收到反馈后，从数据库加载状态和配置，调用graph.update_state()和graph.invoke()继续执行。
5. 这种模式天然与检查点机制结合，非常健壮。

3.3 设计人工介入点的考量

• 中断的粒度：是在一个复杂节点的前后中断，还是将复杂节点拆分成多个小节点，在中间中断？后者更灵活，但图结构会更复杂。
• 超时与备选方案：如果人工审核迟迟没有响应怎么办？你需要设置超时机制，并在工作流中设计“默认路径”或“升级路径”（例如，超时后自动发送提醒，或转给另一位审核者）。
• 反馈的结构化：尽量让人工提供结构化的反馈（如从下拉框选择“通过”、“拒绝并重写”、“需修改XX部分”），而不是纯文本。这能简化后续AI处理反馈的逻辑。你可以在State中定义专门的字段来存储结构化反馈。

4. 多智能体协作：构建角色化团队工作流

单个“全能”的 Agent 往往力有不逮。更强大的模式是模拟一个团队，让多个各司其职的 Agent 协作完成任务。LangGraph 的图结构非常适合对这种协作关系进行建模。

4.1 设计模式：管理者与工作者

一种常见的设计模式是“管理者（Supervisor）-工作者（Worker）”。

• 管理者 Agent：负责理解总体任务，进行任务规划与分解，并将子任务分配给特定的工作者，最后汇总和评估结果。
• 工作者 Agent：是领域专家，负责执行具体的子任务，如编写代码、检索信息、分析数据等。

在 LangGraph 中，管理者和工作者都可以实现为一个节点（Node）。管理者节点根据当前状态和任务，决定下一步调用哪个工作者节点，这通过条件边（Conditional Edge）来实现。

4.2 实战：构建一个技术调研团队

假设我们要组建一个团队来调研“向量数据库的最新进展”，团队包括：

1. 规划师（Planner）：分解调研任务。
2. 研究员（Researcher）：使用网络搜索工具查找信息。
3. 分析师（Analyst）：对搜集的信息进行归纳总结。
4. 写手（Writer）：将分析结果整理成格式良好的报告。

from typing import List from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun from langgraph.graph import StateGraph, END # 定义状态 class ResearchState(TypedDict): original_query: str # 原始问题 plan: List[str] # 调研计划 research_results: List[str] # 搜集到的资料 analysis: str # 分析结论 report: str # 最终报告 current_role: str # 当前正在执行的角色 max_turns: int = 10 # 最大循环次数，防止无限循环 # 初始化工具和模型 search = DuckDuckGoSearchRun() llm = ChatOpenAI(model="gpt-4-turbo") # 1. 规划师节点 def planner_node(state: ResearchState): query = state["original_query"] prompt = f""" 你是一个资深技术调研规划师。请将以下调研任务分解为3-5个具体的、可执行的子任务。 任务：{query} 请以清晰的列表形式返回子任务。 """ messages = [SystemMessage(content="你是一个高效的任务规划师。"), HumanMessage(content=prompt)] response = llm.invoke(messages) # 假设返回格式是 “1. ...\n2. ...” plan = [line.strip() for line in response.content.split('\n') if line.strip().startswith(('1.', '2.', '3.', '4.', '5.', '-'))] return {"plan": plan, "current_role": "planner"} # 2. 研究员节点 def researcher_node(state: ResearchState): # 从计划中取出第一个未完成的调研点（这里简化处理，每次执行一个） # 实际中，状态里可能需要一个 `current_plan_index` if not state.get("plan"): return {"research_results": [], "current_role": "researcher"} # 简化：用第一个计划项去搜索 search_query = state["plan"][0] + " 最新进展 2024" result = search.run(search_query) # 将结果累积到 research_results 中 current_results = state.get("research_results", []) current_results.append(f"调研点：{state['plan'][0]}\n结果：{result[:500]}...") # 截断 return {"research_results": current_results, "current_role": "researcher"} # 3. 分析师节点 def analyst_node(state: ResearchState): all_results = "\n---\n".join(state.get("research_results", [])) prompt = f""" 你是一个技术分析师。请根据以下搜集到的资料，总结关于“{state['original_query']}”的核心观点、技术对比和趋势。 资料： {all_results} 请给出结构化的分析摘要。 """ messages = [SystemMessage(content="你是一个洞察力强的技术分析师。"), HumanMessage(content=prompt)] response = llm.invoke(messages) return {"analysis": response.content, "current_role": "analyst"} # 4. 写手节点 def writer_node(state: ResearchState): analysis = state.get("analysis", "") prompt = f""" 你是一名技术文档写手。请将以下分析内容，整理成一篇适合发布在技术博客上的简短报告。 要求：有标题、引言、核心发现（分点论述）、总结。 分析内容： {analysis} """ messages = [SystemMessage(content="你是一名优秀的科技文章写手。"), HumanMessage(content=prompt)] response = llm.invoke(messages) return {"report": response.content, "current_role": "writer"} # 5. 路由逻辑（管理者逻辑） # 这是一个决定下一个节点是谁的函数 def route_after_planner(state: ResearchState) -> str: # 规划完成后，如果有计划项，就去研究；否则直接分析（可能不需要研究） if state.get("plan"): return "researcher" else: return "analyst" def route_after_researcher(state: ResearchState) -> str: # 研究员完成后，检查是否还有未完成的计划项（这里简化：只执行一次研究） # 实际中，这里可以更复杂，比如循环执行直到计划完成。 # 现在我们假设研究一次后就去分析。 return "analyst" def route_after_analyst(state: ResearchState) -> str: # 分析完成后，交给写手 return "writer" def route_after_writer(state: ResearchState) -> str: # 写手完成后，结束 return END # 构建图 builder = StateGraph(ResearchState) builder.add_node("planner", planner_node) builder.add_node("researcher", researcher_node) builder.add_node("analyst", analyst_node) builder.add_node("writer", writer_node) builder.set_entry_point("planner") # 使用条件边进行路由 builder.add_conditional_edges( "planner", route_after_planner, {"researcher": "researcher", "analyst": "analyst"} # 路由函数返回字符串，映射到节点名 ) builder.add_conditional_edges( "researcher", route_after_researcher, {"analyst": "analyst"} ) builder.add_conditional_edges( "analyst", route_after_analyst, {"writer": "writer"} ) builder.add_edge("writer", END) # 写手之后固定结束 graph = builder.compile() # 运行工作流 initial_state = {"original_query": "向量数据库的最新技术进展", "max_turns": 10} final_state = graph.invoke(initial_state) print("=== 最终报告 ===") print(final_state["report"])

4.3 多 Agent 协作的挑战与优化

1. 共享上下文与信息隔离： 所有 Agent 共享同一个State。这有利于信息传递，但也可能导致“信息过载”或“意外修改”。好的实践是：

• 在State中为每个 Agent 设计清晰的命名空间，例如researcher_findings,analyst_insights。
• 使用Annotated类型和operator.add来安全地追加列表，而不是覆盖。
• 对于敏感信息，可以考虑在子图中进行隔离。

2. 循环与终止条件： 多 Agent 协作容易陷入循环（例如，研究员和分析师互相要求对方提供更多信息）。必须设置明确的终止条件：

• 在State中设置max_turns或max_iterations计数器。
• 在路由逻辑中检查任务是否已完成（例如，所有计划项是否都已处理）。
• 使用“投票”或“共识”机制，当多个 Agent 认为可以结束时才结束。

3. 子图（Subgraph）封装复杂逻辑： 如果一个工作者 Agent 的内部逻辑非常复杂（例如，它自己又是一个包含多步骤的规划-执行循环），你可以将其封装成一个子图。主图只调用这个子图节点，子图内部可以有自己的状态和逻辑。这极大地提升了模块化和可维护性。使用add_node时传入一个已编译的Graph对象即可。

# 假设我们有一个复杂的代码生成子图 code_gen_graph = ... # 另一个编译好的StateGraph builder = StateGraph(MainState) # 将子图作为一个节点加入 builder.add_node("code_generator", code_gen_graph)

4. 异步与并行执行： 上面的例子是顺序执行的。LangGraph 支持基于状态键更新的并行执行。如果researcher和analyst节点不依赖于对方的输出（即它们修改的是State中不同的键），你可以通过配置让它们同时运行，这在处理独立子任务时能大幅提升效率。这需要在定义节点和边时进行精细设计。

5. 进阶整合：构建一个带检查点、人工介入与多 Agent 的生产级工作流

让我们将前面三个概念整合起来，设计一个更贴近实际需求的场景：一个自动化代码审查与合并工作流。

需求描述：

1. 当有新的 Pull Request (PR) 时，自动触发工作流。
2. 代码分析 Agent检查代码风格和潜在 Bug。
3. 测试生成 Agent尝试为变更生成单元测试。
4. 安全扫描 Agent进行基础的安全漏洞检查。
5. 将三个 Agent 的结果汇总，生成一份审查报告。
6. 如果报告中发现关键问题（如高严重性安全漏洞），则暂停工作流，等待项目负责人人工确认。
7. 负责人可以给出“驳回”、“忽略并继续”、“添加评论”等指令。
8. 根据人工指令或自动判断，工作流决定是自动合并、关闭 PR还是添加评论后等待。
9. 整个流程需要支持中断恢复（比如负责人可能几天后才处理）。

这个工作流涵盖了多 Agent 协作（分析、测试、安全）、人工介入（负责人审核）和状态持久化（支持长时间暂停）。其 LangGraph 实现的核心骨架如下：

# 状态设计 class CodeReviewState(TypedDict): pr_id: str code_diff: str style_issues: List[str] generated_tests: str security_findings: List[dict] # 每个finding包含 {“level”: “high/medium/low“, “description”: “...”} review_report: str human_decision: Literal["approve", "reject", "comment", None] = None human_comment: str = "" final_action: Literal["merge", "close", "comment", None] = None status: str = “initialized” # 图结构概览 (伪代码) builder = StateGraph(CodeReviewState) builder.add_node(“analyze_code”, analyze_code_node) # 代码分析Agent builder.add_node(“generate_tests”, generate_tests_node) # 测试生成Agent builder.add_node(“scan_security”, scan_security_node) # 安全扫描Agent builder.add_node(“compile_report”, compile_report_node) # 报告汇总Agent builder.add_node(“wait_for_human”, wait_for_human_node) # 人工介入节点（可能只是一个标记） builder.add_node(“execute_decision”, execute_decision_node) # 执行最终动作的Agent builder.set_entry_point(“analyze_code”) # 并行执行分析、测试、安全扫描（如果它们更新不同的状态键） builder.add_edge(“analyze_code”, “generate_tests”) builder.add_edge(“generate_tests”, “scan_security”) # 或者设计成真正的并行，需要更精细的状态键配置 builder.add_edge(“scan_security”, “compile_report”) # 条件边：根据报告决定是否需要人工介入 def after_report(state): has_critical_issue = any(f[“level”] == “high” for f in state.get(“security_findings”, [])) if has_critical_issue: return “wait_for_human” # 跳转到人工介入节点 else: return “execute_decision” # 自动执行 builder.add_conditional_edges( “compile_report”, after_report, {“wait_for_human”: “wait_for_human”, “execute_decision”: “execute_decision”} ) # 人工介入节点：这里不执行实际逻辑，只是设置一个中断点。 # 实际的人工交互在外部系统（如Webhook）处理。 def wait_for_human_node(state): # 这个节点可以什么都不做，或者只是更新状态表示正在等待 return {“status”: “awaiting_human_review”} # 在 ‘wait_for_human’ 节点之后设置中断 # 当执行到这里，图会暂停，等待外部系统调用 update_state 来提供 human_decision 和 human_comment builder.add_edge(“wait_for_human”, “execute_decision”) # 最终决策节点 def execute_decision_node(state): decision = state.get(“human_decision”) if decision == “approve” or (decision is None and not state.get(“has_critical_issue”)): # 调用GitHub API合并PR action = “merge” elif decision == “reject”: # 调用GitHub API关闭PR action = “close” else: # comment 或其他情况 # 调用GitHub API添加评论 action = “comment” return {“final_action”: action, “status”: “completed”} builder.add_edge(“execute_decision”, END) # 编译图，并启用 SQLite 检查点 memory = SqliteSaver.from_conn_string(“code_review.db”) graph = builder.compile( checkpointer=memory, interrupt_after=[“wait_for_human”] # 在等待人工节点后中断 )

这个工作流可以通过一个 Web 服务器来驱动：

1. GitHub Webhook 收到 PR 事件，触发graph.invoke(initial_state, config={“thread_id”: pr_id})。
2. 工作流运行到wait_for_human节点后中断，状态被保存。
3. 服务器向项目负责人的聊天工具（如 Slack）发送通知，并提供一个审批链接。
4. 负责人点击链接，在网页上看到审查报告并做出决定。
5. 网页后端调用graph.update_state(config, {“human_decision”: “approve”, “human_comment”: “LGTM”})，然后再次graph.invoke({}, config)。
6. 工作流从中断处继续，执行execute_decision_node，完成合并操作。

通过这种方式，我们利用 LangGraph 构建了一个鲁棒、可持久化、支持人机交互、由多个智能体协同的自动化流程，这正是现代 AI 应用工程化的核心所在。