AI Agent全栈架构

主题

Human-in-the-Loop ​

要点 ​

  • 上一篇我们已经把 interrupt() 跑通了，知道图可以在节点里暂停，等外部输入后再继续
  • 审批工作流和前面那些纯自动图不一样
  • 审批节点不只是弹确认框，还要携带草稿、阶段和审核结果
  • Human-in-the-Loop 的核心不是 UI，而是把人工决策写回可恢复的图状态
  • 两级内容审核可以用生成、编辑审核、法务审核、修改、发布几个节点表达
  • 外部系统通常按「启动工作流 → getState 查看暂停信息 → Command resume 传回审核结果」驱动

内容 ​

1. 审批流程通常不是停一次就结束 ​

上一篇我们已经把 interrupt() 跑通了，知道图可以在节点里暂停，等外部输入后再继续。

但真实业务里的 Human-in-the-Loop，通常不只是「停一下，点个确认」这么简单。

更常见的是这样一条链：先生成一版内容，交给编辑审核；编辑通过以后，再交给法务审核；任意一关打回，都回到修改节点重新出一版；整个过程还得留痕，后面能查是谁在什么时候给了什么意见。

如果只会把 interrupt() 塞进一个节点里，你能做出「人工确认」。

但如果要做「审批工作流」，还得把这些暂停点和整张图的状态流转连起来。

这一篇就把这件事完整走一遍。

结合官方 human-in-the-loop 说明，可以先把它拆成三层：

  1. 图内：用 interrupt() 暂停，把草稿、阶段、版本等信息交给外部。
  2. 图外：后台、前端或运营系统读取 getState()，把待审核任务展示给人。
  3. 恢复：人工提交结果后，用 Command({ resume }) 回到同一个 thread，让图继续执行。

所以 Human-in-the-Loop 的重点不是「有没有弹窗」，而是人工决定能不能进入图的状态时间线。只有审核结果、审核人、审核意见都被写回 state，后面才谈得上恢复、追踪和复盘。

2. 先把审批链路拆出来 ​

先把流程拆开看，会更容易写。

这一篇用一个两级审批的例子来讲：generateDraft 先生成文案，editorReview 负责编辑审核，legalReview 负责法务审核，全部通过以后才进入 publish，只要有一关打回，就先进入 reviseDraft，再重新生成。

这条链里更需要关注的不是节点数量，而是职责边界。生成节点只管产出草稿，审核节点只管暂停、接收人工决定、更新审批状态，修改节点只管把流程切回待修改，发布节点只负责把最后结果落成已发布。这样后面你要把「编辑」换成「运营」，或者再加一个「品牌审核」，整张图都不用重写。

3. 状态里要放什么 ​

审批工作流和前面那些纯自动图不一样。状态里至少要放三类东西：当前草稿本身、当前走到哪一个审批阶段、还有审批记录。

前两个字段直接覆盖就够了，审批记录不行。因为每次审核都会往里追加一条记录，最后应该保留完整历史。

typescript
// approval-state.ts
import { StateSchema, ReducedValue } from '@langchain/langgraph'

import { z } from 'zod'

const appendLogs = (
  current: Array<{
    stage: string

    reviewer: string

    approved: boolean

    comment: string
  }>,

  update: Array<{
    stage: string

    reviewer: string

    approved: boolean

    comment: string
  }>,
) => {
  return [...current, ...update]
}

const State = new StateSchema({
  topic: z.string().default(''),

  draft: z.string().default(''),

  feedback: z.string().default(''),

  version: z.number().default(0),

  status: z.string().default('drafting'),

  reviewLogs: new ReducedValue(
    z
      .array(
        z.object({
          stage: z.string(),

          reviewer: z.string(),

          approved: z.boolean(),

          comment: z.string(),
        }),
      )
      .default([]),

    { reducer: appendLogs },
  ),
})

这一段里最值得注意的是 reviewLogs。

它不是普通数组，而是用 ReducedValue 明确声明成「追加日志」。这样编辑审核写一条、法务审核再写一条时，结果会自然接到后面。

4. 审核节点要做的事 ​

审批节点不是只负责弹一个确认框。它还得把当前草稿和阶段信息带给调用方，接收调用方传回来的审核结果，再根据审核结果决定下一步去哪。

也正因为这样，这类节点通常都会把 interrupt() 和 Command 放在一起用。

typescript
// review-node.ts
import type { GraphNode } from '@langchain/langgraph'

import { Command, interrupt } from '@langchain/langgraph'

