【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 1/2Phase 3结构循环 / 两节点流水线多节点 DAG分支tools_condition自定义路由函数人工介入无interrupt 暂停状态messages结构化 ReviewState一句话从「对话循环」进化到「业务流程编排」。二、Phase 3 项目代码审查 Agent2.1 工作流全景图START ↓ parse_diff解析 diff统计变更文件/新增行 ↓ security_scan静态安全规则扫描 ↓ route_after_security条件分支 ├─ 有严重问题 → human_reviewinterrupt 人工确认 │ ↓ │ route_after_human │ ├─ approve → style_review │ └─ reject → reject_end → END └─ 无严重问题 → style_review风格扫描 ↓ generate_reportLLM 生成 Markdown 报告 ↓ END2.2 与 Phase 1/2 的本质区别Phase 1Phase 2Phase 3拓扑环循环链2 节点DAG多节点 分支决策LLM 决定调哪个工具固定先检索后生成代码规则 人工决策暂停无无interrupt 等人输入LLM 作用全程决策最后生成只在 generate_report 节点三、核心机制一条件分支 route_after_security3.1 路由函数安全扫描完成后用条件边决定下一步走哪条路径defroute_after_security(state:ReviewState)-Literal[human_review,style_review]:ifstate.get(has_critical):logger.info([路由] 存在严重安全问题 → human_review)returnhuman_reviewlogger.info([路由] 无严重问题 → style_review)returnstyle_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):returnstyle_review# 人工批准继续审查returnreject_end# 人工拒绝终止流程四、核心机制二interrupt 人机协同4.1 interrupt 是什么interrupt()可以暂停流程图执行并向客户端展示 payload 数据——可以是上下文信息也可以是请求恢复执行所需的输入内容。trace_node(human_review)defhuman_review(state:ReviewState)-ReviewState:decisioninterrupt({action:approve_critical_findings,message:发现严重安全问题是否继续生成审查报告,issues:state.get(security_issues,[]),hint:回复 approve 继续reject 终止,})approvedstr(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__):decisioninput(人工决策).strip()resultapp.invoke(Command(resumedecision),configconfig)4.3 多个 interrupt 的 resume 匹配若一个节点中调用多个interrupt()LangGraph 会根据中断在节点内的出现顺序将 resume 值与中断一一匹配。该 resume 值列表仅适用于当前这次 task不会在不同 thread 之间共享。本项目human_review只有一个 interrupt所以Command(resumeapprove)直接对应那一次暂停。4.4 必须开启 Checkpointer使用 interrupt 前必须开启检查点器——该特性依赖持久化存储图状态才能实现暂停与恢复memoryMemorySaver()appworkflow.compile(checkpointermemory)config{configurable:{thread_id:review-1}}没有 Checkpointer工作流无法在中断点保存状态resume 会失败。五、核心机制三结构化状态 ReviewStatePhase 1 的状态只有messagesPhase 3 扩展为结构化字段classReviewState(TypedDict,totalFalse):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.py6.1 安全规则表Phase 3 的安全扫描不依赖 LLM而是用正则规则扫描 diff 新增行SECURITY_RULES[(r\beval\s*\(,critical,使用 eval() 存在代码注入风险),(rpassword\s*\s*[\],critical,硬编码密码),(rsk-[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 如何添加自己的规则三步在SECURITY_RULES加一行(正则, 级别, 描述)确认级别是critical或error会触发人工审核用--sample risky或自定义 diff 验证七、各节点职责节点类型职责parse_diff处理解析 diff统计文件和行数security_scan扫描正则匹配安全问题route_after_security路由严重 → 人工 / 否则 → 风格human_reviewinterrupt暂停等人 approve/rejectroute_after_human路由批准 → 继续 / 拒绝 → 终止style_review扫描print/TODO/行过长等generate_reportLLM汇总扫描结果生成 Markdown 报告reject_end终止输出拒绝信息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.txt8.2 测试用例命令预期路径考察点--sample riskysecurity →human_review→ style → reportinterrupt 条件分支--sample cleansecurity → 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 1Phase 3图结构环DAG分支逻辑LLM 决定代码规则 人工状态messages结构化 dict暂停恢复无interrupt CheckpointerLLM 参与全程仅报告节点十、踩坑记录10.1 interrupt 不生效原因没开 Checkpointer或 thread_id 不一致。解决workflow.compile(checkpointerMemorySaver())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 / SupervisorPhase 5可观测性与评测Langfuse / 回归测试集Phase 6生产化部署FastAPI Docker十二、参考资料LangGraph 官方文档LangGraph interrupt 概念本项目源码agent-workflow系列文章导航Phase 1工具 AgentLangGraph Function CallingPhase 2RAG 文档问答LangGraph Chroma EmbeddingPhase 2.5工具 RAG 合体Phase 3复杂工作流本文Phase 4多 Agent 协作、生产化部署 —— 规划中如果这篇文章对你有帮助欢迎点赞收藏。有问题欢迎在评论区交流。