LangGraph

LangGraph 的状态图模型与工作流控制¶

🧩 1. LangGraph 的核心概念是什么？State、Node、Edge 三者什么关系？¶

LangGraph 是一个用有向图来编排 Agent 行为的框架。它把 Agent 的每一次决策、每一个工具调用、每一段思考都抽象为图的节点，把控制流和数据流抽象为图的边。整张图共享一份状态。

一图胜千言：

三者的职责与关系：

State（状态）：一份贯穿全图的共享数据字典（TypedDict 或 Pydantic 模型）。每个节点都可以读它、写它。它承载了 Agent 的全部上下文：对话历史、中间结果、控制标志等。
Node（节点）：一个 Python 函数，接收当前 State，返回更新后的 State（或部分更新）。它可以是一个 LLM 调用、一个工具执行、一段业务逻辑。
Edge（边）：定义节点之间的流转规则。普通边就是“A 执行完一定去 B”；条件边是“A 执行完后，根据 State 的某个字段值决定去 B 还是去 C”。

关系一句话：State 是大脑的记忆，Node 是会做的事情，Edge 是做完一件事之后接着做什么的决定。

最简单的代码骨架：

from typing import TypedDict
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: list
    next_step: str

def node_a(state):
    # 处理逻辑...
    return {"messages": state["messages"] + ["A done"]}

def node_b(state):
    return {"messages": state["messages"] + ["B done"]}

def router(state):
    if "done" in state["messages"][-1]:
        return END
    return "node_b"

graph = StateGraph(AgentState)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)
graph.set_entry_point("node_a")
graph.add_conditional_edges("node_a", router, {"node_b": "node_b", END: END})
graph.add_edge("node_b", "node_a")  # 循环
app = graph.compile()

这就是 LangGraph 的世界观：用图来编程 Agent 的行为，而 State 是这张图的血液。

📦 2. LangGraph 的 State 是什么？如何设计一个合理的 AgentState？¶

State 是整个图运行时所有节点都能访问的共享内存。它在 Python 里通常是一个 TypedDict 或 Pydantic 模型，每经过一个节点，State 都可能被更新（增量地覆盖部分字段）。

设计一个合理的 AgentState，要遵循四个原则：

承载全局上下文：所有节点需要用到的信息，都必须存在于 State 中。
字段尽量扁平，避免深层嵌套：节点更新 State 是浅合并的，嵌套太深容易出错，且不利于调试。
区分“持久数据”和“控制字段”：messages 这类持续累积的是数据，next_step、should_continue 这类决定图流向的是控制字段。
为 Checkpoint 和恢复留好位子：比如可以加一个 user_approval 字段，用于中断时暂存外部输入。

好设计 vs 坏设计：

# ❌ 差：把所有东西丢进一个大字典，没有类型
state = {"everything": [...]}

# ❌ 差：控制字段和数据字段混在一起，名字模糊
class BadState(TypedDict):
    flag: bool
    temp: str
    data: dict

# ✅ 好：职责清晰，字段有类型
from typing import TypedDict, List, Optional
from langgraph.graph.message import add_messages

class GoodAgentState(TypedDict):
    # 数据字段
    messages: List[dict]          # 对话历史（用 add_messages 合并）
    user_query: str               # 原始用户输入
    intermediate_results: dict    # 工具返回的中间结果
    final_answer: Optional[str]   # 最终答案

    # 控制字段
    next_agent: str               # 下一个要激活的节点
    should_continue: bool         # 是否继续循环
    error_count: int              # 连续失败次数

三个典型场景的 State 设计：

① 简单的 Chatbot

class ChatState(TypedDict):
    messages: list  # 只需要对话历史

② ReAct Agent

class ReActState(TypedDict):
    messages: list
    tools_results: dict
    iteration: int
    final_answer: Optional[str]

③ 需要人工审批的 Supervisor

class SupervisorState(TypedDict):
    messages: list
    worker_outputs: dict
    pending_action: Optional[str]   # 待审批的动作
    user_decision: Optional[str]    # 用户在中断时输入的决策

设计时的铁律：

每次你想加一个字段，先问自己：“哪个节点需要它？”如果只有一个节点需要，那它可能不该放在全局 State 里，而是该节点内部的局部变量。State 只放跨节点共享的东西。

💾 3. LangGraph 的 Checkpoint 机制是什么？如何实现 Agent 的中断和恢复？¶

Checkpoint（检查点）是 LangGraph 在每个步骤后自动保存的 State 快照。它是图执行持久化的基石，让你可以：

暂停：在某个节点前插入 interrupt()，图运行到这就停止，等待外部输入。
恢复：拿到外部输入后，从断点继续跑下去，State 完好无损。
时间旅行：回溯到之前某个 Checkpoint，重放或分析当时的 State

