摘要：本文是 Agent 实战系列 Phase 3，记录如何用 LangGraph 搭建代码审查多节点工作流：安全扫描、条件分支、interrupt 人工确认、LLM 生成报告。涵盖 route_after_security 路由、Checkpointer 持久化与 scanners 自定义规则，附完整踩坑记录。

关键词：Agent LangGraph 工作流 interrupt 人机协同 代码审查 条件分支

---

一、前言：为什么需要复杂工作流？

Phase 1 的工具 Agent 是 agent ↔ tools 循环——适合「问一句、调工具、答一句」。

Phase 2 的 RAG 是 retrieve → generate 流水线——适合「查文档、生成回答」。

但真实工业场景往往是固定多步骤流水线，且步骤之间有分支、有人工卡点：

PR 提交 → 解析 diff → 安全扫描 → 有严重问题？等人确认 → 风格检查 → 生成报告

这类场景用简单循环搞不定，需要 Phase 3 的复杂工作流：

能力	Phase 1/2	Phase 3
结构	循环 / 两节点流水线	多节点 DAG
分支	tools_condition	自定义路由函数
人工介入	无	interrupt 暂停
状态	messages	结构化 ReviewState

一句话：从「对话循环」进化到「业务流程编排」。

---

二、Phase 3 项目：代码审查 Agent

2.1 工作流全景图

START
  ↓
parse_diff（解析 diff，统计变更文件/新增行）
  ↓
security_scan（静态安全规则扫描）
  ↓
route_after_security（条件分支）
  ├─ 有严重问题 → human_review（interrupt 人工确认）
  │                    ↓
  │              route_after_human
  │                 ├─ approve → style_review
  │                 └─ reject  → reject_end → END
  └─ 无严重问题 → style_review（风格扫描）
                       ↓
                 generate_report（LLM 生成 Markdown 报告）
                       ↓
                     END

2.2 与 Phase 1/2 的本质区别

	Phase 1	Phase 2	Phase 3
拓扑	环（循环）	链（2 节点）	DAG（多节点 + 分支）
决策	LLM 决定调哪个工具	固定先检索后生成	代码规则 + 人工决策
暂停	无	无	interrupt 等人输入
LLM 作用	全程决策	最后生成	只在 generate_report 节点

---

三、核心机制一：条件分支 route_after_security

3.1 路由函数

安全扫描完成后，用条件边决定下一步走哪条路径：

def route_after_security(state: ReviewState) -> Literal["human_review", "style_review"]:
    if state.get("has_critical"):
        logger.info("[路由] 存在严重安全问题 → human_review")
        return "human_review"
    logger.info("[路由] 无严重问题 → style_review")
    return "style_review"

注册到图中：

workflow.add_conditional_edges("security_scan", route_after_security)

理解要点：

• 如果存在严重安全问题 → 进入人工审核节点 human_review
• 否则 → 直接进入风格检查节点 style_review
• 返回值必须是下游节点名，LangGraph 据此连边

3.2 第二道分支：route_after_human

人工审核后还有一道分支：

def route_after_human(state: ReviewState) -> Literal["style_review", "reject_end"]:
    if state.get("human_approved"):
        return "style_review"   # 人工批准，继续审查
    return "reject_end"         # 人工拒绝，终止流程

---

四、核心机制二：interrupt 人机协同

4.1 interrupt 是什么？

interrupt() 可以暂停流程图执行，并向客户端展示 payload 数据——可以是上下文信息，也可以是请求恢复执行所需的输入内容。

@trace_node("human_review")
def human_review(state: ReviewState) -> ReviewState:
    decision = interrupt({
        "action": "approve_critical_findings",
        "message": "发现严重安全问题，是否继续生成审查报告？",
        "issues": state.get("security_issues", []),
        "hint": "回复 approve 继续，reject 终止",
    })
    approved = str(decision).strip().lower() in {"approve", "y", "yes", "继续", "是"}
    return {"human_approved": approved}

运行 --sample risky 时，终端会暂停：

⏸️  工作流暂停：需要人工确认
发现严重安全问题，是否继续生成审查报告？
  🔴 行5 硬编码 API Key
  🔴 行6 硬编码密码
输入 approve 继续 / reject 终止
人工决策：

4.2 恢复执行

检测到 __interrupt__ 后，用 Command(resume=...) 恢复：

while result.get("__interrupt__"):
    decision = input("人工决策：").strip()
    result = app.invoke(Command(resume=decision), config=config)

4.3 多个 interrupt 的 resume 匹配

若一个节点中调用多个 interrupt()，LangGraph 会根据中断在节点内的出现顺序，将 resume 值与中断一一匹配。该 resume 值列表仅适用于当前这次 task，不会在不同 thread 之间共享。

本项目 human_review 只有一个 interrupt，所以 Command(resume="approve") 直接对应那一次暂停。

4.4 必须开启 Checkpointer

使用 interrupt 前必须开启检查点器——该特性依赖持久化存储图状态才能实现暂停与恢复：

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "review-1"}}

没有 Checkpointer，工作流无法在中断点保存状态，resume 会失败。

---

五、核心机制三：结构化状态 ReviewState

Phase 1 的状态只有 messages，Phase 3 扩展为结构化字段：

class ReviewState(TypedDict, total=False):
    diff_text: str           # 原始 diff
    files_changed: list[str] # 变更文件列表
    added_lines: int         # 新增行数
    security_issues: list    # 安全问题
    style_issues: list       # 风格问题
    has_critical: bool       # 是否有严重问题
    human_approved: bool     # 人工是否批准
    report: str              # 最终报告

每个节点只读写自己负责的字段，状态在节点间逐层累积。

---

六、静态扫描：scanners.py

6.1 安全规则表

