【Agent 实战】Phase 3：LangGraph 复杂工作流（代码审查 + 条件分支 + 人机确认 interrupt）

摘要：本文是 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 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 的本质区别

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

3.1 路由函数

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

defroute_after_security(state:ReviewState)->Literal["human_review","style_review"]:ifstate.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

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

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

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

4.1 interrupt 是什么？

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

@trace_node("human_review")defhuman_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=...)恢复：

whileresult.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 扩展为结构化字段：

classReviewState(TypedDict,total=False):diff_text:str# 原始 difffiles_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 严重级别与路由

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

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

6.3 如何添加自己的规则

三步：

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

七、各节点职责

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

八、环境准备与运行

8.1 快速开始

cd agent-workflowcopy..\FirstAgent\.env.env pip install-r requirements.txt# 含安全问题，触发 interruptpython agent.py--sample risky# 干净 diff，跳过人工节点python agent.py--sample clean# 自定义 diff 文件python agent.py--file your_diff.txt

8.2 测试用例

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 「业务流程编排 + 人工卡点」

十、踩坑记录

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 协作、生产化部署 —— 规划中

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