はじめに
LangGraphのGraph APIを学習したので、理解の確認のため、昔流行ったズンドコキヨシをLLMを使わずLangGraphのみで実装してみます。
ズンドコキヨシとは
Twitter(現: X)における以下のツイートを発端とし、2016年ころに流行ったプログラミングネタです。
適度な難易度が相まってか当時は様々な言語で実装されました。
LangGraphとは
LangGraphはLangChainのツールセットの1つです。
LangChainには既に処理を直列に繋げられるChainの機構がありましたが、実際のAIエージェントでは複雑な分岐やループなどが必要で、従来の仕組みでは表現や制御が難しいという課題がありました。
LangGraphはこれらの課題を解決するため開発されました。LangGraphはAIエージェントやワークフローにおける分岐やループ処理をGraph構造で表現できるGraph APIを有しています。
Graph構造はNodeとEdgeで構成され、Node/Edge間でStateを持ち回りながらGraph内を遷移し処理を進めていきます。
画像出典:https://zenn.dev/pharmax/articles/8796b892eed183
この記事で扱うLangGraphの基本的な概念は以下の記事がわかりやすく解説されているのでこちらを参照してください。
LangGraphでズンドコキヨシを実装する
それでは冒頭のツイートの以下の処理をLangGraphで構築していきましょう。
「ズン」「ドコ」のいずれかをランダムで出力し続けて「ズン」「ズン」「ズン」「ズン」「ドコ」の配列が出たら「キ・ヨ・シ!」って出力した後終了
設計
コーディングの前にまずは実装するGraphをフローチャートで設計しました。
このフローチャートに従い、長方形の処理記号はNodeとして、ひし形の判断記号はConditional Edgeとして実装します。Conditional Edgeは条件に応じて遷移先を変えられるEdgeです。
処理記号のうち青に色付けしたzun、doko、kiyoshiの3つのNodeはそれぞれ「ズン」「ドコ」「キ・ヨ・シ!」を出力します。
branch_startとbranch_endは分岐を作るために用意するNodeで、ここでは何もしません。
実装
ではコードを書いていきましょう。実行環境はJupyter Notebookを想定しています。
今回はLLMは使わないので、LangGraphのみをインストールします。
%%capture --no-stderr
%pip install -qU langgraph
Stateの実装
まずはStateを実装します。StateはNode/Edgeで利用できるデータ構造です。
Edgeでランダムに出力した「ズン」「ドコ」が特定のパターンになったか確認したいので、過去の出力は配列として保持しておきたいです。
また通常Stateのプロパティは上書き更新されますが、プロパティを定義する際にReducerを定義すると更新処理を変更できます。
Reducerにoperator.addを指定して各Nodeの出力を過去のStateに連結するようにします。
from operator import add
from typing import List, Literal, Annotated
from typing_extensions import TypedDict
class ZundokoState(TypedDict):
# sequenceは"ズン", "ドコ", "キ・ヨ・シ!"のいずれかを要素に持つリスト
sequence: Annotated[List[Literal["ズン", "ドコ", "キ・ヨ・シ!"]], add]
# Reducerとしてoperator.addを指定するとノードが返したリストを順次結合できる
# Example: add(["a", "b"], ["c"]) => ["a", "b", "c"]
Nodeの処理の実装
次はNodeの処理を実装します。NodeではLLMを呼び出すなどなんらかの副作用を行いStateを更新します。
今回LLMは使わないので単純にStateの更新のみ行います。Stateに「ズン」「ドコ」「キ・ヨ・シ!」追加する関数と、何も追加しない関数を作ります。
# zun, doko, kiyoshi Nodeに紐づける処理
def zun(state: ZundokoState) -> ZundokoState:
return {"sequence": ["ズン"]}
def doko(state: ZundokoState) -> ZundokoState:
return {"sequence": ["ドコ"]}
def kiyoshi(state: ZundokoState) -> ZundokoState:
return {"sequence": ["キ・ヨ・シ!"]}
# branch_start, branch_end Nodeに紐づける処理
def noop(state: ZundokoState) -> ZundokoState:
# 空リストを返す(状態を変えない)
return {"sequence": []}
単純な直列のGraphでの動作確認
まずはここまでで期待通りに動作するか、簡単に zun, doko, kiyoshi, noopを直列に繋いで実行してみます。
from langgraph.graph import StateGraph, START, END
builder = StateGraph(ZundokoState)
builder.add_sequence([zun, doko, kiyoshi, noop])
builder.add_edge(START, "zun")
graph = builder.compile()
LangGraphでは構築したGraphの構造をMermaid.jsを用いて可視化できます。
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))
Graphが構築できたら実行してみます。Reducerにより各Nodeの返り値が結合されるので、["ズン", "ドコ", "キ・ヨ・シ!"]が出力されるはずです。
graph.invoke({"sequence": []})
{'sequence': ['ズン', 'ドコ', 'キ・ヨ・シ!']}
期待通りの結果になったかと思います。
Conditional Edgeの実装
続いてConditional Edgeに紐づける、遷移先を決める関数を実装します。返り値は遷移先のノード名にします。
# Conditional EdgeのRouter関数の実装
import random
def random_select(state: ZundokoState) -> Literal['zun', 'doko']:
"""ランダムに'zun'または'doko'を選択"""
return random.choice(['zun', 'doko'])
def check_pattern(state: ZundokoState) -> Literal['branch_start', 'kiyoshi']:
"""ズンドコのパターンを満たしているかチェック"""
if state["sequence"][-5:] == ["ズン", "ズン", "ズン", "ズン", "ドコ"]:
return 'kiyoshi'
else:
return 'branch_start'
ズンドコグラフの実装
それではこれまでの要素を組み合わせて、最初に設計したズンドコキヨシのグラフを実装していきます。
builder = StateGraph(ZundokoState)
builder.add_node("branch_start", noop)
builder.add_node("zun", zun)
builder.add_node("doko", doko)
builder.add_node("branch_end", noop)
builder.add_node("kiyoshi", kiyoshi)
builder.add_edge(START, "branch_start")
builder.add_conditional_edges("branch_start", random_select)
builder.add_edge("zun", "branch_end")
builder.add_edge("doko", "branch_end")
builder.add_conditional_edges("branch_end", check_pattern)
builder.add_edge("kiyoshi", END)
graph = builder.compile()
設計通りの構造になっているか描画して確認してみます。Conditional Edgeは破線で表示されます。
display(Image(graph.get_graph().draw_mermaid_png()))
それではグラフを実行してみましょう。
デフォルトでは再帰回数の上限が25に設定されています。上限を超過すると例外が発生して実行が止まるため、上限を1000回に引き上げておきます。
state = graph.invoke({"sequence": []}, {"recursion_limit": 1000})
print(''.join(state["sequence"]))
ドコドコドコズンズンズンズンズンズンドコキ・ヨ・シ!
よさそうです。
補遺:LangGraphを学ぶ
LangChainにはLangChain Academyという公式の学習サイトがあります。
コース "Foundation: Introduction to LangGraph" は今回取り扱ったようなLangGraphの基本的な使い方を動画で説明してくれるので、動画に従って順に学んでいきたい場合にはよさそうです。無料ですがユーザー登録が必要です。また説明は英語です。
コースで使用されるノートブックはGitHubで公開されており登録なしで見られます。
またLangChain/LangGraphの公式ドキュメントはチュートリアルをまとめたページやコアコンセプトを解説したページがあるので、公式ドキュメントも併せて確認すると良いかと思います。
おわりに
今回はLangGraph単体の基本的なコンセプトと実装方法を理解するためLLM等は用いず簡単なズンドコキヨシの処理を実装してみました。
実際にAIエージェントのような生成AIアプリケーションを実装していくにはLangChainなども組み合わせて実装が必要であり、公式ドキュメントを確認しながら勉強していくとよさそうです。