【ITニュース解説】A Planner That Spawns N Workers: LangGraph.js Send() for Fan-Out You Can't Hardcode

この解説では、AIエージェントシステムにおいて、タスクを動的に多数の並行処理に分割し、実行する方法、そしてそれに伴う考慮事項を説明する。具体的には、LangGraph.jsというライブラリが提供するSendという機能を中心に、その使い方と重要性を初心者にも分かりやすく解説する。

まず、AIエージェントとは、与えられた複雑な目標を達成するために、自律的に計画を立て、行動し、結果を評価するプログラムのことだ。例えば、「モノレポ内の全てのパッケージをチェックせよ」という指示があったとする。このタスクは、多数のパッケージそれぞれを個別にチェックするという、多くの小さなサブタスクに分解できる。このような場合、事前に「パッケージは常に4つ」といった固定的な数で処理を設定してしまうと、パッケージが312個あったときに、残りの308個を処理できないという問題が生じる。

ここで登場するのが、LangGraph.jsのSend機能だ。LangGraph.jsは、AIエージェントがどのようにタスクを処理し、次のステップに進むかをグラフ構造で定義するためのライブラリだ。通常、グラフの「エッジ」（ノード間の接続）は静的で、プログラム作成時に固定される。しかし、Sendを使うと、処理の幅、つまりサブタスクの数が実行時に初めて決定されるような動的なファンアウト（一つの処理から多数の並行処理を生み出すこと）が可能になる。

Sendの仕組みは次のようになる。まず、「プランナー」と呼ばれるノードが、与えられた大きな目標を複数の小さな「サブタスク」に分解する。このプランナーは、例えば「パッケージをチェックせよ」という目標から、個々のパッケージに対応する312個のサブタスクリストを生成する。次に、プランナーの出力に応じて「条件付きエッジ」が機能し、このエッジがそれぞれのサブタスクに対して一つずつSendオブジェクトを返す。各Sendオブジェクトは、どの「ワーカー」ノードを実行するか、そしてそのワーカーノードに渡すべき個別の入力（例えば「パッケージAをチェックせよ」という具体的な指示）を指定する。

このワーカーノードは、受け取ったサブタスクを並行して処理する。重要なのは、各ワーカーが自分の担当するサブタスクの情報だけを受け取り、他のサブタスクや全体の状況を知らないことだ。これにより、処理の独立性が保たれ、意図しないデータアクセスや競合を防ぐことができる。全てのワーカーノードの処理が完了すると、「ジョイン」ノードが実行され、それぞれのワーカーからの結果を集約し、最終的な回答を生成する。

このような動的な並行処理において、特に注意が必要なのが「状態管理」と「エラーハンドリング」、そして「コスト管理」だ。

まず状態管理について。複数のワーカーが同時にグラフの共有状態を更新しようとすると、問題が発生する可能性がある。例えば、各ワーカーが自分の処理結果を一つのリストに追加しようとした場合、同時に書き込みが発生してデータが壊れたり、一部の結果が失われたりする恐れがある。これを避けるために、LangGraph.jsではReducedValueという仕組みが提供されており、これは複数の並行処理からの一連の更新を、定義されたreducer関数を使って安全に一つの結果にまとめ上げる。これにより、ワーカーの処理完了順序に関わらず、一貫性のある最終結果を得ることが可能になる。

次にエラーハンドリングだ。並行処理中に、例えば312個のサブタスクのうち1つが失敗した場合、どうなるのだろうか。LangGraph.jsのデフォルトの挙動では、並行実行されるノード群は「スーパーステップ」として扱われ、その中のどれか一つでもエラーが発生すると、そのスーパーステップで行われた全ての状態更新が破棄されてしまう。つまり、他の311個のワーカーが正常に処理を完了し、結果を生成していたとしても、それらの結果は失われてしまうのだ。しかも、LLMへのAPI呼び出しなど、実際に発生した処理に対する課金は行われてしまうことが多い。これを回避するには、ワーカーノードの内部でエラーを捕捉し、エラーが発生したことを示す結果（例えばok: falseというフラグとエラーメッセージ）として返すのが効果的だ。これにより、部分的な失敗を許容し、成功した他のワーカーの結果は保持したまま、ジョインノードで最終的な判断を下せるようになる。

最後にコスト管理についてだ。動的に多数のワーカーを生成できるのは強力だが、無制限にタスクを生成してしまうと、想定外のコスト（LLM呼び出し料金など）が発生する可能性がある。これに対して、いくつか制御メカニズムがあるが、その役割を正しく理解することが重要だ。 recursionLimitは、グラフが実行できる「スーパーステップ」の最大数を制限するもので、グラフが無限ループに陥るのを防ぐ役割がある。例えば、計画が失敗した場合に再計画を行うようなループを持つグラフでは、この設定が重要になる。しかし、これは並行処理の「深さ」を制限するものであり、生成されるワーカーの「幅」、つまり総タスク数には影響しない。 maxConcurrencyは、同時に実行できるタスクの最大数を制限する。これは、例えばLLMのAPIに対して一度に大量のリクエストを送ってレート制限に引っかかるのを防ぐために役立つが、総タスク数（課金対象となるモデル呼び出しの総数）を制限するものではない。500個のタスクが生成されても、maxConcurrencyが4であれば、4つずつ順番に処理されるだけで、結局は500回全てのモデル呼び出しが行われることになる。 したがって、本当に総タスク数を制限し、コストを管理するためには、プランナーノード自身が、生成するサブタスクの数を制限する必要がある。プランナーが例えば最大8個のサブタスクしか生成しないようにすることで、それ以上のタスクは「切り捨てられた」ことを状態に記録し、ジョインノードでその事実を最終的な回答に含めることができる。この「幅の制限」をプランナーノードに持たせることで、モデルへの指示とコードによる強制の両面から、予期せぬ大量処理を防ぎ、予算内で運用することが可能になる。

まとめると、LangGraph.jsのSend機能は、AIエージェントが実行時に動的にタスクを分割し、並行処理を行うための非常に強力な手段を提供する。しかし、この柔軟性を最大限に活用し、安定したシステムを構築するためには、適切な状態管理、堅牢なエラーハンドリング、そしてコスト意識を持った処理幅の制限といった、システム全体を見据えた設計が不可欠である。これらの要素を理解し、適切に実装することで、複雑な実世界のタスクに対応できる、効率的で信頼性の高いAIエージェントを開発できるようになるだろう。