分岐が一つ落ちて答えが丸ごと消えた

目標

1つの質問を、複数のデータソースに同時にばらまいて、また集めます。同じ欄に同時に書くとどんな例外が出るか、統合される順序が何で決まるかを自分で確認し、個数が実行時に決まるブランチと、幅の上限、部分的な成功まで作ります。

なぜ重要なのか

LangGraphの実行は、スーパーステップ単位なので、1つのスーパーステップで行けるノードが複数あれば、そのノードたちが一緒に動きます。そのため、ブランチを増やすのは、エッジを数行書けば済みます。難しいのは統合です。同じキーに複数の値が一度に入ってきたとき、リデューサーがないと、InvalidUpdateErrorでグラフが止まります。順番に動くグラフで黙って上書きされていたバグが、並列に変えた瞬間に現れるものなので、この例外は、実は親切なほうです。統合される順序も、誤解しやすいものです。「先に終わったほうが前」ではありません。このラボで、名前を変えながら自分で確認します。データソースが固定でないと、エッジをあらかじめ描けません。Sendは、「このノードをこの入力で動かす」という指示で、リストの長さだけ、そのノードが動きます。そして、そのノードは、状態全体ではなく、Sendが渡した断片だけを見ます。最後に、ブランチ1つが例外を投げると、そのスーパーステップ全体が死に、すでに成功したブランチの結果も一緒に消えます。失敗を値として返せば、部分的な成功を作れ、失敗したデータソースの名前を残せば、部分的な結果が全体の結果のような顔をしなくなります。採点ツールは、書かれた説明を信用しません。書かれたモジュールを実際に読み込んで、任意のトピックと任意のノード名で動かしてみて、採点ツールが別に計算した値と突き合わせます。

ステップ

/root/work/agpara/fanout.pyにSOURCES・hits・State・データソースごとのノード・build_graph()を作成してください。STARTからデータソースの数だけ分かれてENDに向かいます。findingsは連結するリデューサーを使います。
conflict_demo(topic)を追加して、リデューサーのないキーに2つのノードが同じスーパーステップで書くと、どんな例外が出るかを捕まえて、{"error": ..., "message": ..., "note": ...}として返すようにしてください。
merge_order(names)を追加して、ノード名を変えながら、統合される順序を自分で確認できるようにしてください。
combineノードを追加して、ブランチをENDの代わりにcombineに送ってください。combineは、ranked(件数の降順、同点なら名前の昇順)とtotalを作ります。
plan・fan_out・probe・build_dynamic()・search_dynamic(topic, picked)を追加して、Sendで個数が実行時に決まるブランチを作ってください。
MAX_FANOUT = 2とpick_sources(topic, limit=None)を追加して、幅に上限を置いてください。件数が多いデータソースから、同点なら名前の順に選びます。
failuresキーとrun_partial(topic)を追加して、ブランチ1つが見つけられなくても、残りの答えが生き残るようにしてください。
確認したことを記録してください(保存先: /root/work/agpara/fanout_report.json、/root/work/agpara/fanout_report.md)。

参考

実行の契約: 採点ツールは、/root/work/agpara/fanout.pyをPythonモジュールとして読み込んで、上に書いた名前を直接使います。スクリプトとして実行しません。
SOURCESは{자료원이름: {주제: 건수}}です(プレースホルダーはデータソース名、トピック、件数です)。wikiは返金3・配送5・交換2・梱包4、ticketは返金7・配送1・交換4・梱包4、manualは返金2・配送6・交換9・梱包1です。梱包はwikiとticketが同点なので、同点を何で分けるかが表に出ます。hits(source, topic)は、存在しないトピックに0を返します。
データソースのノードは、{"findings": [{"source": 이름, "hits": 건수}]}を返します(プレースホルダーは名前と件数です)。ノード名はデータソース名と同じにしてください。状態のキーと重なってはいけません。重なると、コンパイルのときにValueError: 'x' is already being used as a state keyが出ます。
conflict_demoは、リデューサーがないキー1つとあるキー1つを持つ小さなグラフを別に作って使います。例外を捕まえたら、errorに例外名を、messageにstr(예외)(プレースホルダーは例外です)をそのまま入れ、noteは空の文字列です。例外が出なかったら、errorとmessageが空で、noteに残った値を入れます。
merge_order(names)は、その名前でノードを作って並べ、1回動かしてから、統合されたリストを返します。
combineは、ranked = [건수 내림차순, 동점이면 이름 오름차순으로 정렬한 자료원 이름](角括弧の中は韓国語の説明で、件数の降順、同点なら名前の昇順に並べたデータソース名を意味します)、total = 건수의 합(韓国語の説明で、件数の合計を意味します)です。
Sendで呼んだノードは、状態全体ではなくSendが渡したディクショナリを受け取ります。search_dynamicは、findingsのリストをそのまま返します。
pick_sources(topic, limit=None)は、limitがなければMAX_FANOUTを使います。0以下なら空のリストです。
run_partial(topic)の答え: {"ranked": [...], "total": 정수, "failures": [...정렬됨], "sources": 찾아낸 자료원 수}(プレースホルダーは整数と、ソート済みの意味の語と、見つかったデータソースの数です)。ticketは、自分が知らないトピックについて、例外を投げずに{"failures": ["ticket"]}を返します。
このPodにはインターネットがありません。langgraph 0.2.60がすでに入っています。
公式ドキュメント: Use the graph API・Graph API overview・Typesリファレンス
よくある間違い: ブランチごとにENDに送っておいて、集約するノードを別に置くこと(そうすると、集約するノードがいつ動くのかわかりません)、統合される順序を完了した順序だと信じること、ブランチの中で例外を投げること、幅の制限で同点を適当に分けることです。

