LangGraph 怎么测：三层，和一条容易踩的污染线

为什么图不好测

普通的函数测试很干净：给输入，断言输出。

图不一样，它有三个麻烦：

① 状态在节点之间流动，一个节点错了，错误会传给下游，最后报错的地方不是出错的地方
② 有 checkpointer，跑过一次就有历史，第二次跑可能接着上次继续
③ 一次全链路跑得慢，改一个节点要等全部跑完

前两个是「测不准」，第三个是「测得慢」。对应三种测法。

一、整图测试：从头到尾

最简单的一种，也最像端到端测试：

const out = await graph.invoke({ question: '订单 A1 能退吗' })
expect(out.answer).toContain('可以退')

它验证的是「节点串起来对不对」——边有没有连错、条件边分叉有没有分对、状态字段有没有漏传。

但它有一个前提：每次必须从零开始。

这里有个坑，和「共享实例」无关

流传的说法是「测试用例之间不能共享 graph 实例」。我实测了三种情况，结论要修正一下：

// 场景 1：同一个 graph 实例 + 同一个 thread_id，连跑两次
第一次 n=2，第二次 n=4        ← 第二次接着上次的 n=2 继续了

// 场景 2：同一个 graph 实例 + 不同 thread_id
case-A n=2，case-B n=2        ← 各自从 0 开始，没污染

// 场景 3：每个用例独立 graph 实例
第一次 n=2，第二次 n=2        ← 天然隔离

真正的污染源是 thread_id，不是 graph 实例。

同一个 thread_id 就是同一个会话，第二次 invoke 会基于上次的最终状态继续——于是你断言 n === 2 失败，而失败的原因是上一个用例留下的东西。

所以两条都成立，但优先级不同：

务实的选择： 共享实例 + 用例名当 thread_id（thread_id = testCase.id）。这样既快又干净。

但如果你用的是 MemorySaver，跑几百个用例之后所有历史都堆在内存里。长测试套件还是老老实实每个用例重建一个 checkpointer——用 new MemorySaver() 建图的开销很小。

怎么保证「每次都从头」

const makeGraph = () => buildGraph().compile({ checkpointer: new MemorySaver() })

test('退票流程', async () => {
  const graph = makeGraph()               // 每个用例一份
  const cfg = { configurable: { thread_id: `test-${Date.now()}-${Math.random()}` } }
  const out = await graph.invoke({ orderId: 'A1' }, cfg)
  expect(out.answer).toContain('可以退')
})

thread_id 带随机后缀是双保险：即使哪天忘了重建 checkpointer，也不会串。

二、单节点测试：当成纯函数

一个节点长这样：

const judge = (state) => ({ enough: state.rows.length >= 1 })

输入是 state，输出是「部分更新」。 不需要建图，不需要 checkpointer，直接调：

test('judge：有结果就够', () => {
  expect(judge({ rows: [{ id: 'A1' }], keywords: ['a', 'b'] })).toEqual({ enough: true })
})

test('judge：条件不全就不够', () => {
  expect(judge({ rows: [{ id: 'A1' }], keywords: ['a'] })).toEqual({ enough: false })
})

但它不是纯函数，这点要说清

真实节点几乎都会调外部服务：

const query = async (state) => {
  const rows = await db.query(state.keywords)      // 副作用
  return { rows }
}

它不是纯函数，但应该按纯函数的契约去测：

给定 state → 返回什么更新
其他一切（数据库、HTTP、模型）都 mock 掉

test('query：把关键词转成查询', async () => {
  const db = { query: vi.fn().mockResolvedValue([{ id: 'A1', amount: 1200 }]) }
  const out = await makeQuery(db)({ keywords: ['amount>1000'] })

  expect(db.query).toHaveBeenCalledWith(['amount>1000'])
  expect(out).toEqual({ rows: [{ id: 'A1', amount: 1200 }] })
})

三个原则：

① 只测契约，不测实现。 断言「输出了什么」，不断言「内部调了几次中间函数」。

② 边界比正常路径更重要。 空输入、超长输入、字段缺失、外部服务抛错——图里 90% 的 bug 在这些地方。

③ 别在节点测试里建图。 建图就引入了边界和状态，那就不是单元测试了。

复杂节点怎么拆