const editorReview: GraphNode<typeof State> = (state) => {
  // 先暂停，把当前草稿和说明带出去

  // 这里传给 interrupt 的内容最好保持可序列化，外部系统才容易接住它

  const decision = interrupt({
    stage: 'editor',

    title: '请进行编辑审核',

    draft: state.draft,

    version: state.version,
  })

  // 恢复后，decision 就是外部传回来的审核结果

  if (decision.approved) {
    return new Command({
      update: {
        status: 'editor-approved',

        reviewLogs: [
          {
            stage: 'editor',

            reviewer: decision.reviewer,

            approved: true,

            comment: decision.comment ?? '',
          },
        ],
      },

      goto: 'legalReview',
    })
  }

  return new Command({
    update: {
      status: 'editor-rejected',

      feedback: decision.comment ?? '请继续修改',

      reviewLogs: [
        {
          stage: 'editor',

          reviewer: decision.reviewer,

          approved: false,

          comment: decision.comment ?? '',
        },
      ],
    },

    goto: 'reviseDraft',
  })
}

这里最核心的点是：

审核节点自己就知道下一步该去哪，所以不需要再额外挂一条条件边。通过就往下走，打回就回修改节点，这个决定直接留在节点内部就够了。

5. 把整条审批工作流接起来 ​

下面把完整流程连起来。

typescript
// approval-workflow.ts
import {
  StateGraph,
  StateSchema,
  ReducedValue,
  Command,
  interrupt,
  MemorySaver,
  START,
  END,
} from '@langchain/langgraph'

import type { GraphNode } from '@langchain/langgraph'

import { z } from 'zod'

const appendLogs = (
  current: Array<{
    stage: string

    reviewer: string

    approved: boolean

    comment: string
  }>,

  update: Array<{
    stage: string

    reviewer: string

    approved: boolean

    comment: string
  }>,
) => {
  return [...current, ...update]
}

const State = new StateSchema({
  topic: z.string().default(''),

  draft: z.string().default(''),

  feedback: z.string().default(''),

  version: z.number().default(0),

  status: z.string().default('drafting'),

  reviewLogs: new ReducedValue(
    z
      .array(
        z.object({
          stage: z.string(),

          reviewer: z.string(),

          approved: z.boolean(),

          comment: z.string(),
        }),
      )
      .default([]),

    { reducer: appendLogs },
  ),
})

// 生成节点：第一次生成，或者根据 feedback 重新出一版

const generateDraft: GraphNode<typeof State> = (state) => {
  const nextVersion = state.version + 1

  if (!state.feedback) {
    return {
      draft: `【${state.topic}】LangGraph 让复杂 Agent 的流程控制、状态管理和持久化都回到代码里。`,

      feedback: '',

      version: nextVersion,

      status: 'waiting-editor-review',
    }
  }

  return {
    draft: `【${state.topic}】LangGraph 让复杂 Agent 的流程控制、状态管理和持久化都回到代码里。它适合那些需要分支、暂停、恢复和审批的工作流。（本版根据反馈修改：${state.feedback}）`,

    feedback: '',

    version: nextVersion,

    status: 'waiting-editor-review',
  }
}

// 修改节点：这里只做状态切换，改稿还是回 generateDraft

const reviseDraft: GraphNode<typeof State> = () => {
  return {
    status: 'revising',
  }
}

// 编辑审核：通过后去法务，打回后去修改

const editorReview: GraphNode<typeof State> = (state) => {
  const decision = interrupt({
    stage: 'editor',

    title: '请进行编辑审核',

    draft: state.draft,

    version: state.version,

    instruction:
      '回复 { approved: true, reviewer, comment } 或 { approved: false, reviewer, comment }',
  })

  if (decision.approved) {
    return new Command({
      update: {
        status: 'waiting-legal-review',

        reviewLogs: [
          {
            stage: 'editor',

            reviewer: decision.reviewer,

            approved: true,

            comment: decision.comment ?? '',
          },
        ],
      },

      goto: 'legalReview',
    })
  }

  return new Command({
    update: {
      status: 'editor-rejected',

      feedback: decision.comment ?? '请继续修改',

      reviewLogs: [
        {
          stage: 'editor',

          reviewer: decision.reviewer,

          approved: false,

          comment: decision.comment ?? '',
        },
      ],
    },

    goto: 'reviseDraft',
  })
}

// 法务审核：通过后发布，打回后也回修改

const legalReview: GraphNode<typeof State> = (state) => {
  const decision = interrupt({
    stage: 'legal',

    title: '请进行法务审核',

    draft: state.draft,

    version: state.version,

    instruction:
      '回复 { approved: true, reviewer, comment } 或 { approved: false, reviewer, comment }',
  })

  if (decision.approved) {
    return new Command({
      update: {
        status: 'approved',

        reviewLogs: [
          {
            stage: 'legal',

            reviewer: decision.reviewer,

            approved: true,

            comment: decision.comment ?? '',
          },
        ],
      },

      goto: 'publish',
    })
  }

  return new Command({
    update: {
      status: 'legal-rejected',

      feedback: decision.comment ?? '请继续修改',

      reviewLogs: [
        {
          stage: 'legal',

          reviewer: decision.reviewer,

          approved: false,

          comment: decision.comment ?? '',
        },
      ],
    },

    goto: 'reviseDraft',
  })
}