流程：节点A → [Checkpoint] → 节点B → [Checkpoint] → ...
                     ↑ 可以在这中断
                     外部输入 → 从 Checkpoint 恢复 → 继续执行

Checkpoint 存储什么？

当前的 State（所有字段的值）
图的位置（哪个节点刚执行完，下一个该谁）
配置 ID（thread_id，用于区分不同会话）

内置 Checkpointer：

MemorySaver：存内存，适合开发测试。
SqliteSaver：存 SQLite 文件，适合单机生产。
也可以自己对接 Redis、Postgres 等。

实现中断和恢复的完整示例：人工审批场景

场景：Agent 生成了一封邮件草稿，必须经人工审批后才能发送。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import interrupt  # 中断函数

class EmailState(TypedDict):
    draft: str
    approved: bool

# 节点1：生成邮件草稿
def draft_email(state: EmailState):
    return {"draft": "您好，这是自动生成的合同续约通知..."}

# 节点2：发送邮件（只有审批通过才会执行）
def send_email(state: EmailState):
    # 实际发送逻辑
    return {"draft": state["draft"] + "\n[已发送]"}

# 构建图
graph = StateGraph(EmailState)
graph.add_node("draft", draft_email)
graph.add_node("review", lambda state: state)  # 占位节点，真正的中断在边上
graph.add_node("send", send_email)
graph.set_entry_point("draft")

# 关键：从 draft 到 send 的边上定义中断
def human_review(state: EmailState):
    # interrupt 会暂停整个图，把消息返回给外部
    approval = interrupt(f"请审批以下邮件草稿：\n{state['draft']}\n输入 'yes' 或 'no'")
    if approval == "yes":
        return "send"
    return END

graph.add_conditional_edges("draft", human_review, {"send": "send", END: END})

# 编译，带上 checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# --- 第一次运行：生成草稿，然后中断 ---
config = {"configurable": {"thread_id": "email-001"}}
for event in app.stream({"draft": "", "approved": False}, config):
    print(event)
# 输出到中断处，外部看到提示：“请审批以下邮件草稿...”
# 图在这里暂停，State 被持久化到 MemorySaver

# --- 此时可以关掉进程、过几小时再恢复 ---

# 恢复：向图注入用户的审批决定
from langgraph.types import Command
# 用 Command(resume=...) 传入中断的返回值
resume_value = "yes"  # 假设用户点了“同意”
for event in app.stream(Command(resume=resume_value), config):
    print(event)
# 图从断点继续，human_review 返回 "send"，然后执行 send 节点

核心机制解析：

interrupt(message) 让图在当前节点后暂停，向外部抛出 GraphInterrupt，并保存 Checkpoint。
外部拿到中断消息，向用户展示，等待用户输入。
外部用 Command(resume=value) 重新调用 app.stream()，value 就是 interrupt 的返回值。
thread_id 是区分不同对话的关键，恢复时必须用同一个 config。

Checkpoint 的更高级用法：时间旅行

# 列出所有历史 Checkpoint
checkpoints = list(checkpointer.list(config))
for cp in checkpoints:
    print(cp.config["configurable"]["checkpoint_id"], cp.metadata)

# 恢复到之前某个 Checkpoint 继续运行
old_config = {"configurable": {"thread_id": "email-001", "checkpoint_id": "xxx"}}
app.stream(Command(resume="yes"), old_config)  # 从历史点重新执行

收束：

LangGraph 的 Checkpoint 机制让 Agent 从“一跑到底”变成“可以随时按暂停、改参数、再继续”的受控流程。这意味着你可以把人工决策、外部审批、长时间的异步任务毫无缝地嵌入 Agent 的执行图中。它不是简单的日志记录，而是把整个运行时虚拟化，为 Agent 赋予了“任意时间点均可操作”的工程能力。

LangGraph 如何实现 Human-in-the-Loop（人机协作）？¶

难度：⭐⭐⭐（中断机制、人工审核、工作流编排）

1️⃣ Common Answer：

Human-in-the-Loop 就是让 Agent 在某些节点停下来，等人工确认后再继续。LangGraph 可以设置 interruptbefore 或 interruptafter 参数，在特定节点前后暂停。人可以修改 State，然后 Agent 继续执行。

2️⃣ Impressive Answer：

LangGraph 的 Human-in-the-Loop 基于中断机制 + Checkpoint 持久化，实现了 Agent 与人的无缝协作。核心是在工作流的关键节点插入"人工确认点"，让人类可以审核、修正或引导 Agent 的行为。

三种典型模式：

模式一：审核确认模式（适合敏感操作）

from langgraph.graph import StateGraph, END

# 在执行支付前中断，等待人工确认
app = workflow.compile(
    checkpointer=PostgresCheckpointSaver.from_conn_string(DB_URL),
    interrupt_before=["execute_payment"]
)

