TT Lab
开始
学习 学习路径 课程

AI 智能体 — 是图不是模型

漏掉审批的图不会报错

在 TT Lab 中继续学习

一句话总结

要在无法撤销的事情面前停下来,只有断点是不够的——没有 checkpointer,图会悄悄停住,那一条就这样成了没发生过的事。

为什么需要它

上线退款智能体时,加入了“超过十万韩元由人确认后再放行”。只是一行 interrupt_before=["settle"]。演示也很顺利。

一周后客服中心来了联系。大额退款的那一条既没有出现在待审批列表里,也没有付款。日志里没有错误。智能体是以成功结束的。

原因只有一行。只给了 compile(interrupt_before=["settle"]),没有加上 checkpointer。在本实验镜像的 langgraph 0.2.60 中亲自测量,会是这样。

체크포인터 없이 interrupt_before → 예외 없음. 결과는 {'amount': 100, 'trace': ['assess']}
                                   settle 을 지나지 않았고, 이어서 돌릴 방법도 없다

如果抛了异常,当天就会被发现。由于它悄悄返回了只有一半的状态,所以花了一周时间才发现。

工作原理

暂停是建立在保存之上的。正如 Persistence 文档所说,checkpointer 在每一步都留下状态。暂停的意思是“不在现在执行下一步,而是在留下的状态上,稍后接着做”。没有地方留存,就无法接着做。

所以三样东西必须同时具备。

暂停之后,用 get_state(config) 查看状态。.next 里装着“接下来要运行的节点”,它不为空就意味着“还没结束”。

state = app.get_state(config)
state.next            # ('settle',)  — 여기서 기다리는 중
state.values          # 그 시점의 상태 전체

接着运行时,在输入位置传入 None。app.invoke(None, config) 的意思是“没有新输入。从保存的地方接着做”。

人修改过的值也要经过 Reducer

最常见的情况,其实是审批人把金额砍低或者拒绝。这时用的是 update_state。

app.update_state(config, {"amount": 50000, "decision": "reject"})
app.invoke(None, config)

这里容易栽跟头的地方有两处。

第一,update_state 放入的值也要经过那个键的 Reducer。覆盖型的键会被覆盖,但写到接续拼接型的键(比如 trace)上就会累积。为了留下人修改的痕迹,往 trace 里放一行是很自然的,反过来如果想“把整个列表换掉”而放入,就会与本意不同地变多。

第二,修改后的值可能让分支重新走一遍。update_state 会作为“最后运行的节点写入了这个值”的记录保留下来。所以如果那个节点上挂着条件边,那个条件会被重新评估。在本实验中亲自测量,表现是这样——正在等待审批的 50 万韩元的一条,审批人把金额砍到 5 万韩元,那一条就会离开审批路径,落入自动处理路径。因为 needs_approval 被重新调用,现在变成了“不需要审批”。

这是 bug 还是功能,由业务决定。如果砍成小额后就这样放行也可以,那就是功能;如果是“一旦进入待审批的,必须由人看到最后”,那就是 bug。后者的话,必须把用于判断的值和用于执行的值分开——另外保存原始请求金额,分支按它来决定,只有执行使用修改后的金额。

在节点外停下与在节点内停下

interrupt() 是在节点内部停下。停下时可以一并传递要展示给人的值,恢复时人给出的答案会作为 interrupt() 的返回值传回来。

def confirm(state):
    answer = interrupt({"question": "이 환불을 승인합니까", "amount": state["amount"]})
    return {"decision": "approve" if answer == "yes" else "reject"}

app.invoke(Command(resume="yes"), config)

很方便,但有一个必须知道的性质。恢复后,那个节点会从头再运行一遍。亲自数一数就是这样。

첫 실행 후   노드에 들어온 횟수 1
재개 후      노드에 들어온 횟수 2

所以放在 interrupt() 之前的副作用(发邮件、外部调用、计数器自增)会发生两次。不知道这一点而把支付放在 interrupt() 之前,就会支付两次。规则很简单——interrupt() 之前只放重做也无妨的事。无法撤销的事放在 interrupt() 之后,或者干脆放到下一个节点。

相反,interrupt_before 是在进入节点之前停下。所以那个节点只运行一次。代价是没有单独挑选要展示给人的值的地方,会把状态整个展示出来。

interrupt_before 节点内的 interrupt()
停下的位置 节点之前 节点内部,调用的位置
该节点执行次数 1 2(恢复后从头再来)
交给人的值 整个状态 用 interrupt(값)(占位符为要展示的值)挑选的内容
恢复 invoke(None, config) invoke(Command(resume=답), config)(占位符为人给出的答案)

在现场相遇的样子

第一,漏掉 checkpointer。就是上面的事故。因为不报错,所以能活很久。做了审批路径,就一定要有一个测试,用 get_state().next 确认“是否真的停住了”。

第二,让所有事情都要审批。连小额的简单退款也让人看,队列就会堵,堵住的队列最后谁也不看。把标准写在代码的一处(needs_approval),让修改这个标准就等于修改策略。

第三,没有表达拒绝的办法。只有批准而没有拒绝,审批人就会用“干脆不点”来拒绝。那样那一条就永远留在队列里。拒绝也必须是一种结果。

第四,没有审批记录。比起谁批准了,更应先留下请求的值和实际执行的值是否不同。如果审批人改了金额,这一事实必须在记录里,以后才能解释。

实际工作中真正重要的事

下一项实验要做什么

一步步扩展 /root/work/aghitl/approve.py。先做出区分需要审批与不需要审批的标准和分支,运行故意漏掉 checkpointer 的版本,亲眼看到它悄悄停住。然后加上 checkpointer 和 thread_id,让它真正停下,再接着运行,并在审批人修改金额或拒绝之后接着运行。最后做出在节点内停下的方式,亲手数一数进入该节点几次,并做出把请求值和执行值一起留下的记录。