LangGraph Human-in-the-Loop：关键时刻”踩刹车”

让 AI 在需要时停下来，让人类做决定——这才是真正的智能协作。

引言：为什么需要人工介入？

想象一下，你正在驾驶一辆自动驾驶汽车。大多数时候，它可以自己开得很好，但遇到复杂路况时，它会减速并提示你接管。这种”关键时刻人工介入”的设计理念，在 AI Agent 开发中同样重要。

完全自动化的陷阱

当我们构建 AI Agent 时，很容易陷入一个误区：让 AI 做一切。但现实世界远比代码复杂：

  • 高风险决策：银行转账、医疗诊断、合同签署——这些操作一旦出错，代价高昂
  • 模糊的边界：退款金额超过1000元怎么办？客户投诉涉及敏感话题如何处理？
  • 合规要求：某些业务场景必须人工确认才能继续

LangGraph 的 Human-in-the-Loop（HITL）机制，就是为解决这些问题而生。它不是让 AI 变弱，而是让系统更可靠、更可控。

HITL 的核心价值

场景 完全自动化的问题 HITL 的解决方案
退款审批 AI 可能批准可疑的退款 大额退款人工复核
内容审核 AI 可能误判边界内容 争议内容人工裁决
医疗建议 AI 可能给出错误诊断 关键节点医生确认
财务操作 AI 可能执行错误交易 转账前人工授权

简单来说，HITL 是 AI 的”双保险”——让机器处理它能处理好的事情，在关键时刻让人类介入。

理解 LangGraph 的中断机制

在深入代码之前，我们需要先理解 LangGraph 中 HITL 的两个核心概念：interrupt() 和 Command(resume)。

什么是 Interrupt？

想象你正在看一部电影，突然画面暂停，屏幕上出现一行字：”请输入解锁密码继续观看”。这就是中断——流程暂停，等待外部输入。

在 LangGraph 中，interrupt() 函数的作用就是让图执行暂停，并保存当前状态。等到人类提供输入后，流程可以继续执行。

1
2
3
4
5
6
7
8
from langgraph.types import interrupt

# 当执行到这里时，图会暂停
user_approval = interrupt(
    # 向人类展示的信息
    {"message": "客户申请退款 2000 元，是否批准？"}
)
# user_approval 就是人类输入的值

Command(resume) 的作用

当中断发生时，整个图处于”暂停”状态。要让它继续执行，我们需要使用 Command(resume=...)：

1
2
3
4
5
from langgraph.types import Command

# 恢复执行，并提供人类输入
graph.invoke(Command(resume=True))  # 批准
graph.invoke(Command(resume=False))  # 拒绝

这里的关键点是：LangGraph 会记住中断时的状态，并从断点继续执行，而不是从头开始。这就像一个书签，你可以随时回到标记的位置。

状态管理的魔法

为什么 LangGraph 能做到”从断点继续”？这得益于它的状态管理机制。每个节点的执行结果都会更新状态，而状态是持久化的：

1
2
3
4
5
6
7
8
9
10
11
12
13
[开始] → [检查订单] → [状态: order_id=123, amount=2000]
              ↓
        [interrupt: 等待审批]
              ↓
    [暂停，状态保存到数据库]
              ↓
    [人类输入: approve=True]
              ↓
    [Command(resume)]
              ↓
        [处理退款] → [状态: refund_processed=True]
              ↓
            [结束]

即使服务器重启，LangGraph 也能从持久化存储中恢复状态，让工作流从中断处继续。

设计优雅的中断点

不是所有地方都需要中断。设计好的中断点，是 HITL 成功的关键。

中断点的选择原则

应该设置中断点的地方：

  1. 高风险操作前 —— 任何不可逆或代价高昂的操作
  2. 模糊决策点 —— AI 无法确定最佳行动方案时
  3. 合规检查点 —— 法规要求人工确认的业务环节
  4. 异常处理 —— 流程遇到意外情况时

不应该设置中断点的地方：

  1. 简单的数据转换 —— 纯计算操作，人工干预没有意义
  2. 高频重复操作 —— 会导致人工疲劳，降低效率
  3. 已经验证的安全路径 —— 经过充分测试的标准流程

中断信息的友好设计

