LangGraphでAIエージェントを「グラフ構造」で組み立てる — ツール呼び出し・会話メモリ・条件分岐を自前実装なしで実現する方法

7月20日、Bala Priya Cが「Building Agentic Workflows in Python with LangGraph」と題した記事を公開した。LLMを組み込んだシステムを本番運用しようとすると、単発の質問応答だけでは済まない場面が多い。外部データを参照するツール呼び出し、過去のやり取りを保持するメモリ、処理フローの可視性——こうした要件を毎回自前で実装するのはコストが高い。LangGraphはこの問題に対し、エージェントの動作をグラフ構造（ノード・エッジ・共有ステート）として表現することで解決する。

LangGraphの基本構造：State・Node・Edge

LangGraphのグラフは3つの要素で成り立つ。

State：グラフ全体の共有メモリ。TypedDictとして定義し、すべてのノードがここから読み書きする
Node：処理の単位。普通のPython関数で、引数にStateを受け取り更新内容をdictで返す
Edge：実行順序の定義。add_edge(A, B)でAの後にBを実行、add_conditional_edgesで条件分岐ができる

特に重要なのがStateのリデューサーだ。デフォルトでは新しい値が上書きされるが、Annotated[list, operator.add]のようにリデューサー関数を指定すると追記モードになる。会話履歴の蓄積にはこの仕組みが使われている。

class TicketState(TypedDict):
    customer_message: str
    log: Annotated[list, operator.add]

MessagesStateでの会話履歴管理

会話エージェントを作る場合、メッセージ履歴の管理を自前で書く必要はない。LangGraphが提供する**MessagesState**を使えば、messagesフィールドが最初からadd_messagesリデューサー付きで用意されている。各ノードが返す新しいメッセージは、既存リストに追記される。

from langgraph.graph import MessagesState

モデル呼び出しノードはこのように書く：

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]}

SystemMessageはStateに保存せず毎回付与することで、永続化される会話履歴をクリーンに保つ設計になっている。プロバイダーをAnthropicやOllamaに切り替えたい場合も、importとモデル名を変えるだけでノード本体は変わらない。

ツール呼び出しとReActループの実装

この記事で最も実践的な部分がツール統合だ。モデルが「アカウント情報が必要」と判断したとき、自律的にツールを呼び出して結果を受け取り、最終回答を生成するReActループを構築する。

ツールは@toolデコレータで定義する。docstringがモデルへの説明文になるため、曖昧な記述は呼び出し漏れや引数エラーの原因になる。

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")

グラフへの組み込みはToolNodeとtools_conditionを使う：

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()

tools_conditionはモデルの出力にtool_callsが含まれていればtoolsノードへ、なければ__end__へルーティングする。toolsからrun_modelへのエッジがループを閉じており、ツール結果をモデルに戻して最終回答を生成させる。

実際にグラフを実行すると、Stateのmessagesリストには以下のシーケンスが蓄積される：

HumanMessage（ユーザー入力）
AIMessage（tool_callsフィールド付き）
ToolMessage（ツールの実行結果）
AIMessage（最終回答）

このシーケンスをトレースすれば、モデルが何をどの順番で判断したかが完全に追える。

チェックポインターによる会話の永続化

デフォルトではgraph.invoke()を呼ぶたびに新しい会話として扱われる。チェックポインターを使うと、thread_idで会話を識別して履歴を跨いで継続できる。

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "session_001"}}
graph.invoke({"messages": [HumanMessage("...")]}, config=config)

同じthread_idで呼び出すと、前回の会話コンテキストが自動的に復元される。MemorySaverはインメモリ実装なのでプロセスをまたぐ永続化には別途ストレージの設定が必要だが、動作確認やプロトタイピングには十分だ。

セットアップ

pip install langgraph langchain-openai python-dotenv

.envファイルにOpenAI APIキーを記述し、スクリプト冒頭で読み込む：

from dotenv import load_dotenv
load_dotenv()

詳細はBuilding Agentic Workflows in Python with LangGraphを参照していただきたい。