LLM（大規模言語モデル）を活用したアプリケーションの開発が加速しており、LangChainを用いた開発フレームワークのひとつとしてLangGraphが注目されています。
LangGraphは、LLMワークフローをグラフ構造で管理できる強力なツールですが、State（状態）の管理が煩雑になりがちです。特に、Stateが肥大化すると、デバッグや拡張が困難になるという課題が発生します。
「Stateの整理がうまくいかない…」
「どこで何のデータを使っているのかわからなくなった…」
そんな悩みを解決するため、本記事ではLangGraphのState管理を効率化する分割アプローチを徹底解説します！
✅ Stateの役割と課題を整理
✅ Stateを適切に分割する3つの戦略
✅ LangGraphの内部動作を理解し、最適なState設計を実現
これを読めば、LangGraphのStateをスッキリ整理し、スケールしやすいLLMアプリ設計ができるようになります！
それでは、詳しく見ていきましょう！ 💡
【本記事のもくじ】
🔹 LangGraphにおけるStateとは？
Stateの役割
LangGraphにおけるStateは、ワークフローの各ノード間で受け渡されるデータです。Stateの設計次第で、アプリケーションの可読性、拡張性、保守性が大きく変わります。
Stateの課題
Stateを適切に設計しないと、以下のような問題が発生します。
- すべてのデータを単一のStateに格納 → スケールしにくい
- 不要なデータがノードに渡る → パフォーマンス低下
- どのノードで何が使われているか不明確 → デバッグ困難
このような課題を解決するために、Stateの分割戦略が重要になります。
🛠 LangGraphの基本的なState設計
1. 基本的なStateの定義
LangGraphでは、StateはTypedDictで定義し、StateGraph に渡します。
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
# Stateを定義
class State(TypedDict):
value: str
# Graphを作成
graph = StateGraph(State)
2. Stateを使用するノード
各ノードでは、Stateを受け取り、処理を行います。
from langchain_core.runnables import RunnableConfig
# ノードの作成
def node(state: State, config: RunnableConfig):
return {"value": "1"}
graph.add_node("node", node)
3. Graphの実行
Stateの初期値を渡して実行します。
graph.set_entry_point("node")
print(graph.invoke({"value": ""}))
この方法はシンプルで直感的ですが、Stateの管理が煩雑になりやすいため、適切な分割が求められます。
🎯 Stateを効率的に管理する3つの分割戦略
① InputとOutputの分割
Stateに入力用のデータと出力用のデータが混在すると、どこで何が使われるか分かりにくくなります。
LangGraphのinputとoutput引数を活用し、InputStateとOutputStateを明確に分離しましょう。
✅ 具体的な実装
class InputState(TypedDict):
input_value: str
class OutputState(TypedDict):
output_value: str
class OverallState(InputState, OutputState):
pass
graph = StateGraph(
state_schema=OverallState,
input=InputState,
output=OutputState,
)
# ノードの定義
def node(state: InputState, config: RunnableConfig):
return {'output_value': '2'}
graph.add_node("node", node)
graph.set_entry_point("node")
# Graphを実行
print(graph.invoke({'input_value': '1'})) # {'output_value': '2'}
✅ メリット
- InputとOutputが分離され、データの流れが明確になる
- 不要なデータがノードに渡らないため、デバッグが容易
- 出力を明示的に指定できるため、可読性が向上
② ノード間専用のStateを定義
特定のノード間でのみ使用するデータをInputやOutputのStateに含めると、全体のStateが肥大化して管理が難しくなります。
→ 「ノード間専用のState」を作成し、適切な範囲でのみ利用する！
✅ 具体的な実装
class InputState(TypedDict):
input_value: str
class OutputState(TypedDict):
output_value: str
class OverallState(InputState, OutputState):
pass
# ノード間でのみ使う中間State
class PrivateState(TypedDict):
private_value: str
graph = StateGraph(
state_schema=OverallState,
input=InputState,
output=OutputState,
)
# PrivateStateを使うノード
def node(state: InputState, config: RunnableConfig):
return {'private_value': '2'} # PrivateStateに書き込み
# PrivateStateを受け取るノード
def node2(state: PrivateState, config: RunnableConfig):
return {'output_value': '3'}
graph.add_node("node", node)
graph.add_node("node2", node2)
graph.set_entry_point("node")
print(graph.invoke({'input_value': '1'})) # {'output_value': '3'}
✅ メリット
- Stateの肥大化を防ぎ、全体の可読性が向上
- 特定のノード間でのみ使用するデータを明確化
- 余計なデータの受け渡しを削減し、パフォーマンス向上
③ Stateのスキーマを動的に管理
LangGraphはStateを動的にフィルタリングする機能を持っています。これにより、各ノードが必要なデータのみを受け取ることが可能です。
→ この仕組みを活用し、ノードごとの適切なState設計を行う！
🔍 LangGraphのStateフィルタリングの仕組み
LangGraphでは、Stateの受け渡し時に、ノードが必要なプロパティのみがフィルタリングされます。
🛠 内部の動作
- StateGraphの初期化時に、
state_schema,input,outputを保存 - **ノード追加時（add_node）**に、関数のシグネチャをチェックし、必要なスキーマを登録
- **実行時（invoke）**に、登録済みのスキーマ情報を元に、必要なデータのみを抽出してノードに渡す
この仕組みのおかげで、ノードが不要なデータを受け取らないため、コードの可読性と実行効率が向上します。
🔎 まとめ
LangGraphのStateを適切に管理することで、複雑なワークフローでもスケールしやすい設計が可能になります。
✅ 最適なState分割戦略
- InputとOutputを分離してデータの流れを明確化
- ノード間専用のStateを導入し、肥大化を防ぐ
- Stateフィルタリングの仕組みを活用し、必要なデータのみを渡す