当中断发生时，向人类展示的信息要足够清晰，让决策者能快速理解情况：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ❌ 不好的做法
interrupt({"data": order_data})

# ✅ 好的做法
interrupt({
    "type": "refund_approval",
    "title": "退款审批申请",
    "order_id": "ORD-2024-001",
    "customer": "张三",
    "refund_amount": 2000.00,
    "reason": "商品质量问题",
    "risk_score": "中等",
    "suggested_action": "建议批准",
    "required_decision": "是否批准此退款申请？"
})

好的中断信息应该包含：

  • 上下文：这是什么决策？
  • 关键数据：需要哪些信息来做决定？
  • 建议：AI 的推荐是什么？（帮助人类，但人类可以否决）
  • 行动选项：明确的选择是什么？

实战：退款审批工作流

现在，让我们实现一个完整的退款审批系统。这个工作流演示了自动处理和人工介入的结合使用。

场景设计

假设我们运营一个电商平台，客户可以申请退款。我们设计这样的流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
┌─────────────────────────────────────────────────────────────┐
│                    退款审批工作流                            │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│   [开始]                                                     │
│     │                                                        │
│     ▼                                                        │
│   [验证申请] ──► 验证失败 ──► [拒绝退款]                       │
│     │                                                        │
│     ▼                                                        │
│   [自动风险评估]                                              │
│     │                                                        │
│     ├── 金额 ≤ 100元 ──► [自动批准] ──► [执行退款]            │
│     │                                                        │
│     ├── 金额 ≤ 500元 ──► [快速审批] ──► 批准? ──► [执行退款]  │
│     │                        │                               │
│     │                        └── 拒绝? ──► [拒绝退款]         │
│     │                                                        │
│     └── 金额 > 500元 ──► [人工审批] ──► 批准? ──► [执行退款]  │
│                               │                              │
│                               └── 拒绝? ──► [拒绝退款]        │
│                                                              │
│     ▼                                                        │
│   [发送通知]                                                 │
│     │                                                        │
│     ▼                                                        │
│   [结束]                                                     │
│                                                              │
└─────────────────────────────────────────────────────────────┘

完整代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
# refund_workflow.py
import json
from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver


# ═══════════════════════════════════════════════════════════════
# 1. 定义状态结构
# ═══════════════════════════════════════════════════════════════

class RefundState(TypedDict):
    """退款工作流的状态定义"""
    # 输入数据
    order_id: str
    customer_name: str
    refund_amount: float
    reason: str
    
    # 流程产生的数据
    validation_passed: Optional[bool]
    risk_level: Optional[Literal["low", "medium", "high"]]
    approval_status: Optional[Literal["pending", "approved", "rejected"]]
    approved_by: Optional[str]  # "auto" 或人工ID
    processed_at: Optional[str]
    
    # 最终结果
    result: Optional[dict]


# ═══════════════════════════════════════════════════════════════
# 2. 定义节点函数
# ═══════════════════════════════════════════════════════════════

def validate_application(state: RefundState) -> RefundState:
    """
    第一步：验证退款申请的基本信息
    - 检查金额是否为正
    - 检查必填字段
    """
    print(f"\n{'='*50}")
    print("📋 步骤 1：验证退款申请")
    print(f"{'='*50}")
    print(f"订单号: {state['order_id']}")
    print(f"客户: {state['customer_name']}")
    print(f"退款金额: ¥{state['refund_amount']}")
    print(f"原因: {state['reason']}")
    
    # 简单的验证逻辑
    is_valid = (
        state['refund_amount'] > 0 
        and len(state['reason']) >= 5
        and state['order_id'].startswith('ORD-')
    )
    
    state['validation_passed'] = is_valid
    
    if is_valid:
        print("✅ 验证通过")
    else:
        print("❌ 验证失败")
    
    return state


