mastering-langgraph

LangGraph 開発ガイド

ノード（ステップ）をエッジ（遷移）でつないだグラフを定義して、状態を保持する AI エージェントやワークフローを構築します。

目次

• クイックスタート
• 一般的な構築シナリオ
• 基本原則
• 開発ワークフロー
• よくある落とし穴
• 環境設定
• 簡単な動作確認
• API の要点
• 次のステップ

クイックスタート

メモリ機能を備えた最小限のチャットボット：

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

# 1. Define state
class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # Append mode

# 2. Define node
llm = ChatOpenAI(model="gpt-4")

def chat(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 3. Build graph
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# 4. Compile with memory
chain = graph.compile(checkpointer=InMemorySaver())

# 5. Invoke with thread_id for persistence
result = chain.invoke(
    {"messages": [HumanMessage(content="Hello!")]},
    config={"configurable": {"thread_id": "user-123"}}
)
print(result["messages"][-1].content)

主なパターン：

• Annotated[list, operator.add] — 置き換えずにリストに追加します
• InMemorySaver() — 呼び出しをまたいでメモリを保持します
• thread_id — 永続化のために会話を識別します

一般的な構築シナリオ

シンプルなチャットボット／Q&A

上記のクイックスタートで説明しています。必要に応じて、前処理や後処理用のノードを追加してください。

ツールを使用するエージェント

タスクが完了するまで、外部ツール（API、計算機、検索）をループ内で呼び出すエージェントです。 → references/tool-agent-pattern.md を参照してください

構造化ワークフロー

条件分岐、並列実行、またはプロンプトチェーンを含む複数ステップのパイプラインです。 → references/workflow-patterns.md を参照してください

長期メモリを備えたエージェント

セッションをまたいで会話を永続化し、タイムトラベルデバッグを可能にして、クラッシュにも対応します。 → references/persistence-memory.md を参照してください

Human-in-the-Loop

ワークフローの途中で一時停止し、人間による承認、修正、または追加の入力を求めます。 → references/hitl-patterns.md を参照してください

デバッグ／本番環境の監視

ノードをユニットテストし、グラフを可視化して、LangSmith でトレースします。 → references/debugging-monitoring.md を参照してください

マルチエージェントシステム

ハンドオフツールを使用して、スーパーバイザー型またはスウォーム型のマルチエージェントワークフローを構築します。 → references/multi-agent-patterns.md を参照してください

本番環境へのデプロイ

LangGraph Platform（クラウド／セルフホスト可能）またはカスタムインフラストラクチャにデプロイします。 → references/production-deployment.md を参照してください

LangGraph を初めて使う方へ

State、Nodes、Edges、Graph API などの基本概念を学びます。 → references/core-api.md を参照してください

基本原則

1. 状態は加工せずに保持する

フォーマット済みのプロンプトではなく、事実を保存します。各ノードで必要に応じてデータを整形できます。

# ✓ Good: raw data
class State(TypedDict):
    user_question: str
    retrieved_docs: list[str]
    intent: str

# ✗ Bad: pre-formatted
class State(TypedDict):
    full_prompt: str  # Mixes data with formatting

2. ノードの役割は一つに絞る

各ノードは一つの処理を行います。内容がわかる名前を付けてください。

# ✓ Good: clear responsibilities
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_knowledge", search_knowledge)
graph.add_node("generate_response", generate_response)

3. ルーティングを明示する

判断には条件付きエッジを使います。ルーティングロジックをノード内に隠さないでください。

def route_by_intent(state) -> str:
    if state["intent"] == "billing":
        return "billing_handler"
    return "general_handler"

graph.add_conditional_edges("classify", route_by_intent, 
    ["billing_handler", "general_handler"])

4. リストにはアグリゲーターを使う

値を蓄積するリストフィールドには、operator.add が必要です。

class State(TypedDict):
    messages: Annotated[list, operator.add]      # ✓ Appends
    current_step: str                             # Replaces (no annotation)

5. エラーを意図的に処理する

エラーの種類	対処方法
一時的（ネットワーク）	ノードに RetryPolicy を使用する
LLM で復旧可能（解析失敗）	状態を通じてエラーを LLM に渡し、ループバックする
ユーザーが修正可能（情報不足）	interrupt() を使って一時停止し、確認する
想定外（バグ）	デバッグのためにそのまま上位へ伝播させる

開発ワークフロー

1. ステップを定義する — タスクを個別の処理に分割します（それぞれがノードになります）
2. ステップを分類する — LLM 呼び出しですか？ データ取得ですか？ アクションですか？ ユーザー入力ですか？
3. 状態を設計する — 必要なフィールドをすべて含む TypedDict を使い、加工せずに保持します
4. ノードを実装する — 各ステップに def node(state) -> dict を定義します
5. グラフを接続する — add_node()、add_edge()、add_conditional_edges() を使います
6. コンパイルしてテストする — graph.compile() を実行し、サンプル入力でテストします

よくある落とし穴

1. リストで operator.add を指定し忘れる

症状： メッセージが消え、最後のメッセージだけが保持されます。

# ✗ Wrong: messages: list[AnyMessage]
# ✓ Fix: messages: Annotated[list[AnyMessage], operator.add]

2. メモリに thread_id を指定し忘れる

症状： エージェントが前のターンを忘れます。

# ✓ Fix: Always pass config with thread_id
chain.invoke(input, config={"configurable": {"thread_id": "unique-id"}})

3. 呼び出す前にコンパイルしない

症状： グラフオブジェクトで AttributeError が発生します。

# ✗ Wrong: graph.invoke(input)
# ✓ Fix: chain = graph.compile(); chain.invoke(input)

4. @task を使わない非決定的なノード

症状： チェックポイントから再開すると結果が異なります。

from langgraph.func import task

@task  # Wrap for durable execution
def fetch_data(state):
    return {"data": requests.get(url).json()}

5. 型ヒントによる循環インポート

症状： 状態クラスの定義時に ImportError が発生します。

# ✓ Fix: Use string annotations
from __future__ import annotations

環境設定

# Core
pip install -U langgraph

# LLM providers (pick one or more)
pip install langchain-openai
pip install langchain-anthropic

# Production persistence
pip install langgraph-checkpoint-postgres

# Observability
pip install langsmith

環境変数：

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export LANGSMITH_API_KEY="ls-..."
export LANGSMITH_TRACING=true

簡単な動作確認

ビルド前

• python -c "import langgraph; print(langgraph.__version__)" が正常に動作する
• LLM API キーが設定されている（OPENAI_API_KEY または ANTHROPIC_API_KEY）
• 任意：トレース用の LANGSMITH_API_KEY

ビルド後

• グラフがエラーなくコンパイルされる：chain = graph.compile()
• 可視化が出力される：print(chain.get_graph().draw_mermaid())
• サンプル入力で呼び出しに成功する：chain.invoke({...})
• リストが正しく累積される（operator.add の注釈を確認）
• 呼び出し間でメモリが保持される（同じ thread_id で2回テスト）
• 条件分岐によるルーティングが想定どおりに動作する（各分岐をテスト）

API の要点

# Imports
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, Annotated
import operator

# State with append-mode list
class State(TypedDict):
    messages: Annotated[list, operator.add]

# Node signature
def node(state: State) -> dict:
    return {"messages": [new_message]}

# Graph construction
graph = StateGraph(State)
graph.add_node("name", node_fn)
graph.add_edge(START, "name")
graph.add_edge("name", END)

# Conditional routing
graph.add_conditional_edges("from", router_fn, ["option1", "option2", END])

# Compile and run
chain = graph.compile(checkpointer=InMemorySaver())
result = chain.invoke(input, config={"configurable": {"thread_id": "id"}})

# Visualization
print(chain.get_graph().draw_mermaid())

API リファレンスの詳細 → references/core-api.md を参照してください

次のステップ

• ツールエージェント：references/tool-agent-pattern.md
• ワークフロー：references/workflow-patterns.md
• 永続化：references/persistence-memory.md
• 人間参加型：references/hitl-patterns.md
• テスト／モニタリング：references/debugging-monitoring.md
• マルチエージェント：references/multi-agent-patterns.md
• 本番環境：references/production-deployment.md
• 主要な概念：references/core-api.md
• 公式ドキュメント：references/official-resources.md