通った道が一つしか残らない

目標

エージェントの状態を、キーの一覧ではなく統合ルールとして設計します。リデューサーがないときに何が起きるかを自分で再現したあと、連結・重複除去・最大値・ウィンドウの切り詰めの4つのリデューサーを作り、入力と出力のスキーマを分けて、外に出るキーを絞ります。

なぜ重要なのか

LangGraphの状態は、キーごとにチャネルが1つずつある構造です。ノードは変更したいキーだけを入れたディクショナリを返し、グラフがチャネルごとに適用します。このとき、リデューサーを書いていないキーは上書きされます。3つのノードが順に書いたtraceが、最後の1つだけになるのは、そのためです。エラーが出ないので、「ログが残らない」ように見え、ログのコードを疑って時間を無駄にします。反対側の問題もあります。リデューサーを付けると、そのキーは際限なく増えますが、チェックポインターは状態をまるごと保存します。ノードが30回動けばチェックポイントも30個で、その1つ1つが、その時点の状態全体です。そのため、増えるキーには、上限をリデューサーの中に書いておきます。ノードごとに散らすと、1か所でも抜けると、黙って漏れ出します。最後に、状態には、外に見えてはいけないものが混ざります。出力スキーマで、外に出るキーを許可リストで決めれば、新しいキーが増えても、自然に防がれます。採点ツールは、書かれた説明を信用しません。書かれたモジュールを実際に読み込んで、リデューサーを毎回異なる値で直接叩いて試し、グラフを動かして出た状態を、採点ツールが別に計算した値と突き合わせます。ノード名と数字は、実行のたびに変わります。

ステップ

/root/work/agstate/state.pyにCATALOG・State・3つのノード(intake・lookup・finish)・build_graph()を作成してください。ノードは、変更するキーだけを返し、触れていないキーは、そのまま残る必要があります。
overwrite_demo(names)を追加して、リデューサーのないキーが上書きされることを再現させてください。名前ごとにノードを1つずつ作って一列につなぎ、ノードごとにseenに自分の名前を1つ書きます。
StateのtraceにAnnotated[list, operator.add]を付けて、3つのノードの足跡が順に積み上がるようにしてください。
リデューサーmerge_sources(old, new)を作って、sourcesキーに付けてください。同じ出典は1回だけ残り、順序は最初に見た順序を守ります。
used_callsにoperator.addを、peak_msに自分で作ったkeep_max(old, new)を付けてください。
RECENT_KEEP = 3とリデューサーkeep_recent(old, new)を作って、recentキーが後ろから3つだけ残るようにしてください。finishは1回で2つを書きます。
InputState・OutputStateを作り、build_public()がStateGraph(State, input=InputState, output=OutputState)でコンパイルしたグラフを返すようにしてください。結果にはanswerだけが出る必要があります。
状態の設計を記録してください(保存先: /root/work/agstate/state_report.json、/root/work/agstate/state_report.md)。

参考

実行の契約: 採点ツールは、/root/work/agstate/state.pyをPythonモジュールとして読み込んで、CATALOG・State・build_graph・overwrite_demo・merge_sources・keep_max・keep_recent・RECENT_KEEP・build_publicを直接使います。スクリプトとして実行しないので、if __name__ == "__main__"はなくてもかまいません。
build_graph()は、compile()されたグラフを返します。StateGraphそのものを返すと、.invoke()がありません。
ノードの順序はintake → lookup → finishです。traceに書く名前も、その3つと同じにしてください。
intakeは質問からCATALOGのキーを探してtargetに入れ、lookupはfoundに個数を(なければ-1)入れ、finishはanswerに"<이름> 재고는 <개수>개입니다"を入れます(この韓国語の文は「<名前>の在庫は<個数>個です」という意味です)。
リデューサーは、(옛값, 새값)(プレースホルダーは古い値と新しい値です)の2つの引数を受け取る普通の関数です。組み込み関数をそのまま渡してはいけません。Annotated[int, max]は、コンパイルのときにValueError: no signature found for builtin maxで死にます。
sourcesに、intakeは["질문"]、lookupは["질문", "재고목록"]、finishは["재고목록"]を書きます(韓国語で、1つ目の値は「質問」、2つ目の値は「在庫一覧」を意味します)。リデューサーが正しければ、結果は2つだけです。
used_callsは、ノードごとに1、peak_msは、intakeが4・lookupが17・finishが9を書きます。
recentに、intakeは["intake"]、lookupは["lookup"]、finishは["finish:작성", "finish:검토"]を書きます(コロンの後ろの2語は、韓国語で「作成」と「検討」を意味します)。
このPodにはインターネットがありません。pip installはできません。langgraph 0.2.60がすでに入っています(python3 -c "import langgraph")。
公式ドキュメント: Graph API overview・Use the graph API・Persistence
よくある間違い: ノードが状態全体を返すこと(変更するキーだけを返してください)、Annotatedを外してリデューサー関数だけを定義しておくこと、組み込みのmaxをリデューサーとして渡すこと、ウィンドウの切り詰めをリデューサーではなくノードに入れることです。

ノードは変更するキーだけを返す