const publish: GraphNode<typeof State> = (state) => {
  return {
    status: 'published',

    draft: `[已发布 V${state.version}] ${state.draft}`,
  }
}

const graph = new StateGraph(State)

  .addNode('generateDraft', generateDraft)

  .addNode('reviseDraft', reviseDraft)

  .addNode('editorReview', editorReview, {
    ends: ['legalReview', 'reviseDraft'],
  })

  .addNode('legalReview', legalReview, { ends: ['publish', 'reviseDraft'] })

  .addNode('publish', publish)

  .addEdge(START, 'generateDraft')

  .addEdge('generateDraft', 'editorReview')

  .addEdge('reviseDraft', 'generateDraft')

  .addEdge('publish', END)

  .compile({ checkpointer: new MemorySaver() })

这张图里最值得看的，是两个审核节点和那条回退边。

editorReview 和 legalReview 都在节点内部完成了暂停、接收人工结果、决定下一步路由这三件事。reviseDraft -> generateDraft 这条普通边则把「打回以后重走一版」这件事接了起来，所以流程不会停在打回那里，而是会重新进入下一轮审批。

6. 外部怎么驱动这条图 ​

图写完以后，外部驱动的节奏其实很固定：先启动工作流，让它跑到第一个审批点；再通过 getState() 看当前停在哪个节点、带出来了什么审核信息；最后用 Command({ resume }) 把人工审核结果传回去。

typescript
// approval-run.ts
// 整条审批链都要沿用同一个 thread_id，这样恢复时才能接回原来的流程

const config = { configurable: { thread_id: 'approval-001' } }

// ① 启动：先生成第一版，然后暂停在编辑审核

const initial = await graph.invoke({ topic: 'LangGraph 审批流入门' }, config)

let snapshot = await graph.getState(config)

console.log(snapshot.next)

// → ['editorReview']

// 第一次 invoke 返回的 __interrupt__ 就是外部界面最适合直接读取的审批信息

console.log(initial.__interrupt__)

// → {

//     stage: 'editor',

//     title: '请进行编辑审核',

//     draft: '【LangGraph 审批流入门】...',

//     version: 1,

//   }

// ② 编辑通过，继续往法务走

await graph.invoke(
  new Command({
    resume: {
      approved: true,

      reviewer: '编辑 Alice',

      comment: '内容没问题，可以继续',
    },
  }),

  config,
)

snapshot = await graph.getState(config)

console.log(snapshot.next)

// → ['legalReview']

// ③ 法务通过，图走到发布节点并结束

const result = await graph.invoke(
  new Command({
    resume: {
      approved: true,

      reviewer: '法务 Bob',

      comment: '可以发布',
    },
  }),

  config,
)

console.log(result.status)

// → published

console.log(result.reviewLogs)

// → [

//   { stage: 'editor', reviewer: '编辑 Alice', approved: true, comment: '内容没问题，可以继续' },

//   { stage: 'legal', reviewer: '法务 Bob', approved: true, comment: '可以发布' },

// ]

如果其中任意一关打回，恢复时把 approved 设成 false 就行。

图会自己回到 reviseDraft -> generateDraft 这条链，然后再重新停到下一轮审批节点。

7. 这种工作流里最容易踩的坑 ​

最容易出问题的，往往不是 interrupt() 本身，而是审批状态怎么设计。

如果把所有审核信息都塞进一段文本里，短期看着省事，后面一旦要查「第几版是谁打回的」或者统计某一关的通过率，马上就会变得很难处理。审批记录最好像前面的 reviewLogs 一样，用结构化字段存。

另外一个常见问题是恢复时换了 thread_id。审批流暂停以后，外部系统再回来恢复时，必须把同一条线程重新接上。只要线程换了，恢复就不是沿着原来的审批流继续跑。

还有一个坑是把副作用放在 interrupt() 前面。比如节点一进来先发通知、写数据库、调第三方接口，然后才 interrupt()。恢复时节点会从头再跑一遍，这些动作就会重复发生。所以审批节点前半段尽量只做准备数据，副作用放到拿到审核结果以后。

8. 总结 ​

到了这里，interrupt() 就不再只是一个「暂停函数」了。它和 Command、Checkpointer 放在一起以后，已经能落成一条完整的人工审批工作流。

前面几篇讲的状态、路由、持久化和控制流，这一篇都放进了一条能工作的内容审核流程里。现在这条链已经不只是能暂停，还能打回、重走、留痕、再继续往下发布。

下一篇讲 子图。到那时，问题会从「怎么把审批流跑起来」变成「这条审批流能不能拆出来，变成别的图也能复用的一块模块」。

基于 MIT 协议开源

Copyright © 2026 AI Agent全栈架构