保存してあるから、その場所に戻れる

目標

チェックポインターを付けたグラフで、何がどこに残るのかを、自分で数えます。同じthread_idがつながることを確認し、チェックポイントをリストとして読み、過去の座標に巻き戻し、値を直してブランチを切り、状態をシリアライズしてバイトを数えます。最後に、帳簿を新しく作ると何が消えるかを再現します。

なぜ重要なのか

チェックポインターは、ステップごとにその時点の状態をまるごと残します。この1文が、よいことと悪いことを同時に説明します。よい面は、続きから動かせることです。8番目のステップで失敗しても、最初からやり直す必要がなく、「3番目で別の資料を使っていたらどうなっただろう」という質問に、同じ条件で答えられます。過去のスナップショットのconfigでinvoke(None, ...)すれば、その場所でもう一度動き、update_stateで値を直して入れれば、そこからブランチができます。このとき、元のブランチは消えませんが、スレッドの「現在」は新しいブランチに移るので、元の結果をもう一度見るには、その時点のconfigを手に持っておく必要があります。悪い面は、保存されるのが状態の全体だということです。状態に大きな値を1つ入れると、その値がチェックポイントの数だけ繰り返し保存されます。このラボでは、それを時間ではなくバイトで測ります。時間はマシンと負荷によって変わりますが、サイズは同じ状態ならいつも同じだからです。採点ツールは、書かれた説明を信用しません。書かれたモジュールを実際に読み込んでグラフを動かし、チェックポイントを自分で数えて、返された値と突き合わせます。トピックと入れる値のサイズは、実行のたびに変わります。

ステップ

/root/work/agckpt/ckpt.pyにANGLES・State・3つのノード(plan・write・review)・build_graph(checkpointer=None)・new_saver()・cfg(thread_id)を作成してください。同じthread_idはつながり、違うものは別々に残る必要があります。
history(app, config)・history_len(app, config)・pending(app, config)を追加して、チェックポイントを古いものから読んで数えるようにしてください。
snapshot_before(app, config, node)を追加して、そのノードをまだ動いていない最も新しいチェックポイントを探すようにしてください。
replay(app, config, node)を追加して、その座標からinvoke(None, snapshot.config)でもう一度動くようにしてください。
fork(app, config, node, values)を追加して、update_stateでブランチを切り、元のブランチの終端も一緒に返すようにしてください。
state_bytes(values)・history_bytes(app, config)・weigh(topic, payload)と状態キーbulkを追加して、チェックポイントのサイズをバイトで測るようにしてください。
lost_demo(topic)を追加して、帳簿を新しく作ると、同じthread_idでも何も残らないことを再現してください。
測った値を記録してください(保存先: /root/work/agckpt/ckpt_report.json、/root/work/agckpt/ckpt_report.md)。

参考