Phase 3 的安全扫描不依赖 LLM，而是用正则规则扫描 diff 新增行：

SECURITY_RULES = [
    (r"\beval\s*\(", "critical", "使用 eval() 存在代码注入风险"),
    (r"password\s*=\s*['\"]", "critical", "硬编码密码"),
    (r"sk-[a-zA-Z0-9]{10,}", "critical", "疑似 API Key 泄露"),
    (r"\bos\.system\s*\(", "critical", "os.system 存在命令注入风险"),
    # 自定义规则：数据库硬编码
    (r"(db_|mysql|pg|database).*[\"'](root|admin).*@.*[\"']", "error",
     "代码硬编码数据库账号密码，存在数据泄露高危风险"),
]

6.2 严重级别与路由

def has_critical(issues: list[Issue]) -> bool:
    """critical / error 均视为严重问题，触发人工审核。"""
    return any(i["severity"] in {"critical", "error"} for i in issues)

踩坑记录：最初 has_critical 只认 "critical"，新增规则用了 "error" 级别，导致命中后不进人工审核。修复方式：在 has_critical 中统一维护「需人工审核」的级别集合。

6.3 如何添加自己的规则

三步：

1. 在 SECURITY_RULES 加一行 (正则, 级别, 描述)
2. 确认级别是 critical 或 error（会触发人工审核）
3. 用 --sample risky 或自定义 diff 验证

---

七、各节点职责

节点	类型	职责
parse_diff	处理	解析 diff，统计文件和行数
security_scan	扫描	正则匹配安全问题
route_after_security	路由	严重 → 人工 / 否则 → 风格
human_review	interrupt	暂停，等人 approve/reject
route_after_human	路由	批准 → 继续 / 拒绝 → 终止
style_review	扫描	print/TODO/行过长等
generate_report	LLM	汇总扫描结果，生成 Markdown 报告
reject_end	终止	输出拒绝信息

LLM 只在最后一个节点出场——前面全是确定性逻辑，成本低、可控、可审计。

---

八、环境准备与运行

8.1 快速开始

cd agent-workflow
copy ..\FirstAgent\.env .env
pip install -r requirements.txt

# 含安全问题，触发 interrupt
python agent.py --sample risky

# 干净 diff，跳过人工节点
python agent.py --sample clean

# 自定义 diff 文件
python agent.py --file your_diff.txt

8.2 测试用例

命令	预期路径	考察点
--sample risky	security → human_review → style → report	interrupt + 条件分支
--sample clean	security → style → report	跳过人工节点
输入 reject	终止，输出拒绝信息	人工拒绝分支
输入 approve	继续生成完整报告	人工批准分支

8.3 Trace 日志解读

>> 进入节点: security_scan
[安全] 发现 5 个问题，严重: True
[路由] 存在严重安全问题 → human_review
>> 进入节点: human_review
（暂停，等待人工输入）
[人工] 决策: 通过
[路由] 人工已通过 → style_review
>> 进入节点: style_review
>> 进入节点: generate_report
[报告] 已生成，长度 850 字符

---

九、Phase 1 → 2 → 3 进化对比

Phase 1  agent ↔ tools 循环
         「LLM 决策 + 工具执行」

Phase 2  retrieve → generate 流水线
         「检索 + 生成，固定两步」

Phase 3  多节点 DAG + 条件分支 + interrupt
         「业务流程编排 + 人工卡点」

进化维度	Phase 1	Phase 3
图结构	环	DAG
分支逻辑	LLM 决定	代码规则 + 人工
状态	messages	结构化 dict
暂停恢复	无	interrupt + Checkpointer
LLM 参与	全程	仅报告节点

---

十、踩坑记录

10.1 interrupt 不生效

原因：没开 Checkpointer，或 thread_id 不一致。

解决：workflow.compile(checkpointer=MemorySaver())，resume 时用同一个 thread_id。

10.2 新规则不触发人工审核

原因：规则 severity 用了 "error"，但 has_critical 只认 "critical"。

解决：统一严重级别集合 {"critical", "error"}。

10.3 条件边返回值写错

原因：route_after_security 返回的字符串必须与节点名完全一致。

解决：返回值 "human_review" 对应 workflow.add_node("human_review", ...)。

---

十一、学习总结

11.1 我的理解（学习检验）

route_after_security：存在严重安全问题 → human_review，否则 → style_review。

interrupt 暂停流程，展示 payload，用 Command(resume=...) 恢复；必须配合 Checkpointer。

多 interrupt 按出现顺序匹配 resume 值，且仅作用于当前 task。

在 scanners.py 添加安全规则，critical/error 级别触发人工审核。

以上理解正确，Phase 3 可以毕业。

11.2 通关清单

• 能画出工作流 DAG 图
• 理解 route_after_security / route_after_human
• 理解 interrupt + Checkpointer 机制
• 能看懂 Trace 里 [路由] 日志
• 能在 scanners.py 添加安全规则

11.3 下一步

• Phase 4：多 Agent 协作（Researcher / Writer / Reviewer / Supervisor）
• Phase 5：可观测性与评测（Langfuse / 回归测试集）
• Phase 6：生产化部署（FastAPI + Docker）

---

十二、参考资料

• LangGraph 官方文档
• LangGraph interrupt 概念
• 本项目源码：agent-workflow

---

系列文章导航

• Phase 1：工具 Agent（LangGraph + Function Calling）
• Phase 2：RAG 文档问答（LangGraph + Chroma + Embedding）
• Phase 2.5：工具 + RAG 合体
• Phase 3：复杂工作流（本文）
• Phase 4+：多 Agent 协作、生产化部署 —— 规划中

---

如果这篇文章对你有帮助，欢迎点赞收藏。有问题欢迎在评论区交流。