/root/work/agstate/state.pyにCATALOG・State・3つのノード(intake・lookup・finish)・build_graph()を作成してください。build_graph()はcompile()されたグラフを返し、ノードが触れていないキーは、そのまま残る必要があります。

TypedDictで状態を定義し、StateGraph(State)にノードを追加してから、STARTからENDまでつなぎます。ノードは、状態全体ではなく変更したいキーだけを入れたディクショナリを返します。そうしないと、ほかのノードが入れておいた値が消えてしまいます。total=Falseを付ければ、キーがすべて埋まっていなくてもかまいません。

リデューサーがないと最後の1つだけが残る

overwrite_demo(names)を追加してください。namesの名前ごとにノードを1つずつ作って一列につなぎ、ノードごとに、リデューサーのないキーseenに自分の名前を1つ書きます。返す値は{"writes": [...], "state_after": [...]}です。

ノードをループで作るとき、ラムダが最後の名前だけを捕まえてしまう古典的な落とし穴に注意してください。デフォルト引数か、包む関数で、名前を束縛する必要があります。writesには、作成したノードの名前を順に、state_afterには、グラフを動かしたあとに実際に残ったseenの値を入れます。このステップは、わざと間違った設計を再現するものです。

足跡を積み上げるリデューサー

StateのtraceをAnnotated[list, operator.add]に変えて、3つのノードの足跡が["intake", "lookup", "finish"]の順に積み上がるようにしてください。

from typing import Annotatedとimport operatorが必要です。Annotatedの2番目の位置に置いた関数が、そのキーのリデューサーになります。ノード側のコードはそのままにしてください。変わるのは、状態の定義の1行だけです。それがこの設計の要点です。

同じ出典を2回書かない

リデューサーmerge_sources(old, new)を作って、sourcesキーに付けてください。同じ値は1回だけ残り、順序は最初に見た順序を守ります。3つのノードは、それぞれ["질문"]・["질문", "재고목록"]・["재고목록"]を書きます(韓国語で、1つ目の値は「質問」、2つ目の値は「在庫一覧」を意味します)。

リデューサーは、(옛값, 새값)(プレースホルダーは古い値と新しい値です)の2つの引数を受け取って、新しい値を返す普通の関数です。setで重複を除くと、順序が崩れるので、順序を守りながら絞り込む必要があります。古い値がまだない場合もあるので、old or []のように、空の値に耐えられるように書いてください。

足すものと、最大のものだけを残すもの

used_callsにoperator.addを、peak_msに自分で作ったkeep_max(old, new)を付けてください。ノードごとに、used_callsは1を、peak_msはintakeが4・lookupが17・finishが9を書きます。

数値にもリデューサーを付けられます。数えるルールをノードごとに散らさず、状態の定義の1行にまとめるのが要点です。組み込みのmaxをリデューサーとしてそのまま渡してはいけません。LangGraphがリデューサーのシグネチャを調べるのに、組み込み関数にはシグネチャがなく、コンパイルで死にます。1枚包んでください。

増えるキーに上限をかける

RECENT_KEEP = 3とリデューサーkeep_recent(old, new)を作って、recentキーが後ろから3つだけ残るようにしてください。intakeは["intake"]、lookupは["lookup"]、finishは["finish:작성", "finish:검토"]を書きます(コロンの後ろの2語は、韓国語で「作成」と「検討」を意味します)。

4つが入ってきて3つだけが残る必要があるので、まとめたあとに切り詰める順序が重要です。上限をノードに入れず、リデューサーに入れてください。ノードは今後も増えるはずで、1か所でも抜けると、黙って漏れ出します。チェックポインターが状態をまるごと保存することが、この上限の理由です。

出すものだけを出す

InputState(質問だけ)・OutputState(答えだけ)を作り、build_public()がStateGraph(State, input=InputState, output=OutputState)でコンパイルしたグラフを返すようにしてください。{"question": ...}だけを入れて動かした結果には、answerキーが1つだけある必要があります。

ノードは相変わらず状態全体を見て、狭くなるのは入口と出口だけです。ノードをつなぐコードを関数に切り出しておけば、build_graph()とbuild_public()が同じ配線を共有できます。出るものを消すコードで防ぐと、新しいキーが増えるたびに遅れをとります。許可リストが自然に防いでくれます。

なぜそう決めたかを残す

/root/work/agstate/state_report.jsonにreducers・trace・sources・used_calls・peak_ms・recent_len・public_keysを、/root/work/agstate/state_report.mdに## 무엇을 상태에 두었나 ## 어떤 리듀서를 왜 붙였나 ## 자라지 않게 막은 곳 ## 바깥에 내보내지 않는 것の4つの節を書いてください(4つの見出しは、順に韓国語で「何を状態に置いたか」「どのリデューサーをなぜ付けたか」「増え続けないように防いだ場所」「外に出さないもの」を意味します)。

数値は手で書かず、自分のグラフを実際に動かして得た値で埋めてください。reducersは、キー名をキーに、付けたリデューサーの名前を値にしたオブジェクトです(例: operator.add)。public_keysは、build_public()の結果に実際に出たキーの一覧です。採点ツールは、同じものを別に計算して突き合わせます。