実行の契約: 採点ツールは、/root/work/agckpt/ckpt.pyをPythonモジュールとして読み込んで、ANGLES・State・build_graph・new_saver・cfg・history・history_len・pending・snapshot_before・replay・fork・state_bytes・history_bytes・weigh・lost_demoを直接使います。スクリプトとして実行しないので、if __name__ == "__main__"はなくてもかまいません。
ANGLES = {"기본": 1, "요약": 2, "비교": 3}です(韓国語で、順にキーは「基本」「要約」「比較」を意味します)。ブランチ名ごとに、下書きを何回繰り返すかを入れます。
ノードの動作は、正確には次のとおりです。planは{"stage": "plan", "angle": 지금 angle 또는 "기본", "notes": ["plan:<topic>"]}(コードの中の韓国語は、1つ目の部分が「現在のangle、または」、2つ目の語が「基本」を意味します)を、writeは"<topic>/<angle> "をANGLES[angle]回連結したあとで両端の空白を取り除いたものをdraftに入れ、{"stage": "write", "notes": ["write:<angle>"]}を、reviewは{"stage": "review", "score": len(draft), "notes": ["review:<score>"]}を返します。
notesにはoperator.addリデューサーを付けます。残りのキーは上書きです。
build_graph(checkpointer=None)は、graph.compile(checkpointer=checkpointer)で終わります。チェックポインターを渡さなければ、何も保存されません。
cfg(thread_id)は、{"configurable": {"thread_id": thread_id}}を返します。
get_state_history(config)は、新しいものから返します。history()は、それを逆にして、古いものから返してください。pending()は、各スナップショットのnextの最初の要素を、空なら空の文字列を入れたリストです。
replay()は{"from": node, "result": 결과, "added": 늘어난 체크포인트 수}(プレースホルダーは結果と、増えたチェックポイントの数です)を返します。fork()は{"forked_config": ..., "result": ..., "original": 분기 전 끝의 값 딕셔너리, "total": 지금 체크포인트 수}(プレースホルダーは、フォーク前の終端の値のディクショナリと、現在のチェックポイント数です)を返します。
state_bytes(values)は、json.dumps(values, ensure_ascii=False, sort_keys=True, default=str)した文字列をUTF-8でエンコードした長さです。採点ツールが同じように計算して突き合わせるので、この3つの引数をそのまま使ってください。
weigh(topic, payload)は、{"checkpoints": ..., "thin_total": ..., "fat_total": ..., "payload_bytes": ..., "carrying": ...}を返します。carryingは、同じ場所同士で比べて、サイズの差がpayload_bytes以上のチェックポイントの個数です。
lost_demo(topic)は、{"kept": ..., "after_restart": ..., "values_after_restart": ...}を返します。スレッド名はrun-1を使ってください。
このPodにはインターネットがありません。pip installはできません。langgraph 0.2.60がすでに入っています(python3 -c "import langgraph")。
公式ドキュメント: Persistence・Use time-travel・Graph API overview
よくある間違い: compile()にチェックポインターを渡さずにget_stateを呼ぶこと(ValueError: No checkpointer set)、巻き戻すときに入力にNoneの代わりに状態を渡すこと(新しく始まります)、update_stateが返したconfigを捨てて元のconfigで続きを動かすこと、get_state_historyの順序を逆にしないことです。

同じスレッドはつながり、違うスレッドは別々に残る

/root/work/agckpt/ckpt.pyにANGLES・State・3つのノード(plan・write・review)・build_graph(checkpointer=None)・new_saver()・cfg(thread_id)を作成してください。チェックポインターを渡したグラフは、同じthread_idでもう一度動かすと前のnotesの上に積み上がり、違うthread_idは空の状態から始まる必要があります。

graph.compile(checkpointer=checkpointer)の1行が全部です。cfg(thread_id)は、{"configurable": {"thread_id": thread_id}}を返す小さな関数にしておくと、あとのステップが楽です。notesにoperator.addを付けて初めて、つながっていることが目に見えます。付けないと、保存はされますが値が上書きされて、つながったことがわかりません。

チェックポイントはステップごとに残る

history(app, config)・history_len(app, config)・pending(app, config)を追加してください。history()は、チェックポイントを古いものから並べ、pending()は、スナップショットごとに次に動くノード名を(すべて終わった場所は空の文字列)入れたリストを返します。

app.get_state_history(config)はジェネレーターで、新しいものから出てきます。list()で受け取ってreversed()してください。スナップショットのnextはタプルです。空なら、もう動くものがないという意味です。数えると、ノード数より2つ多いですが、入力を受け取った場所と、すべて終わった場所が1つずつ加わるからです。

巻き戻す座標を探す

snapshot_before(app, config, node)を追加してください。そのノードをまだ動いていない最も新しいチェックポイントを返し、そのような場所がなければNoneを返します。

探す条件は、スナップショットのnextが(node,)であることです。古いものから並べたリストを後ろから走査すると、「最も新しいもの」に先に出会います。返すのはスナップショットそのものです。あとのステップが使うのは、その中のconfigで、そこにはthread_idのほかにcheckpoint_idが入っています。

その場所からもう一度動く

