⛓️

[LangGraph] Command機能による動的なルーティング

hakotensan

2024/12/25に公開

はじめに

こんにちは。PharmaXでエンジニアをしている諸岡（@hakoten）です。

この記事では、先日LangGraphに新しく追加された機能である、Command について解説します。

環境

この記事執筆時点では、以下のバージョンを使用しています。
LangChain周りは非常に開発速度が早いため、現在の最新バージョンを合わせてご確認ください

python: 3.12.4
langchain: 0.3.11
langgraph: 0.2.59

LangGraphのCommand機能

2024/12/10 時点で、LangGraphに「Command」という新しい機能が追加されたという発表がありました。

Command自体の機能は以前のバージョンから存在していたようですが、今回の発表に合わせてリリースされたLangGraphのバージョンは「0.2.58」になります。

Commandの概要

Commandは、LangGraphのグラフ構築のためのコンポーネントです。これは「Node」「Edge」「State」「Send」などと同様、グラフを構成するための基本的な要素として位置づけられています。

Commandを簡単に説明すると、「状態を更新すると同時に、次に実行するNodeを指定する」機能を提供しています。

これまで行っていた「Nodeの戻り値による状態更新」と「add_conditional_edges で行っていた次の実行Nodeの動的な指定」を同時に行えるコンポーネントと考えるとわかりやすかと思います。

基本的な使い方

基本的な使い方を、簡単なグラフの例を用いて説明します。

class OverallState(TypedDict):
    path: Annotated[list[str], add]

graph_builder = StateGraph(OverallState)

def node(state: OverallState) -> Command[Literal['node2']]:
    # nodeの戻り値として、Commandのインスタンスを返す
    # goto => 次に実行するNode名
    # update => このNodeで更新する状態を指定
    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)

# Edgeの宣言
graph_builder.set_entry_point('node')
graph_builder.set_finish_point('node2')

graph = graph_builder.compile()

print(graph.get_graph().draw_mermaid())
print(graph.invoke({'path': []}))

Commandの使用箇所は以下の部分です。

...
def node(state: OverallState) -> Command[Literal['node2']]:
    # nodeの戻り値として、Commandのインスタンスを返す
    # goto => 次に実行するNode名
    # update => このNodeで更新する状態を指定
    return Command(goto='node2', update={'path': ['node']})
...

これまでのNodeの戻り値では、Stateを更新するためのdictを返していましたが、これからは Command のインスタンスを返すこともできるようになります。

Commandのパラメータは次のとおりです。

パラメータ	説明
graph	コマンドを送信するグラフを指定できる。 - None(未指定): 現在のグラフ（デフォルト） - Command.PARENT: 最も近い親グラフ
update	グラフの状態を更新するための値(dict)。
resume	実行を再開するための値。interrupt()と共に使用される。
goto	次に移動するNode。指定できるのは - Node名(str) - Node名(str)のSequence - Sendオブジェクト - SendオブジェクトののSequence

・・・
# Edgeの宣言(Commandで指定しているルーティングは宣言不要)
graph_builder.set_entry_point('node')
graph_builder.set_finish_point('node2')
・・・

Commandでルーティングを指定する場合には、Nodeの遷移を示すEdgeの宣言は、省略することができます。

このグラフをMermaidで出力すると、以下のようになります。

実行結果は次のとおりです。node の後に node2 が呼ばれ、Stateも更新されていることがわかります。

node: {'path': []}
node2: {'path': ['node']}
{'path': ['node', 'node2']}

Sendを引数に渡す

Command の初期化パラメータである goto には、Send を渡すことも可能です。Send は動的に次のNodeを指定するためのコンポーネントで、これまではadd_conditional_edges の中で使用されていました。

Sendの詳細については、次の記事もぜひご覧ください。

以下にサンプルコードを示します。

class OverallState(TypedDict):
    path: Annotated[list[str], add]

graph_builder = StateGraph(OverallState)