def assess_risk(state: RefundState) -> RefundState:
    """
    第二步：自动风险评估
    根据金额和原因判断风险等级
    """
    print(f"\n{'='*50}")
    print("🔍 步骤 2：风险评估")
    print(f"{'='*50}")
    
    amount = state['refund_amount']
    reason = state['reason'].lower()
    
    # 风险评级逻辑
    if amount <= 100:
        risk = "low"
        print(f"金额 ¥{amount} ≤ 100，风险等级：低")
    elif amount <= 500:
        # 中等金额，检查原因关键词
        high_risk_keywords = ['欺诈', '恶意', '诈骗', '刷']
        if any(kw in reason for kw in high_risk_keywords):
            risk = "high"
            print(f"检测到高风险关键词，风险等级：高")
        else:
            risk = "medium"
            print(f"金额 ¥{amount} 在100-500之间，风险等级：中")
    else:
        risk = "high"
        print(f"金额 ¥{amount} > 500，风险等级：高（需人工审批）")
    
    state['risk_level'] = risk
    return state


def auto_approve(state: RefundState) -> RefundState:
    """
    低风险情况：自动批准
    """
    print(f"\n{'='*50}")
    print("🤖 步骤 3a：自动批准")
    print(f"{'='*50}")
    print(f"低风险交易，系统自动批准")
    
    state['approval_status'] = "approved"
    state['approved_by'] = "auto"
    return state


def quick_approval_interrupt(state: RefundState) -> RefundState:
    """
    中等风险：快速审批（人工确认）
    """
    print(f"\n{'='*50}")
    print("⏸️ 步骤 3b：快速审批（等待人工输入）")
    print(f"{'='*50}")
    
    # 中断，等待人类决策
    decision = interrupt({
        "type": "quick_approval",
        "title": "快速审批",
        "message": "这是一笔中等金额的退款申请",
        "order_id": state['order_id'],
        "customer": state['customer_name'],
        "amount": f"¥{state['refund_amount']}",
        "reason": state['reason'],
        "risk_level": "中等",
        "suggested_action": "建议批准",
        "options": ["approve", "reject"],
        "timeout_seconds": 300  # 5分钟超时
    })
    
    # 人类输入后继续
    print(f"人工决策: {decision}")
    
    if decision == "approve":
        state['approval_status'] = "approved"
        state['approved_by'] = "human_quick"
    else:
        state['approval_status'] = "rejected"
        state['approved_by'] = "human_quick"
    
    return state


def manual_approval_interrupt(state: RefundState) -> RefundState:
    """
    高风险：人工详细审批
    """
    print(f"\n{'='*50}")
    print("⏸️ 步骤 3c：人工审批（等待人工输入）")
    print(f"{'='*50}")
    print("⚠️ 这是一笔大额退款，需要详细审核")
    
    # 更详细的中断信息
    decision = interrupt({
        "type": "manual_approval",
        "title": "⚠️ 高风险退款审批",
        "alert": "金额超过500元，请仔细审核",
        "order_details": {
            "订单号": state['order_id'],
            "客户": state['customer_name'],
            "退款金额": f"¥{state['refund_amount']}",
            "申请原因": state['reason'],
            "风险等级": "高"
        },
        "checklist": [
            "订单是否真实存在？",
            "退款原因是否合理？",
            "客户历史记录如何？",
            "金额是否准确？"
        ],
        "options": ["approve", "reject", "escalate"],
        "notes": "escalate 将提交给上级经理处理"
    })
    
    print(f"人工决策: {decision}")
    
    if decision == "approve":
        state['approval_status'] = "approved"
        state['approved_by'] = "human_manual"
    elif decision == "reject":
        state['approval_status'] = "rejected"
        state['approved_by'] = "human_manual"
    else:
        # escalate - 这里简化处理，实际可以触发另一个流程
        state['approval_status'] = "rejected"
        state['approved_by'] = "escalated"
    
    return state


def process_refund(state: RefundState) -> RefundState:
    """
    执行退款操作
    """
    print(f"\n{'='*50}")
    print("💰 步骤 4：执行退款")
    print(f"{'='*50}")
    
    if state['approval_status'] == "approved":
        # 模拟退款处理
        import datetime
        state['processed_at'] = datetime.datetime.now().isoformat()
        state['result'] = {
            "status": "success",
            "refund_id": f"RFD-{state['order_id']}",
            "amount": state['refund_amount'],
            "method": "原路退回",
            "eta": "1-3个工作日"
        }
        print(f"✅ 退款已处理")
        print(f"   退款ID: {state['result']['refund_id']}")
        print(f"   金额: ¥{state['result']['amount']}")
    else:
        state['result'] = {
            "status": "rejected",
            "reason": "未通过审批"
        }
        print("❌ 退款被拒绝")
    
    return state