replay(app, config, node)を追加してください。snapshot_beforeで座標を探して、app.invoke(None, snapshot.config)で続きを動かし、{"from": node, "result": 결과, "added": 늘어난 체크포인트 수}(プレースホルダーは結果と、増えたチェックポイントの数です)を返します。座標がなければValueErrorを上げてください。

入力をNoneにするのが核心です。状態を渡すと新しく始まり、Noneは「保存されたその時点から続きを動かす」という意味になります。addedは、巻き戻しの前後でhistory_lenを測って引いた値です。もう一度動いたノード数だけ増えます。

値を直して別のブランチへ行く

fork(app, config, node, values)を追加してください。フォーク前のスレッドの終端を先に押さえておき、app.update_state(snapshot.config, values)でブランチを作ってから、そのconfigで続きを動かします。{"forked_config": ..., "result": ..., "original": 분기 전 끝의 값 딕셔너리, "total": 지금 체크포인트 수}(プレースホルダーは、フォーク前の終端の値のディクショナリと、現在のチェックポイント数です)を返してください。

update_stateは、新しいcheckpoint_idが入ったconfigを返します。それを捨てて元のconfigで続きを動かすと、フォークになりません。そして、フォークしたあとは、thread_idだけで尋ねた「現在」が新しいブランチの終端を指すので、元の結果は、フォーク前に押さえておいたスナップショットのconfigで、別に読む必要があります。

サイズで測る

状態キーbulkとstate_bytes(values)・history_bytes(app, config)・weigh(topic, payload)を追加してください。state_bytesは、json.dumps(values, ensure_ascii=False, sort_keys=True, default=str)した文字列のUTF-8の長さです。weighは、同じグラフをbulkなしで1回、bulkにpayloadを入れて1回動かして、{"checkpoints": ..., "thin_total": ..., "fat_total": ..., "payload_bytes": ..., "carrying": ...}を返します。

carryingは、2つのリストを同じ場所同士で比べて、サイズの差がpayload_bytes以上のチェックポイントの個数です。その大きな値を実際に持っている場所の数、という意味です。2回動かすときは、帳簿を別に使ってください(new_saver()を2回)。時間を測らないでください。マシンと負荷によって変わります。バイト数は、同じ状態ならいつも同じです。

帳簿を新しく作ると消える

lost_demo(topic)を追加してください。新しい帳簿でグラフを作ってrun-1スレッドを1回動かし、チェックポイント数を測ったあと、別の新しい帳簿でグラフをもう1つ作って、同じthread_idで尋ねます。{"kept": ..., "after_restart": ..., "values_after_restart": ...}を返してください。

MemorySaverは、プロセスの中のメモリに入れます。新しいMemorySaverを作るのが、プロセスがもう一度立ち上がったことを真似る、最も正直な方法です。新しい帳簿に尋ねると、エラーは出ずに空の値が返ってきます。そのため、この問題は静かです。ラボやテストには向いていますが、本番には合わない理由が、これです。

測った値で記録する

/root/work/agckpt/ckpt_report.jsonにtopic・checkpoints_per_run・pending・base_score・fork_angle・fork_score・checkpoints_after_fork・payload_bytes・carrying・survives_restartを、/root/work/agckpt/ckpt_report.mdに## 무엇이 저장되는가 ## 되감기와 분기는 어떻게 다른가 ## 크기로 잰 것 ## 사라지는 것の4つの節を書いてください(4つの見出しは、順に韓国語で「何が保存されるのか」「巻き戻しとフォークはどう違うのか」「サイズで測ったもの」「消えるもの」を意味します)。

数値は手で書かず、自分のモジュールを実際に動かして得た値で埋めてください。topicは、自分が使ったトピックの文字列をそのまま、fork_angleは、기본(韓国語で「基本」を意味する語です)ではないブランチ名です。採点ツールが、その2つでスコアを再計算して突き合わせます。フォークはwriteの前で行ってください。payload_bytesは300以上である必要があり、そうして初めて差が目に見えます。survives_restartは真偽値です。