# 戻り値に指定したCommandの型を元にグラフが描画される
def node(state: OverallState) -> Command[Literal['node2', 'node3']]:
    print(f'node: {state}')
    return Command(
        # gotoには、SendおよびSendのlistが指定可能
        goto=[Send('node2', {'path': []}), Send('node3', {'path': []})],
        update={'path': ['node']},
    )

def node2(state: OverallState):
    print(f'node2: {state}')
    return {'path': ['node2']}

def node3(state: OverallState):
    print(f'node3: {state}')
    return {'path': ['node3']}

graph_builder.add_node('node', node)
graph_builder.add_node('node2', node2)
graph_builder.add_node('node3', node3)

graph_builder.set_entry_point('node')
graph_builder.set_finish_point('node2')
graph_builder.set_finish_point('node3')

graph = graph_builder.compile()

print(graph.invoke({'path': []}))
print(graph.get_graph().draw_mermaid())

Commandを使っているのは次の箇所です。

...
# 戻り値に指定したCommandの型を元にグラフが描画される
def node(state: OverallState) -> Command[Literal['node2', 'node3']]:
    print(f'node: {state}')
    return Command(
        # gotoには、SendおよびSendのlistが指定可能
        goto=[Send('node2', {'path': []}), Send('node3', {'path': []})],
        update={'path': ['node']},
    )
...

goto に Send のリストを渡すことで、動的に次のNodeを決定することが可能です。Sendに渡す引数のStateは親のStateを継承しないため、個別に指定する必要があります。(Sendの基本的な指定方法は、add_conditional_edgesでの指定と同じです。)

戻り値の Command[Literal['node2', 'node3']] は実行グラフを描画する際の型ヒントとして使われます。Literal を使ってNode名を指定すると、Mermaidで出力されるグラフに反映されます。

実行結果は次のとおりです。

node: {'path': []}
node2: {'path': []}
node3: {'path': []}
{'path': ['node', 'node2', 'node3']}

子のグラフ(SubGraph)から親のグラフのNodeを呼び出す

Commandの機能は、ほとんどの場合 add_conditional_edges でも代替可能です。しかし興味深い機能として、Commandを使うと「子グラフ（SubGraph）から直接、親グラフのNodeを呼び出す」ことができます。

次のようなグラフを例に Commandを使って、子のNodeから親のNodeを呼び出す方法を紹介します。

このグラフでは、親のグラフから子のグラフ（child_graph）を呼び出しています。

実際のコードを見てみましょう。

コード全体

from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import StateGraph
from langgraph.types import Command


class ChildState(TypedDict):
    path: Annotated[list[str], add]


child_graph = StateGraph(ChildState)


def child_node(state: ChildState):
    print(f'child_node: {state}')
    return Command(
        graph=Command.PARENT,
        goto='parent_node2',
        update={'path': ['child_node']},
    )


def child_node2(state: ChildState):
    print(f'child_node2: {state}')
    return {'path': ['child_node2']}


child_graph.add_node('child_node', child_node)
child_graph.add_node('child_node2', child_node2)

child_graph.set_entry_point('child_node')
child_graph.add_edge('child_node', 'child_node2')
child_graph.set_finish_point('child_node2')


class ParentState(TypedDict):
    path: Annotated[list[str], add]


parent_graph = StateGraph(ParentState)


def parent_node(state: ParentState):
    print(f'parent_node: {state}')
    return {'path': ['parent_node']}


def parent_node2(state: ParentState):
    print(f'parent_node2: {state}')
    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')

parent_graph.set_entry_point('parent_node')
parent_graph.set_finish_point('parent_node2')

graph = parent_graph.compile()

print(graph.invoke({'path': []}))
print(graph.get_graph(xray=2).draw_mermaid())

まずは親グラフの定義です。ここでは特別なことはなく、child_graph を parent_node から呼び出すようになっています。

...
class ParentState(TypedDict):
    path: Annotated[list[str], add]


parent_graph = StateGraph(ParentState)


def parent_node(state: ParentState):
    print(f'parent_node: {state}')
    return {'path': ['parent_node']}


def parent_node2(state: ParentState):
    print(f'parent_node2: {state}')
    return {'path': ['parent_node2']}

