あなた:
ChatGPT:
from langgraph.graph import StateGraph, Command
from typing import Annotated, Literal, TypedDict
# 状態の型定義
class OverallState(TypedDict):
path: Annotated[list[str], 'add']
graph_builder = StateGraph(OverallState)
# Nodeの定義
def node(state: OverallState) -> Command[Literal['node2']]:
return Command(goto='node2', update={'path': ['node']})
def node2(state: OverallState):
return {'path': ['node2']}
# Nodeの登録
graph_builder.add_node('node', node)
graph_builder.add_node('node2', node2)
# スタート地点と終了地点を設定
graph_builder.set_entry_point('node')
graph_builder.set_finish_point('node2')
graph = graph_builder.compile()
# 実行
print(graph.invoke({'path': []}))
🔹 Commandのポイント
✅ goto='node2'
→ 次に実行するNodeを指定
✅ update={'path': ['node']}
→ 状態を更新
このコードを実行すると、
1️⃣ node
が実行され、状態 {'path': ['node']}
に更新
2️⃣ node2
に遷移し、最終状態 {'path': ['node', 'node2']}
を出力
と動作します。
3. Commandのパラメータ詳細
Commandの初期化パラメータには、ルーティングをより柔軟にするためのオプション があります。
パラメータ | 説明 |
---|---|
graph |
コマンドを送信するグラフ(デフォルトは現在のグラフ) |
update |
状態を更新するための値(dict) |
resume |
実行を再開するための値(interrupt()と併用) |
goto |
次に移動するNode(str, Sendオブジェクト) |
この中でも特に強力なのが goto
に Send を渡す機能 です。
4. Sendを活用した動的ルーティング
🔹 Sendとは?
Commandの goto
には、単にNode名を指定するだけでなく、Send
オブジェクトを渡して次の遷移先を動的に決定 することが可能です。
from langgraph.graph import Send
def node(state: OverallState) -> Command[Literal['node2', 'node3']]:
return Command(
goto=[Send('node2', {'path': []}), Send('node3', {'path': []})],
update={'path': ['node']},
)
✅ Send('node2', {'path': []})
→ node2
に遷移し、初期化された状態で実行
✅ Send('node3', {'path': []})
→ node3
に遷移し、初期化された状態で実行
これにより、動的なルーティングが実現できる ため、エージェントの意思決定や条件分岐に応用可能 です。
5. サブグラフとの連携(親グラフへの遷移)
Commandは、通常のNode遷移だけでなく、サブグラフ(SubGraph)から親グラフのNodeを直接呼び出す ことも可能です。
🔹 親グラフと子グラフの連携
class ParentState(TypedDict):
path: Annotated[list[str], 'add']
parent_graph = StateGraph(ParentState)
# 親Node
def parent_node(state: ParentState):
return {'path': ['parent_node']}
# 親Node2
def parent_node2(state: ParentState):
return {'path': ['parent_node2']}
parent_graph.add_node('parent_node', parent_node)
parent_graph.add_node('parent_node2', parent_node2)
# 子グラフを登録
parent_graph.add_node('child_graph', child_graph.compile())
parent_graph.add_edge('parent_node', 'child_graph')
parent_graph.add_edge('child_graph', 'parent_node2')
# 子グラフの定義
class ChildState(TypedDict):
path: Annotated[list[str], 'add']
child_graph = StateGraph(ChildState)
def child_node(state: ChildState):
return Command(graph=Command.PARENT, goto='parent_node2', update={'path': ['child_node']})
child_graph.add_node('child_node', child_node)
child_graph.set_entry_point('child_node')
✅ 子グラフ child_node
から親グラフ parent_node2
へ直接遷移
✅ Command.PARENT
を指定することで、親グラフのNodeを実行可能
この機能は、マルチエージェントの設計において強力なツール になります!
6. まとめ|Command機能の活用ポイント
✅ 「状態の更新」と「次のNodeの指定」を1つの処理で行える
✅ 従来の add_conditional_edges() に比べて直感的な記述が可能
✅ Send を活用すれば動的ルーティングが容易に実装可能
✅ サブグラフと親グラフ間の柔軟な遷移も実現
LangGraphを活用するなら、Command機能は必須スキル!
開発の効率化にぜひ役立ててください!🚀