一个节点如果超过 50 行、或者里面有三层 if，它测起来会很难。这时候问题不在测试，在节点本身。

先把它拆成几个内部函数，各自单测；节点只负责编排。 节点本身的测试就只剩「调用顺序对不对」。

三、部分执行测试：只测中间一段

改了第 4 个节点，跑全链路要 3 分钟 —— 一天改二十次就是一小时。

LangGraph 提供了 updateState，可以直接把状态注入到某个位置，从那里开始跑：

// 5 节点：a → b → c → d → e
// 假装 a、b 已经跑完了，只测 c 之后的部分
await graph.updateState(
  { configurable: { thread_id: 'partial-1' } },
  { log: ['a', 'b'] },   // 注入的状态
  'b'                     // 当作哪个节点的输出
)
const out = await graph.invoke(null, { configurable: { thread_id: 'partial-1' } })

第三个参数 asNode 是关键：它告诉框架「这份状态是 b 的输出」，于是下一步就是 b 的下游 c。

实测：

=== 对照组：完整跑一遍 ===
执行次数: {"a":1,"b":1,"c":1,"d":1,"e":1}  状态: ["a","b","c","d","e"]

=== 部分执行：注入到 b 之后 ===
注入后的 next: ["c"]
执行次数: {"c":1,"d":1,"e":1}     ← a、b 没被重跑
最终状态: ["a","b","c","d","e"]

next 直接变成了 ["c"]，a 和 b 一次都没执行。 换个起点同理（注入到 d 之后，只跑 e）。

什么时候用

注意 invoke 的第一个参数要传 null，不是 {}。传 {} 相当于又给了一次输入，可能覆盖掉刚注入的状态。

一个副产品：用 next 断言分支

条件边是最容易错的地方，但它的输出不是「数据」而是「走向」。这时候直接断言 next：

await graph.updateState(cfg, { rows: [], keywords: ['a', 'b'] }, 'query')
expect((await graph.getState(cfg)).next).toEqual(['parse'])   // 不够，回去补条件

await graph.updateState(cfg2, { rows: [{ id: 'A1' }], keywords: ['a', 'b'] }, 'query')
expect((await graph2.getState(cfg2)).next).toEqual(['approve'])  // 够了，走审批

比跑完整条链路再猜「它到底走了哪条路」清楚得多。

四、一个可以照抄的骨架

describe('退票图', () => {
  // ① 节点级：纯函数
  describe('judge 节点', () => {
    it('条件齐全且有结果 → 够了', () => {
      expect(judge({ rows: [1], keywords: ['a', 'b'] })).toEqual({ enough: true })
    })
    it('缺条件 → 不够', () => {
      expect(judge({ rows: [1], keywords: ['a'] })).toEqual({ enough: false })
    })
  })

  // ② 分支级：只注入状态，断言走向
  describe('judge 之后的分支', () => {
    it('够了走 approve', async () => {
      const { graph, cfg } = fresh()
      await graph.updateState(cfg, { rows: [1], keywords: ['a', 'b'] }, 'query')
      expect((await graph.getState(cfg)).next).toEqual(['approve'])
    })
  })

  // ③ 整图级：端到端，独立实例
  describe('完整流程', () => {
    it('查到 3 笔 → 人审 → 执行', async () => {
      const { graph, cfg } = fresh()
      const out = await graph.invoke({ question: '近三个月超 1000 的订单' }, cfg)
      expect(out.approved).toBe(true)
    })
  })
})

function fresh() {
  return {
    graph: buildGraph().compile({ checkpointer: new MemorySaver() }),
    cfg: { configurable: { thread_id: `t-${crypto.randomUUID()}` } },
  }
}

比例建议：节点测试占大头，整图测试只留几条主干。 因为节点测试毫秒级、失败信息精确；整图测试慢，失败时只知道「这条链走不通」。

五、三条可以直接记住的

1. 污染源是 thread_id，不是 graph 实例。 共享实例 + 唯一 thread 是安全的；共享 thread 必炸。
2. 节点不是纯函数，但要按纯函数的契约测。 外部依赖全 mock，只断言「输入 → 输出更新」。
3. updateState(state, values, asNode) 是测中间段的关键。 第三个参数决定下一步是谁，配合断言 next 可以精确测分支。