def send_notification(state: RefundState) -> RefundState:
    """
    发送通知给客户
    """
    print(f"\n{'='*50}")
    print("📧 步骤 5：发送通知")
    print(f"{'='*50}")
    
    if state['approval_status'] == "approved":
        print(f"已发送批准通知给客户 {state['customer_name']}")
        print(f"退款将在 {state['result']['eta']} 内到账")
    else:
        print(f"已发送拒绝通知给客户 {state['customer_name']}")
    
    return state


def reject_refund(state: RefundState) -> RefundState:
    """
    拒绝退款
    """
    print(f"\n{'='*50}")
    print("❌ 退款被拒绝")
    print(f"{'='*50}")
    print("原因: 申请验证失败")
    
    state['approval_status'] = "rejected"
    state['result'] = {
        "status": "rejected",
        "reason": "申请信息验证失败"
    }
    
    return state


# ═══════════════════════════════════════════════════════════════
# 3. 路由函数：决定流程走向
# ═══════════════════════════════════════════════════════════════

def route_after_validation(state: RefundState) -> str:
    """验证后路由"""
    if state['validation_passed']:
        return "assess_risk"
    return "reject_refund"

def route_by_risk(state: RefundState) -> str:
    """根据风险等级路由"""
    risk = state['risk_level']
    if risk == "low":
        return "auto_approve"
    elif risk == "medium":
        return "quick_approval"
    else:
        return "manual_approval"

def route_after_approval(state: RefundState) -> str:
    """审批后路由"""
    # 无论批准还是拒绝，都要执行相应的处理
    if state['approval_status'] == "approved":
        return "process_refund"
    return "send_notification"


# ═══════════════════════════════════════════════════════════════
# 4. 构建工作流图
# ═══════════════════════════════════════════════════════════════

def create_refund_workflow():
    """创建退款审批工作流"""
    
    # 创建图构建器
    builder = StateGraph(RefundState)
    
    # 添加节点
    builder.add_node("validate", validate_application)
    builder.add_node("assess_risk", assess_risk)
    builder.add_node("auto_approve", auto_approve)
    builder.add_node("quick_approval", quick_approval_interrupt)
    builder.add_node("manual_approval", manual_approval_interrupt)
    builder.add_node("process_refund", process_refund)
    builder.add_node("send_notification", send_notification)
    builder.add_node("reject_refund", reject_refund)
    
    # 添加边
    builder.add_edge(START, "validate")
    
    # 验证后的条件分支
    builder.add_conditional_edges(
        "validate",
        route_after_validation,
        {
            "assess_risk": "assess_risk",
            "reject_refund": "reject_refund"
        }
    )
    
    # 风险评估后的条件分支
    builder.add_conditional_edges(
        "assess_risk",
        route_by_risk,
        {
            "auto_approve": "auto_approve",
            "quick_approval": "quick_approval",
            "manual_approval": "manual_approval"
        }
    )
    
    # 自动批准 -> 处理退款
    builder.add_edge("auto_approve", "process_refund")
    
    # 快速审批和人工审批后的条件分支
    builder.add_conditional_edges(
        "quick_approval",
        route_after_approval,
        {
            "process_refund": "process_refund",
            "send_notification": "send_notification"
        }
    )
    
    builder.add_conditional_edges(
        "manual_approval",
        route_after_approval,
        {
            "process_refund": "process_refund",
            "send_notification": "send_notification"
        }
    )
    
    # 处理退款 -> 发送通知
    builder.add_edge("process_refund", "send_notification")
    
    # 发送通知和拒绝退款 -> 结束
    builder.add_edge("send_notification", END)
    builder.add_edge("reject_refund", END)
    
    # 使用内存检查点（生产环境可使用 PostgresSaver）
    checkpointer = MemorySaver()
    
    return builder.compile(checkpointer=checkpointer)


# ═══════════════════════════════════════════════════════════════
# 5. 运行示例
# ═══════════════════════════════════════════════════════════════