一度に3か所を調べる

/root/work/agpara/fanout.pyにSOURCES・hits・State・データソースごとのノード・build_graph()を作成してください。STARTからデータソースの数だけ分かれてENDに向かい、各ノードはfindingsに自分の結果を1つ書きます。

ブランチを作るのは、add_edge(START, 이름)(プレースホルダーは名前です)をデータソースの数だけ呼ぶだけです。重要なのは、findingsに連結するリデューサーを付けることです。ないと、次のステップで見ることになる例外が、ここで先に出ます。ノード名はデータソース名と同じにしつつ、状態のキーとは重ならないようにしてください。

同じ欄に同時に書くと止まる

conflict_demo(topic)を追加して、リデューサーのないキーに2つのノードが同じスーパーステップで書く小さなグラフを作り、出た例外を捕まえて{"error": 예외이름, "message": 예외메시지, "note": ""}(プレースホルダーは例外名と例外メッセージです)として返すようにしてください。

langgraph.errors.InvalidUpdateErrorです。メッセージをでっち上げず、str(예외)(プレースホルダーは例外です)をそのまま入れてください。どのキーが問題かが、そこに書かれています。順番に動くグラフなら、黙って上書きされたはずのことだという点が、このステップの要点です。

統合される順序は偶然ではない

merge_order(names)を追加してください。その名前でノードを作って並べ、1回動かしてから、連結するキーに統合されたリストを返します。

同じ名前を、順序だけ変えて渡してみてください。結果がどう出るかが、このステップの答えです。ループでノードを作るとき、ラムダが最後の名前だけを捕まえてしまう落とし穴に注意してください。デフォルト引数か、包む関数で、名前を束縛する必要があります。

ブランチがすべて終わったあとに1回集める

combineノードを追加して、データソースのノードをENDの代わりにcombineに送ってください。combineは、ranked(件数の降順、同点なら名前の昇順)とtotal(件数の合計)を作ります。

ブランチを集約するノードに集めて送れば、そのノードは、ブランチがすべて終わったあとに1回だけ動きます。ブランチごとにENDに送って集約するノードを別に置くと、そのノードがいつ動くのかわかりません。同点を名前で分けるのは、答えを決定的にするためです。

個数が実行時に決まる

plan・fan_out・probe・build_dynamic()・search_dynamic(topic, picked)を追加してください。fan_outはSendのリストを返し、probeはSendが渡した断片だけを見ます。

Send("노드이름", 넘길딕셔너리)(プレースホルダーはノード名と、渡すディクショナリです)です。条件付きエッジの3番目の引数には、行けるノード名のリストを渡します。probeの中でstate["source"]を使うのは見慣れないかもしれませんが、そのノードが受け取るのは、状態全体ではなく、Sendが渡したものだからです。

幅に上限を置く

MAX_FANOUT = 2とpick_sources(topic, limit=None)を追加してください。件数が多いデータソースから、同点なら名前の昇順で選び、limitがなければMAX_FANOUTを使います。0以下なら空のリストです。

データソースが20個になると、20回の呼び出しが一度に出ていきます。上限を置くことより重要なのは、選ぶルールが決定的でなければならないことです。同点のときに適当に選ぶと、同じ質問に違う答えが出て、昨日と今日を比べられなくなります。

1つが失敗しても残りは生かす

failuresキーとrun_partial(topic)を追加してください。ticketは、自分が知らないトピックについて、例外を投げずに{"failures": ["ticket"]}を返します。答えは{"ranked": [...], "total": 정수, "failures": [...정렬됨], "sources": 정수}です(プレースホルダーは整数と、ソート済みの意味の語です)。

ブランチ1つが例外を投げると、そのスーパーステップ全体が死んで、すでに成功したブランチの結果も一緒に消えます。失敗を値として返せば、集約するノードが「3つのうち2つで見つかり、1つは失敗」を知ることができます。失敗したデータソースの名前を残さないと、部分的な結果が全体の結果のような顔をすることになります。

確認したことを記録する

/root/work/agpara/fanout_report.jsonにconflict_error・merge_order・ranked・total・dynamic・picked・partialを、/root/work/agpara/fanout_report.mdに## 누가 같은 칸에 쓰는가 ## 합쳐지는 순서는 무엇으로 정해지나 ## 갯수가 실행 시점에 정해질 때 ## 하나가 실패하면の4つの節を書いてください(4つの見出しは、順に韓国語で「誰が同じ欄に書くのか」「統合される順序は何で決まるのか」「個数が実行時に決まるとき」「1つが失敗すると」を意味します)。

merge_orderは{"input": [...], "output": [...]}で、自分が渡した名前と出た順序を一緒に入れます。ranked・totalは、환불(韓国語で「返金」を意味する語です)のトピックでグラフを動かした結果、dynamicはsearch_dynamicの答え、pickedはpick_sources('배송')(韓国語で「配送」を意味する語です)の答え、partialはrun_partial('쿠폰')(韓国語で「クーポン」を意味する語です)の答えからfailuresとsourcesだけを入れます。すべて動かして得てください。