概要
langgraphを触り始めたので、挙動を理解するためのクイズを作ってみました。
公式のチュートリアルでは、LLMの呼び出しなどのnodeを例にしていますが、少しnode自体の挙動が複雑でSateGraphの本質的な挙動の理解に集中できなかったので、よりシンプルな関数をnodeにしてみることで理解を試みるのが目的です。
問1
以下のスクリプトの出力は？
- "a"
- "b"
- "ab"
- "ba"
from langgraph.graph import END, StateGraph
def func_a(state: str):
return "a"
workflow = StateGraph(str)
workflow.add_node("node_a", func_a)
workflow.add_edge("node_a", END)
workflow.set_entry_point("node_a")
app = workflow.compile()
response = app.invoke("b")
print(response)
正解
1の"a"です。
解説
基本的に、END手前のnodeの出力がapp.invoke全体の戻り値になります。
今回は、func_aがEND手前の最終nodeで、stateの値と無関係に固定値"a"を返す関数なので、それがそのままapp.invokeの戻り値となります。
問2
以下のスクリプトの出力は？
- "a"
- "b"
- "ab"
- "ba"
ポイントは、func_aの戻り値が問１の"a"と比べて、state+"a"になったことです。
from langgraph.graph import END, StateGraph
def func_a(state: str):
return state + "a"
workflow = StateGraph(str)
workflow.add_node("node_a", func_a)
workflow.add_edge("node_a", END)
workflow.set_entry_point("node_a")
app = workflow.compile()
response = app.invoke("b")
print(response)
正解
4の"ba"です。
解説
func_aのstateに何が代入されているかを理解できるとよいです。
stateは通常、直前のnodeの出力が代入されており、nodeがentry_pointの場合は、app.invokeの入力が代入されています。よって、今回のケースでは、app.invokeの入力である"b"がfunc_aにおけるstateの値です。それに"a"を加算して返しているので、"ba"となります。
問3
以下のスクリプトの出力は？
- "a"
- "b"
- "ab"
- "ba"
ポイントはtyping.Annotatedを使って、strにoperator.addというメタデータを付与しているところです。
from langgraph.graph import END, StateGraph
from typing import Annotated
import operator
annotated_str = Annotated[str, operator.add]
def func_a(state: annotated_str):
return "a"
workflow = StateGraph(annotated_str)
workflow.add_node("node_a", func_a)
workflow.add_edge("node_a", END)
workflow.set_entry_point("node_a")
app = workflow.compile()
response = app.invoke("b")
print(response)
正解
4の"ba"です。
解説
typing.Annotatedモジュールは型にmetadataを付与するライブラリ（？）です。
付与されたmetadataは__metadata__[0]みたいな感じで取り出すことができます。
StateGraphはstateを更新するとき、__metadata__が存在しない型に対しては代入し、__metadata__が存在する型は紐づいた関数を実行してstateを更新します。
今回はoperator.addを紐づけているので、func_aの戻り値はoperator.addによってstateに加算されます。
問4
以下のスクリプトの出力は？
- "a"
- エラー
ポイントは、app.invokeにStateGraphの初期化時に与えたstrとは異なる型を入力しているところです。
from langgraph.graph import END, StateGraph
from typing import Annotated
def func_a(state: str):
return "a"
workflow = StateGraph(str)
workflow.add_node("node_a", func_a)
workflow.add_edge("node_a", END)
workflow.set_entry_point("node_a")
app = workflow.compile()
response = app.invoke(["b"])
print(response)
正解
1の"a"
解説
StateGraphが想定するstateと異なる型をinvokeで与えてもエラーにはならないこともあります。
揃えたほうが無難だとは思いますが。
問5
以下のスクリプトの出力は？
- {"messages": "a"}
- {"messages2": "b", "messages": "a"}
- エラー
ポイントはStateGraphの初期化時に与える型がTypedDictになったことです。
from langgraph.graph import END, StateGraph
from typing import Annotated, TypedDict
class State(TypedDict):
messages: str
def func_a(state: State):
return {"messages": "a"}
workflow = StateGraph(State)
workflow.add_node("node_a", func_a)
workflow.add_edge("node_a", END)
workflow.set_entry_point("node_a")
app = workflow.compile()
response = app.invoke({"messages2": "b"})
print(response)
正解
3のエラーです。
ちなみに以下のエラーです。
InvalidUpdateError: Must write to at least one of ['messages']
解説
問4でstateの型とnodeの入出力の型は一致していなくてもエラーになりませんでしたが、StateがTypedDictの場合は、TypedDictで指定されているkeyを少なくとも１つ含めた入力をする必要があります。
公式チュートリアル含め多くの使い方ではStateはTypedDictなので、基本的にはStateとnodeの入出力の型は一致させると覚えるほうが無難かもしれません。
おわりに
公式チュートリアルよりもシンプルな設定で、nodeやStateGraphの入出力を調べるためのクイズを作ってみました。
個人的には、stateがどのように更新されているかが表に出てこずに難しかったので、nodeの戻り値がstateにどう影響するのか、少し理解が深まってよかったです。
今回は、StateGraph自体はほぼブラックボックスとして、入出力を眺めただけでしたが、そのうちソースコードも見てみたく思っています。