def main():
    """演示三种不同的退款场景"""
    
    workflow = create_refund_workflow()
    config = {"configurable": {"thread_id": "demo"}}
    
    print("\n" + "═"*60)
    print("  🧪 场景 1：低风险自动退款（金额：80元）")
    print("═"*60)
    
    # 场景1：小额退款，自动处理
    result1 = workflow.invoke({
        "order_id": "ORD-2024-001",
        "customer_name": "张三",
        "refund_amount": 80.00,
        "reason": "商品与描述不符"
    }, config)
    
    print("\n📊 最终结果:")
    print(json.dumps(result1['result'], indent=2, ensure_ascii=False))
    
    # 场景2：中等风险，需要快速审批
    print("\n" + "═"*60)
    print("  🧪 场景 2：中等风险快速审批（金额：300元）")
    print("═"*60)
    
    config2 = {"configurable": {"thread_id": "demo-2"}}
    
    # 启动工作流，它会在中断点暂停
    result2 = workflow.invoke({
        "order_id": "ORD-2024-002",
        "customer_name": "李四",
        "refund_amount": 300.00,
        "reason": "商品质量问题"
    }, config2)
    
    print("\n⏸️ 工作流已暂停，等待人工输入...")
    print("模拟人类决策: 批准")
    
    # 模拟人类批准
    result2 = workflow.invoke(Command(resume="approve"), config2)
    
    print("\n📊 最终结果:")
    print(json.dumps(result2['result'], indent=2, ensure_ascii=False))
    
    # 场景3：高风险，需要人工详细审批
    print("\n" + "═"*60)
    print("  🧪 场景 3：高风险人工审批（金额：2000元）")
    print("═"*60)
    
    config3 = {"configurable": {"thread_id": "demo-3"}}
    
    # 启动工作流
    result3 = workflow.invoke({
        "order_id": "ORD-2024-003",
        "customer_name": "王五",
        "refund_amount": 2000.00,
        "reason": "收到商品损坏"
    }, config3)
    
    print("\n⏸️ 工作流已暂停，等待人工输入...")
    print("模拟人类决策: 批准")
    
    # 模拟人类批准
    result3 = workflow.invoke(Command(resume="approve"), config3)
    
    print("\n📊 最终结果:")
    print(json.dumps(result3['result'], indent=2, ensure_ascii=False))
    
    print("\n" + "═"*60)
    print("  ✅ 所有场景演示完成！")
    print("═"*60)


if __name__ == "__main__":
    main()

代码要点解析

1. 状态驱动设计

整个工作流基于 RefundState 状态对象。每个节点接收当前状态，修改它，然后返回。这种设计让流程非常清晰，也便于调试。

2. 条件路由

通过 add_conditional_edges 实现智能分流：

  • 验证失败 → 直接拒绝
  • 风险低 → 自动批准
  • 风险中 → 快速人工确认
  • 风险高 → 详细人工审批

3. 中断点的优雅实现

interrupt() 函数让流程暂停，等待外部输入。注意我们传入了结构化的信息，包含上下文、选项和建议，帮助决策者快速做出判断。

4. 持久化检查点

MemorySaver() 让状态保存在内存中（适合演示）。生产环境可以换成 PostgresSaver，将状态持久化到数据库，即使服务重启也能恢复。

运行结果

运行这段代码，你会看到三个场景的完整执行流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
════════════════════════════════════════════════════════════
  🧪 场景 1：低风险自动退款（金额：80元）
════════════════════════════════════════════════════════════

==================================================
📋 步骤 1：验证退款申请
==================================================
订单号: ORD-2024-001
客户: 张三
退款金额: ¥80.0
原因: 商品与描述不符
✅ 验证通过

==================================================
🔍 步骤 2：风险评估
==================================================
金额 ¥80.0 ≤ 100，风险等级：低

==================================================
🤖 步骤 3a：自动批准
==================================================
低风险交易，系统自动批准

==================================================
💰 步骤 4：执行退款
==================================================
✅ 退款已处理
   退款ID: RFD-ORD-2024-001
   金额: ¥80.0

...

进阶：构建人机协作界面

在实际生产环境中，中断点需要连接到一个用户界面。以下是几种常见的实现方式：

