承認を飛ばしたグラフはエラーを出さない
一言でいうと
取り消せない作業の前で止まるには、中断点だけでは足りません。チェックポインターがないと、グラフは黙って止まり、その1件は、なかったことになります。
なぜ必要なのか
返金エージェントを立ち上げるときに、「10万ウォンを超えたら、人が見てから出す」を入れました。interrupt_before=["settle"]の1行でした。デモもうまくいきました。
1週間後、カスタマーセンターから連絡が来ました。高額の返金案件が、承認待ち一覧に表示されもせず、支払いもされていません。ログにエラーはありません。エージェントは成功として終わっていました。
原因は1行でした。compile(interrupt_before=["settle"])だけを渡して、checkpointerを付けなかったのです。このラボのイメージのlanggraph 0.2.60で実測すると、次のようになります。
체크포인터 없이 interrupt_before → 예외 없음. 결과는 {'amount': 100, 'trace': ['assess']}
settle 을 지나지 않았고, 이어서 돌릴 방법도 없다
このコードブロックの韓国語の出力は、チェックポインターなしで中断点を指定しても、例外は出ず、settleを通らずに終わり、続きから動かす方法もない、という意味です。
例外が出ていたら、その日のうちに見つかったはずです。黙って半分だけの状態を返したので、1週間かかりました。
どう動くのか
中断は、保存の上に築かれています。Persistenceのドキュメントが述べるとおり、チェックポインターはステップごとに状態を残します。中断とは、「次のステップを今は実行せず、残しておいた状態から、あとで続きを行う」という意味です。残しておく場所がなければ、続きを行うこともできません。
そのため、3つが一緒にある必要があります。
- チェックポインター:
compile(checkpointer=MemorySaver()) - thread_id:
{"configurable": {"thread_id": "건-1042"}}。どの案件の続きを行うかを指します(値の先頭の語は、韓国語で「案件」を意味します)。 - 中断点:
interrupt_before=["settle"]、またはノードの中のinterrupt(...)
止まったあとは、get_state(config)で状態を見ます。.nextが「次に動くノード」を持っていて、これが空でないということが、そのまま「まだ終わっていない」という意味です。
state = app.get_state(config)
state.next # ('settle',) — 여기서 기다리는 중
state.values # 그 시점의 상태 전체
続きを動かすときは、入力の場所にNoneを渡します。app.invoke(None, config)は、「新しい入力はない。保存された場所から続きを実行する」という意味です。
人が直した値は、リデューサーを通る
承認者が金額を減らしたり、拒否したりすることが、実際には最もよくあります。そのときに使うのがupdate_stateです。
app.update_state(config, {"amount": 50000, "decision": "reject"})
app.invoke(None, config)
ここで引っかかりやすい点が2つあります。
1つ目: update_stateが入れる値も、そのキーのリデューサーを通ります。上書きするキーは上書きしますが、連結するキー(traceのようなもの)に書くと、積み上がります。人が直した痕跡を残すためにtraceに1行入れるのは、そのため自然で、逆に「リストをまるごと入れ替えるため」に入れると、意図と違って増えます。
2つ目: 直した値が、ブランチをもう一度走らせることがあります。update_stateは、「最後に動いたノードがその値を書いた」という記録として残ります。そのため、そのノードに条件付きエッジが付いていれば、その条件が再評価されます。このラボで実測すると、次のように現れます。承認待ちの50万ウォンの案件の金額を、承認者が5万ウォンに減らすと、その案件は承認の経路を外れて、自動処理の経路に抜けます。needs_approvalがもう一度呼ばれて、今度は「承認は不要」になるからです。
これがバグなのか機能なのかは、業務が決めます。減らして少額になったので、そのまま出してよいなら機能で、「一度承認待ちに上がった案件は、人が最後まで見届ける必要がある」ならバグです。後者なら、判断に使う値と実行に使う値を分けなければなりません。元の要求金額を別に持ち歩いてブランチはそれで決め、実行だけを直した金額で行います。
ノードの外で止まることと、ノードの中で止まること
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()の前に置いた副作用(メール送信、外部呼び出し、カウンターの増加)は、2回起きます。これを知らずに、interrupt()の前で決済を呼ぶと、2回決済されます。規則は単純です。interrupt()の前には、もう一度行ってもよいことだけを置きます。取り消せないことは、interrupt()の後ろ、またはいっそ次のノードに置きます。
逆に、interrupt_beforeは、ノードに入る前に止まります。そのため、そのノードは1回だけ動きます。その代わり、人に見せる値を別に選ぶ場所がなく、状態をまるごと見せることになります。
interrupt_before |
ノードの中のinterrupt() |
|
|---|---|---|
| 止まる場所 | ノードの前 | ノードの中、呼んだ地点 |
| そのノードの実行回数 | 1 | 2(再開すると最初からもう一度) |
| 人に渡す値 | 状態全体 | interrupt(값)(プレースホルダーは値です)で選んだもの |
| 再開 | invoke(None, config) |
invoke(Command(resume=답), config)(プレースホルダーは答えです) |
現場での姿
1つ目: チェックポインターを忘れる。上の事故です。エラーが出ないので、長く生き残ります。承認経路を作ったら、「本当に止まるのか」をget_state().nextで確認するテストを、必ず置きます。
2つ目: すべてを承認させる。少額の単純な返金まで人が見ると、待ち行列が詰まり、詰まった待ち行列は、結局誰も見なくなります。基準をコードの1か所(needs_approval)に書き、その基準を変えることが、そのまま方針を変えることになるようにします。
3つ目: 拒否を表現する方法がない。承認だけがあって拒否がないと、承認者は「ただ押さない」ことで拒否します。すると、その案件は永遠に待ち行列に残ります。拒否も結果である必要があります。
4つ目: 承認の記録がない。誰が承認したかより先に、要求された値と実際に出た値が違うかを残す必要があります。承認者が金額を直したなら、その事実が記録にあって初めて、あとで説明できます。
実務で本当に大切なこと
- 中断は保存の上にあります。チェックポインターとthread_idなしで中断点だけを渡すと、黙って消えます。
interrupt()の前には、もう一度行ってもよいことだけを置きます。再開すると、そのノードは最初からもう一度動きます。- 拒否も結果です。承認・拒否・修正の3つがすべて最後まで流れる必要があります。
- 要求値と実行値を一緒に残します。承認記録の核心は、人の名前ではなく、変わった値です。
次のラボですること
/root/work/aghitl/approve.pyを、1ステップずつ育てます。まず、承認が必要な作業とそうでない作業を分ける基準とブランチを作り、チェックポインターをわざと外したグラフを動かして、黙って止まる様子を自分の目で見ます。次に、チェックポインターとthread_idを付けて本当に止め、続きを動かし、承認者が金額を直すか拒否したあとで続きを動かします。最後に、ノードの中で止まる方式を作って、そのノードに何回入ってくるかを自分で数え、要求値と実行値を一緒に残す記録を作ります。