# Agent 执行到 execute_payment 前会暂停
config = {"configurable": {"thread_id": "order_456"}}
result = app.invoke({"messages": [{"role": "user", "content": "支付 100 元"}]}, config)
# 此时 result.status == "interrupted"

# 人工审核后继续执行
approved_state = app.get_state(config)
approved_state.values["approval_status"] = "approved"
app.update_state(config, approved_state.values)
result = app.invoke(None, config)  # 继续执行

模式二：人工修正模式（适合 Agent 出错）

# Agent 生成错误回答，人工修正后继续
state = app.get_state(config)
state.values["messages"].append({
    "role": "human",
    "content": "你的回答有误，正确答案是..."
})
app.update_state(config, state.values)

模式三：人工引导模式（适合复杂决策）

# 在决策节点插入人工选择，决策后暂停让人工确认
app = workflow.compile(
    checkpointer=memory,
    interrupt_after=["decide_strategy"]
)

工程实践要点：

状态可视化：通过 app.get_state(config) 获取当前 State，展示给人工审核
超时处理：设置人工确认的超时时间，超时后自动降级或拒绝
审计日志：记录每次人工干预的操作人、时间、修改内容
权限控制：结合 RBAC，只有授权用户才能执行确认操作

3️⃣ Key Differences

LangGraph

LangGraph 的状态图模型与工作流控制¶

🧩 1. LangGraph 的核心概念是什么？State、Node、Edge 三者什么关系？¶

📦 2. LangGraph 的 State 是什么？如何设计一个合理的 AgentState？¶

💾 3. LangGraph 的 Checkpoint 机制是什么？如何实现 Agent 的中断和恢复？¶

LangGraph 如何实现 Human-in-the-Loop（人机协作）？¶

用 LangGraph 实现多工具 ReAct Agent，如何处理工具调用失败和重试？¶

LangGraph 的条件边如何实现动态路由？¶

LangGraph 与 CrewAI 的编排思路有什么本质区别？¶

用 LangGraph 实现一个带人工审批节点的 Agent 工作流，如何设计？¶

LangGraph 的 Checkpointer 机制是什么？它如何实现状态持久化和时间旅行调试？MemorySaver 和 PostgresSaver 有什么区别？⭐⭐⭐¶

LangGraph 中如何实现子图（Subgraph）？子图和父图之间的状态如何传递和隔离？⭐⭐⭐¶

用 LangGraph 实现一个带重试、降级和超时控制的多步骤 Agent 工作流（如：数据采集→清洗→分析→报告生成），如何设计错误处理和容错机制？⭐⭐⭐⭐¶

在 LangGraph 里如何实现"LLM 有工具调用就去执行工具，否则结束"这个 ReAct 循环？¶

LangGraph Checkpointing 与持久化记忆¶

LangGraph 的 MemorySaver 和 PostgresSaver 有什么区别？¶

LangGraph 的 Checkpointing 机制是如何工作的？thread_id 的作用是什么？生产环境如何选持久化方案？¶

用户的多轮对话跨了多次 HTTP 请求，LangGraph 如何实现无需客户端传历史的记忆？¶

4、容易一起考的题¶

LangGraph Human-in-the-Loop：interrupt 与 resume 机制¶

1、LangGraph 的 interrupt_before 是什么？¶

2、LangGraph 如何实现 Human-in-the-Loop？请解释 interrupt() 的工作原理和 Command(resume=...) 恢复机制？¶

3、金融 Agent 要执行一笔大额转账前需要人工确认，如何用 LangGraph 设计这个审核流程？¶

5.3 LangGraph 多 Agent 架构¶

LangGraph Multi-Agent Supervisor 模式¶

1、Supervisor 模式和普通顺序图有什么区别？¶

2、LangGraph Multi-Agent Supervisor 模式的设计思路是什么？Supervisor 如何动态路由任务，子 Agent 如何返回结果？¶

3、自动化研究报告生成系统需要搜索、分析、写作三类能力，如何用 Supervisor 模式设计？¶

LangGraph 子图（Subgraph）组合与模块化设计¶

1、LangGraph 子图是什么？¶

2、LangGraph 子图如何实现父子 State 的隔离与通信？它解决了什么工程问题？¶

3、一个 Agent 系统需要同时查金融知识库和法律知识库，如何用子图实现并行双路 RAG？¶

LangGraph 的错误处理与节点级重试机制¶

1、LangGraph 节点抛出异常时的默认行为是什么？¶

2、如何在 LangGraph 中设计节点级重试策略和错误状态路由，构建健壮的 Agent 错误处理体系？¶

3、Agent 在调用外部 API 时频繁遇到限流（429 错误），如何在 LangGraph 层面系统性解决？¶

2、LangGraph 如何实现 Human-in-the-Loop？请解释 `interrupt()` 的工作原理和 `Command(resume=...)` 恢复机制？¶