方式一：Web 仪表板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/pending-approvals')
def get_pending():
    """获取所有待审批的中断"""
    # 从检查点存储查询所有等待中的状态
    pending = checkpointer.list_pending_interrupts()
    return jsonify(pending)

@app.route('/api/approve/<thread_id>', methods=['POST'])
def approve(thread_id):
    """批准特定请求"""
    decision = request.json.get('decision')
    config = {"configurable": {"thread_id": thread_id}}
    
    # 恢复工作流
    result = workflow.invoke(Command(resume=decision), config)
    return jsonify({"status": "success", "result": result})

方式二：钉钉/企业微信通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import requests

def send_approval_notification(interrupt_data, thread_id):
    """发送审批通知到钉钉"""
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=xxx"
    
    message = {
        "msgtype": "action_card",
        "action_card": {
            "title": interrupt_data['title'],
            "markdown": f"""
## {interrupt_data['title']}

**订单号**: {interrupt_data['order_id']}
**客户**: {interrupt_data['customer']}
**金额**: {interrupt_data['amount']}
**原因**: {interrupt_data['reason']}

[批准](https://your-app.com/approve/{thread_id}?decision=approve)
[拒绝](https://your-app.com/approve/{thread_id}?decision=reject)
            """,
            "single_title": "查看详情",
            "single_url": f"https://your-app.com/detail/{thread_id}"
        }
    }
    
    requests.post(webhook_url, json=message)

方式三：邮件审批

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

def send_approval_email(interrupt_data, thread_id):
    """发送审批邮件"""
    approve_url = f"https://your-app.com/api/approve/{thread_id}?token=xxx"
    
    html_content = f"""
    <h2>{interrupt_data['title']}</h2>
    <p><strong>订单号:</strong> {interrupt_data['order_id']}</p>
    <p><strong>退款金额:</strong> {interrupt_data['amount']}</p>
    
    <div style="margin: 20px 0;">
        <a href="{approve_url}&decision=approve" 
           style="background: #4CAF50; color: white; padding: 12px 24px; 
                  text-decoration: none; border-radius: 4px; margin-right: 10px;">
            批准
        </a>
        <a href="{approve_url}&decision=reject"
           style="background: #f44336; color: white; padding: 12px 24px;
                  text-decoration: none; border-radius: 4px;">
            拒绝
        </a>
    </div>
    """
    
    message = Mail(
        from_email='approvals@your-company.com',
        to_emails='manager@your-company.com',
        subject=f"【审批请求】{interrupt_data['title']}",
        html_content=html_content
    )
    
    sg = SendGridAPIClient(os.environ.get('SENDGRID_API_KEY'))
    sg.send(message)

总结与下篇预告

核心要点回顾

今天我们学习了 LangGraph 的 Human-in-the-Loop 机制：

  1. interrupt() —— 在流程中创建暂停点，等待人类输入
  2. Command(resume=...) —— 提供人类决策，让流程继续执行
  3. 状态持久化 —— LangGraph 自动保存中断状态，可以从断点恢复
  4. 优雅的中断设计 —— 清晰的信息、明确的选项、合理的上下文

何时使用 HITL

✅ 使用 HITL ❌ 不需要 HITL
高风险操作前 纯数据处理
模糊的决策边界 确定的业务规则
合规要求 已充分测试的标准流程
异常情况 高频重复操作

生产环境建议

  1. 使用持久化检查点：生产环境使用 PostgresSaver 替代 MemorySaver
  2. 设置超时机制：中断不应该永远等待，设置合理的超时时间
  3. 记录审计日志：所有人工决策都要记录，便于合规审计
  4. 优雅的降级：当人类无法及时响应时，应该有默认处理方式

🚀 下篇预告

在《LangGraph 零基础入门》系列的下一篇中，我们将深入探讨 “持久化与记忆：让 Agent 拥有长期记忆”。

我们将学习：

  • Thread 管理 —— 如何管理多个对话线程
  • 检查点策略 —— 何时保存、如何恢复
  • 长期记忆 —— 让 Agent 记住用户偏好和历史
  • 多租户隔离 —— 确保用户数据安全隔离

敬请期待！

📌 完整代码已开源：https://github.com/your-username/langgraph-tutorials

💬 有问题？ 欢迎在评论区留言讨论！