この記事では、PythonでLangGraphを使って完全なエージェントワークフローを構築する方法を学びます。単一のモデル呼び出しから、ツールを使用するエージェントと永続的な会話メモリまでをカバーします。
取り上げるトピックは以下の通りです：
- 状態、ノード、エッジがどのように組み合わさってLangGraphエージェントの実行フローを定義するのか。
- ツールを登録し、モデルのツール呼び出しをグラフの推論ループを通じてルーティングする方法。
- チェックポインターが、別々のグラフ呼び出し間で会話履歴を永続化する方法。
早速始めましょう。
はじめに
ほとんどの AIエージェント のセットアップは、単一ターンのケースをうまく処理します。つまり、質問を受け取り、モデルを呼び出し、回答を返します。しかし、その直後により難しい問題が発生します。エージェントはデータベースにクエリを実行したり、以前のメッセージのコンテキストを記憶したり、モデルが何を決定したのか、そしてその理由を正確に可視化したりする必要があります。すべてのユースケースに対してカスタムの配管を構築せずにこれらの課題を解決することは、多くの実装が崩れ始めるポイントです。
LangGraph は、これらの各問題を処理するためのクリーンな構造を提供します。エージェントはグラフとして表現され、ノードは作業の単位、エッジは次に何が実行されるかを定義し、共有状態オブジェクトはすべてのステップを通じて完全なメッセージ履歴を運びます。モデルはノード内で実行されるため、すべての推論ステップ、ツール呼び出し、応答がグラフの状態の一部になります。これにより、実行フロー全体が可視化可能、検査可能になり、その後に実行される任意のノードから利用可能になります。
この記事では、すべてのLangGraphグラフの構築基盤となる状態、ノード、エッジのプリミティブを理解する方法、MessagesState で会話履歴を自動的に管理する方法、ノード内で言語モデルを呼び出してグラフに接続する方法、ツールを登録してツール呼び出しをモデルにルーティングする方法、完全なメッセージシーケンスをトレースして各ステップでモデルが何を行うかを確認する方法、そしてチェックポインターで別々の呼び出し間で会話を永続化する方法を学びます。インストール手順から始めて、グラフをゼロから構築していきます。
セットアップ
必要なパッケージをインストールします：
pip install langgraph langchain-openai python-dotenv |
次に、プロジェクトのルートに.envファイルを作成し、OpenAI APIキーを設定します：
OPENAI_API_KEY="your_key_here" |
LangChainまたはLangGraphのインポートより前に、スクリプトの先頭で読み込みます：
from dotenv import load_dotenv load_dotenv() |
python-dotenv は.envファイルを読み込み、キーを環境変数として設定します。
状態、ノード、エッジの理解
すべてのLangGraphグラフは以下の3つのコンポーネントから構築されます。これらを最初に正しく理解することで、グラフが複雑になったときの混乱を避けられます。
状態は、グラフ全体の共有メモリとして機能する TypedDict です。すべてのノードは状態から読み取り、更新を書き戻します。ノード間で他の方法で値が渡されることはありません。ノードで更新しないフィールドは変更されず、変更したいものだけを返します。
ノードは通常のPython関数です。ノードは現在の状態を引数として受け取り、更新したいフィールドの辞書を返します。add_node で関数を登録することで、特別なデコレータや基底クラスを必要とせずにグラフの一部になります。名前文字列なしで関数だけを渡した場合、LangGraphは自動的に関数名を使用します。
エッジは実行順序を定義します。add_edge(A, B) は、ノードAの完了後にノードBを実行することを意味します。add_conditional_edges は、ノードAの完了後にルーティング関数を呼び出し、その指示に従って移動することを意味します。すべてのグラフにはエントリポイントとして START が必要で、少なくとも1つの END へのパスが必要です。
デフォルトでは、ノードが状態フィールドの値を返すと、その値は既存の値を置き換えます。ログやメッセージ履歴など、ノード間で累積すべきフィールドについては、reducer関数 でフィールドに注釈を付けます。以下の例では、リストフィールドに対するoperator.addは、置換ではなく追加を意味します：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
from typing import Annotated import operator from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class TicketState(TypedDict): customer_message: str log: Annotated[list, operator.add] def log_received(state: TicketState) -> dict: return {"log": [f"Received: {state['customer_message']}"]} def log_assigned(state: TicketState) -> dict: return {"log": ["Assigned to support queue"]} builder = StateGraph(TicketState) builder.add_node("log_received", log_received) builder.add_node("log_assigned", log_assigned) builder.add_edge(START, "log_received") builder.add_edge("log_received", "log_assigned") builder.add_edge("log_assigned", END) graph = builder.compile() result = graph.invoke({"customer_message": "My invoice looks wrong", "log": []}) print(result) |
出力は以下のようになります：
{'customer_message': 'My invoice looks wrong', 'log': ['Received: My invoice looks wrong', 'Assigned to support queue']} |
両方のノードがログに書き込み、両方のエントリが存在します。customer_messageはどちらのノードも返さなかったため、変更されずにそのままです。これはまさにMessagesStateがmessagesフィールドを処理する方法で、メッセージオブジェクトの重複排除と順序付けも扱う少し特殊なreducerであるadd_messagesを使用しています。
MessagesStateを使った会話履歴の管理
LangGraphグラフ内のすべてのノードは、現在の状態を読み取り、更新を状態に書き戻します。会話エージェントの場合、状態は完全なメッセージ履歴（ユーザー入力、モデルの応答、ツールの出力）を保持する必要があります。これにより、モデルは次に何をするかを決定する際に必要なコンテキストを常に持つことができます。
LangGraphには、まさにこれを実現するための組み込み状態型が用意されています：MessagesState。これはTypedDictで、単一のmessagesフィールドを持ち、単純な上書きではなくadd_messages reducerを使用します。ノードが新しいメッセージを返すたびに、既存のリストに追加され、置き換えられることはありません。会話履歴を手動でつなぎ合わせる必要はありません。
from langgraph.graph import MessagesState |
これは、ほとんどの単一エージェントグラフに必要な状態定義です。customer_id、priorityフラグなど、ノードが必要とする追加フィールドで拡張できます。しかし、messagesはすでに存在し、累積するようにすでに設定されています。
ノード内でのモデルの呼び出し
状態が整ったら、任意のLangGraphエージェントのコアノードは、現在のメッセージリストをモデルに渡して応答を追加する関数です。モデルはAIMessageを返します。これを「messages」をキーとする辞書内で返すだけで、状態に追加できます。
from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage llm = ChatOpenAI(model="gpt-4o-mini") def run_model(state: MessagesState) -> dict: system = SystemMessage("You are a support agent for a SaaS product. " "Be concise and helpful.") response = llm.invoke([system] + state["messages"]) return {"messages": [response]} |
ChatOpenAIはOpenAI APIをLangChainの標準的なチャットモデルインターフェースでラップします。別のプロバイダー（Anthropic、Google、Ollama経由のローカルモデル）に切り替える場合は、インポートとモデル文字列を変更するだけで、ノードの残りの部分は同じままです。SystemMessage は、状態に保存されることなく、毎回の呼び出しでモデルの役割を設定し、永続的な履歴をクリーンに保ちます。
これをグラフに組み込んで実行します：
from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage builder = StateGraph(MessagesState) builder.add_node("run_model", run_model) builder.add_edge(START, "run_model") builder.add_edge("run_model", END) graph = builder.compile() result = graph.invoke({"messages": [HumanMessage("My dashboard isn't loading. What should I try?")]}) print(result["messages"][-1].content) |
result["messages"] は完全なリストで、元のHumanMessageとモデルが生成したAIMessageが含まれます。[-1]で最新のものを取得します。
ツールの登録とツール呼び出しのルーティング
モデルはトレーニングデータから一般的な質問に答えることができますが、データに固有のもの（アカウント詳細、サブスクリプションティア、チケット履歴）にはツール呼び出しが必要です。モデルはツールが必要なタイミングを決定し、コードはそれが何をするかを定義します。
@tool デコレータでツールを定義します：
from langchain_core.tools import tool @tool def get_customer_tier(customer_id: str) -> str: """Look up the subscription tier for a customer by their ID. Returns 'free', 'pro', or 'enterprise'.""" tiers = { "cust_1001": "enterprise", "cust_2002": "pro", "cust_3003": "free", } return tiers.get(customer_id, "not found") |
docstringは、モデルがこのツールを呼び出すかどうかを決定し、どの引数を渡すかを決定する際に読み取るものです。曖昧なdocstringは呼び出しの見逃しや不正な引数につながるため、正確に保つことが重要です。
ツールをモデルにバインドしてツールの存在を認識させ、ノードを更新します：
tools = [get_customer_tier] llm_with_tools = llm.bind_tools(tools) def run_model(state: MessagesState) -> dict: system = SystemMessage("You are a support agent for a SaaS product. " "Use available tools when you need account-specific information.") response = llm_with_tools.invoke([system] + state["messages"]) return {"messages": [response]} |
bind_tools は、ツールのスキーマをすべてのリクエストと一緒にモデルに送信します。モデルがツールを使用することを決定すると、応答はcontentにプレーンテキストではなく、tool_callsフィールドが設定されたAIMessageとして返ってきます。
実行を処理するためのToolNodeを追加し、ルーティングを接続します：
from langgraph.prebuilt import ToolNode, tools_condition tool_node = ToolNode(tools) builder = StateGraph(MessagesState) builder.add_node("run_model", run_model) builder.add_node("tools", tool_node) builder.add_edge(START, "run_model") builder.add_conditional_edges("run_model", tools_condition) builder.add_edge("tools", "run_model") graph = builder.compile() |
ToolNode は最後のAIMessageからtool_callsを読み取り、モデルが指定した引数で一致する関数を実行し、結果をstateに追加されたToolMessageでラップします。tools_condition は、すべてのモデル呼び出し後に最後のAIMessageをチェックします。tool_callsが空でない場合は「tools」にルーティングし、そうでない場合は「__end__」にルーティングします。「tools」から「run_model」へのエッジはループを閉じるものです。ツールの結果をモデルに送り返し、最終的な回答を生成できるようにします。
推論ループのトレース
次に進む前に、モデルがツールを使用するときにグラフ内で実際に何が起こっているかを考えてみましょう。最終出力が示唆する以上に多くのことが起こっているからです。
result = graph.invoke({"messages": [ HumanMessage("Can you check what plan customer cust_1001 is on?") ]}) for msg in result["messages"]: print(type(msg).__name__, ":", msg.content or msg.tool_calls) |
サンプル出力：
HumanMessage : Can you check what plan customer cust_1001 is on? AIMessage : [{'name': 'get_customer_tier', 'args': {'customer_id': 'cust_1001'}, 'id': 'call_Rx7kLmNpQ2wJtA3s', 'type': 'tool_call'}] ToolMessage : enterprise AIMessage : Customer cust_1001 is on the enterprise plan. |
ここでは、4つのメッセージと2回のモデル呼び出しがあります。最初のモデル呼び出しは、tool_callsが設定されcontentが空のAIMessageを生成します。モデルはまだ回答ではなく、何をしたいかをシグナリングしています。tools_conditionはそれを見てToolNodeにルーティングし、ToolNodeはget_customer_tier("cust_1001")を実行して結果を持つToolMessageを追加します。
「run_model」へのエッジが再び発火します。今、モデルはこれまでの3つのメッセージすべてをコンテキストに持ち、ルックアップが成功したことを理解し、回答をcontentに持つ最終的なAIMessageを書き込みます。tools_conditionはもう一度実行され、ツール呼び出しがないことを確認してグラフを終了します。
このループ（モデル呼び出し、ツール実行、モデル呼び出しの繰り返し）は、標準的なReActパターンです。すべてのツール使用には2回のモデル呼び出しが必要です。1回は検索対象を決定するため、もう1回は結果を解釈するためです。これは、ツールを追加する際にレイテンシとコストを考える上で有用な知識です。
呼び出し間の会話の永続化
上記のすべてのgraph.invoke()は、新しいグラフ状態から開始します。永続化がない場合、モデルは以前のやり取りを記憶しません。
呼び出し間で状態を永続化するには、グラフのコンパイル時にチェックポインターをアタッチします：
from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() graph = builder.compile(checkpointer=checkpointer) |
次に、すべての呼び出しで同じthread_idを渡します：
config = {"configurable": {"thread_id": "ticket-7741"}} graph.invoke( {"messages": [HumanMessage("Hi, I can't access my account.")]}, config, ) result = graph.invoke( {"messages": [HumanMessage("My ID is cust_2002, can you check my plan?")]}, config, ) print(result["messages"][-1].content) |
サンプル出力：
You're on the pro plan, cust_2002. Since you're having trouble accessing your account, I'd recommend resetting your password first. Pro accounts also have priority support available if the issue continues. |
2回目の呼び出しは、1回目の会話を見ることができます。なぜなら、チェックポインターが実行前にスレッドの状態を復元し、実行後に更新された状態を保存したからです。異なるthread_idを使用すると、別々の空の状態から開始されます。
InMemorySaver はプロセスメモリにチェックポイントを保存するため、開発やテストに便利です。本番環境では、通常、データベースやその他の永続的なストレージでバックアップされた永続的なチェックポインターに置き換えます。グラフコードの残りの部分は同じままです。
Checkpointers はスレッドのグラフ状態を永続化します。アプリケーションが会話とは独立してデータ（ユーザープロファイル、設定、複数のスレッドで共有される長期記憶など）を永続化する必要がある場合、Storeを使用します。Storesはチェックポインターを補完し、実行中にグラフがアクセスできる永続的なアプリケーションレベルのストレージを提供します。
まとめ
この記事では、LangGraphエージェントをゼロから構築しました。その過程で、状態がグラフを通じてどのように流れるか、ノードがどのように作業を実行するか、ツールが実行ループにどのように適合するか、チェックポインターが別々の呼び出し間で会話を保存するかを学びました。これらの同じビルディングブロックは、シンプルなチャットボットからより洗練されたエージェントワークフローまでスケールします。
LangGraphの強みの1つは、各部分が独立していることです。言語モデルを交換したり、新しいツールを登録したり、会話を永続化する方法を変更したりしても、グラフの残りの部分を再設計する必要はありません。すべてが共有状態を通じて通信するため、グラフは予測可能で拡張しやすくなります。
同じアイデアはマルチエージェントシステムにも適用できます。リクエストを専門エージェントにルーティングするコーディネーターは、状態、ノード、条件付きエッジを持つグラフのままです。アーキテクチャは大きくなりますが、基盤となるプリミティブは同じままです。
さらに詳しく知りたい場合は、以下のリソースが良い出発点です：
- LangGraph persistence docs
- Tool calling with LangChain
- MessagesState and add_messages
- LangGraph prebuilt components
Happy building!
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.