# Nodeの宣言
parent_graph.add_node('parent_node', parent_node)
parent_graph.add_node('parent_node2', parent_node2)
# 子グラフをNodeとして実行
parent_graph.add_node('child_graph', child_graph.compile())

# Edgeの宣言
parent_graph.add_edge('parent_node', 'child_graph')
parent_graph.add_edge('child_graph', 'parent_node2')

parent_graph.set_entry_point('parent_node')
parent_graph.set_finish_point('parent_node2')

graph = parent_graph.compile()
...

次に child_graph の定義です。

...
class ChildState(TypedDict):
    path: Annotated[list[str], add]


child_graph = StateGraph(ChildState)

# Commandを使って親のノードを呼び出す
def child_node(state: ChildState):
    print(f'child_node: {state}')
    # graphに Command.PARENT を指定することで、親のNodeをgotoで指定できる
    return Command(
        graph=Command.PARENT,
        goto='parent_node2',
        update={'path': ['child_node']},
    )


def child_node2(state: ChildState):
    print(f'child_node2: {state}')
    return {'path': ['child_node2']}


child_graph.add_node('child_node', child_node)
child_graph.add_node('child_node2', child_node2)

child_graph.set_entry_point('child_node')
child_graph.add_edge('child_node', 'child_node2')
child_graph.set_finish_point('child_node2')
...

ここでは、2つのNode(child_node, child_node2)を定義しています。ポイントは child_node の戻り値に Command を使用している部分です。

サブグラフの本来の遷移先は child_node2 ですが、このコードでは親のNode parent_node2 を指定しています。
この時 Command の引数 graph に Command.PARENT を指定することで、親グラフの parent_node2 を直接呼び出せるようになります。

実行結果は以下のとおりです。

parent_node: {'path': []}
child_node: {'path': ['parent_node']}
parent_node2: {'path': ['parent_node', 'child_node']}
{'path': ['parent_node', 'child_node', 'parent_node2']}

結果として、子グラフの child_node2 には遷移せず、親グラフの parent_node2 が呼び出されていることがわかります。このように Command で Command.PARENT を指定すると、サブグラフの処理を中断して親のNodeを直接呼び出すことが可能になります。

この機能のユースケースとしては、より複雑なエージェント開発において、サブグラフのエージェントから他のサブグラフのエージェントを呼び出すケースなどが想定されているようです。

詳しくは、LangGraphのマルチエージェントについてまとめられた公式ドキュメントを参照すると、よりイメージが湧くと思います。

公式ドキュメントのリンク

マルチエージェントのHandoffs

Command機能のHow-to Guide

Conceptual Guides

終わりに

LangGraphに新しく追加された Command を使うと、「Stateの更新」と「次に実行するNodeの指定」を同時に行えるようになり、複雑で大規模なエージェントを開発するうえで、柔軟なグラフ構築が可能となります。

一方で、これまでの add_conditional_edges を使ったルーティングと機能が一部重複しているため、使い方には注意が必要です。比較的シンプルなルーティングを構築する場合は、これまでのようにEdgeに集約した方がメンテナンス性が高いかもしれません。

とはいえ、柔軟にグラフを構築できる手段が増えたことは喜ばしいことですので、今後複雑なグラフ構築では積極的に使っていきたいと考えています。

PharmaXでは、AIやLLMに関連する技術の活用を積極的に進めています。もし、この記事に興味を持たれた方や、LangGraphの活用に関心がある方は、ぜひ私のXアカウント（@hakoten）やコメントで気軽にお声がけください。PharmaXのエンジニアチームで一緒に働けることを楽しみにしています。

まずはカジュアルにお話できることを楽しみにしています！

PharmaXテックブログPublication

PharmaXエンジニアチームのテックブログです。エンジニアメンバーが、PharmaXの事業を通じて得た技術的な知見や、チームマネジメントについての知見を共有します。 PharmaXエンジニアチームやメンバーの雰囲気が分かるような記事は、note（note.com/pharmax）もご覧ください。