LangGraph入門チュートリアル

LangGraph は LangChain チームによって開発された低レベルの Agent オーケストレーションフレームワーク、ステートフル（Stateful）で長時間実行される AI ワークフローの構築に特化して設計されています。

従来の線形LLM呼び出しチェーンとは異なり、LangGraphはワークフローを次のようにモデル化します。有向グラフ（Directed Graph）：

• ノード（Node）：具体的な操作を実行する関数（LLMの呼び出し、ツールの実行、データ処理など）
• エッジ（Edge）：ノード間の遷移パスを定義し、条件分岐をサポートする
• 状態（State）：ワークフロー全体で共有・受け渡しされるデータ

オープンソースのURL：https://github.com/langchain-ai/langgraph。

交響楽の演奏を指揮しているところを想像してください。従来のLLM Chainは最初から最後まで順番に演奏するだけの曲のようなものです。一方LangGraphは指揮者のようなもので、会場の観客の反応に応じて演奏順序を自由に調整し、ある楽章を繰り返したり、特定の段落へジャンプしたりできます。これによりAIワークフローは「指揮」の知恵を持ち、ループ・分岐・バックトラックが可能になり、複雑な自律的決定を真に実現します。

なぜLangGraphを選ぶのか？

以下の表は、従来のLLM ChainとLangGraphの主な相違点を比較しています。

適用シナリオ

• チャットボット：複数ターンの会話コンテキストの記憶が必要
• 自律型Agent：計画立案、ツール使用、反復的思考が可能
• マルチAgentシステム：複数のAIが協力して複雑なタスクを完了する
• 承認ワークフロー：人間による審査が必要な自動化プロセス
• 研究アシスタント：多段階の推論と情報検索が必要

中核概念

コードを書き始める前に、LangGraphの3つの中核概念を理解しておきましょう。

Graph（グラフ）

Graphはワークフロー全体の青写真であり、Agentの完全な論理構造を定義します。ノード（Nodes）とエッジ（Edges）で構成されます：

StateGraph
   |-- Nodes（节点）
   |     |-- node_a
   |     |-- node_b
   |     +-- node_c
   +-- Edges（边）
         |-- START -> node_a
         |-- node_a -> node_b（条件边）
         |-- node_a -> node_c（条件边）
         +-- node_b -> END

State（状態）

Stateはグラフ全体を貫通する共有データ構造です。各ノードはStateを読み取り・更新でき、更新されたStateは次のノードに渡されます。

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表（自动追加）
    user_name: str                            # 用户名称
    step_count: int                           # 步骤计数

Nodes（ノード）

ノードは通常のPython関数で、現在のStateを受け取り、更新されたState（部分フィールド）を返します。

def my_node(state: MyState) -> dict:
    # 读取状态
    messages = state["messages"]
    
    # 执行操作...
    result = "处理结果"
    
    # 返回更新的字段（不需要返回所有字段）
    return {"messages": [{"role": "ai", "content": result}]}

Edges（エッジ）

エッジはノード間の遷移方法を定義します：

• 通常のエッジ：固定パス、node_a -> node_b
• 条件付きエッジ：Stateに基づく動的ルーティング、node_a -> node_b 或 node_c
• 開始エッジ：START -> 第一个节点
• 終了エッジ：某节点 -> END

環境構築

依存関係のインストール

国内ミラーを使用してLangGraphと関連依存関係をインストールします：

# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate      # macOS/Linux
# venv\Scripts\activate       # Windows

# 安装 LangGraph 和 LangChain
pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/

# 可选：安装开发工具
pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/

APIキーの設定

プロジェクトのルートディレクトリに.envファイルを作成し、同時にOpenAIとDeepSeekを設定します：

# .env 文件内容

# OpenAI 配置（国外用户）
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://api.openai.com/v1

# DeepSeek 配置（国内用户推荐）
DEEPSEEK_API_KEY=sk-xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-pro

インストールの確認

import langgraph
print(f"LangGraph 版本: {langgraph.__version__}")

最初のLangGraphプログラム

最も簡単な例から始めましょう。2つのノードだけの線形ワークフローです。

実行結果：

[greet_node] 收到消息: 世界
[process_node] 处理消息: 你好！世界

最终结果: {'message': '你好！世界', 'processed': True}

グラフ構造の可視化

Jupyter Notebook ではグラフ構造を直接可視化できます：

# 在 Jupyter Notebook 中可视化
from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

# 或者打印 Mermaid 格式
print(graph.get_graph().draw_mermaid())

State（状態管理）

TypedDictを使用して状態を定義

from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 消息历史（add_messages reducer 自动追加而非覆盖）
    messages: Annotated[list, add_messages]
    
    # 普通字段（直接覆盖）
    user_id: str
    session_id: str
    
    # 可选字段
    error: Optional[str]
    
    # 计数器（使用 operator.add 作为 reducer）
    retry_count: Annotated[int, lambda x, y: x + y]

Pydanticを使用して状態を定義（本番環境での利用を推奨）

from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class ProductionState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    user_id: str = ""
    confidence_score: float = 0.0
    
    class Config:
        arbitrary_types_allowed = True

MessagesState（組み込みの簡易状態）

LangGraph は組み込みのMessagesStateMessagesState を提供しており、会話シナリオ向けに設計されています：

from langgraph.graph import MessagesState

# MessagesState 等价于:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], add_messages]

# 直接使用，无需自定义
builder = StateGraph(MessagesState)

Nodes（ノード）

通常の関数ノード

def simple_node(state: AgentState) -> dict:
    # 读取状态
    last_message = state["messages"][-1]
    
    # 执行操作
    response = f"收到: {last_message.content}"
    
    # 返回部分状态更新
    return {
        "messages": [{"role": "assistant", "content": response}]
    }

LLM呼び出しノード

非同期ノード

import asyncio

async def async_node(state: AgentState) -> dict:
    """异步节点，适合 I/O 密集型操作"""
    # 模拟异步操作（如 API 调用、数据库查询）
    await asyncio.sleep(0.1)
    
    result = await some_async_api_call(state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": result}]}

# 使用异步图
result = await graph.ainvoke({"messages": [...]})

クラスをノードとして使用

class RouterNode:
    def __init__(self, llm, system_prompt: str):
        self.llm = llm
        self.system_prompt = system_prompt
    
    def __call__(self, state: AgentState) -> dict:
        """类实例可以作为节点使用"""
        messages = [
            SystemMessage(content=self.system_prompt),
            *state["messages"]
        ]
        response = self.llm.invoke(messages)
        return {"messages": [response]}

# 添加类节点
router = RouterNode(llm, "你是一个专业的路由助手。")
builder.add_node("router", router)

Edges（エッジ）と条件付きルーティング

通常のエッジ

# 固定路径：node_a 完成后始终执行 node_b
builder.add_edge("node_a", "node_b")

# 结束：node_a 完成后图结束
builder.add_edge("node_a", END)

条件付きエッジ

条件付きエッジは LangGraph の中心的な機能であり、現在の State に基づいて次のステップを動的に決定します。

def route_after_llm(state: AgentState) -> str:
    """
    路由函数：根据 LLM 的最新输出决定走哪条路径
    返回值必须是已注册节点名称或 END
    """
    last_message = state["messages"][-1]
    
    # 如果 LLM 请求使用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return END

# 添加条件边
builder.add_conditional_edges(
    "llm",              # 源节点
    route_after_llm,    # 路由函数
    {
        "tools": "tool_executor",   # 返回 "tools" 时 -> tool_executor 节点
        END: END                    # 返回 END 时 -> 结束
    }
)

並列実行（Fan-out）

# 从一个节点并行分叉到多个节点
builder.add_edge("start_node", "branch_a")
builder.add_edge("start_node", "branch_b")
builder.add_edge("start_node", "branch_c")

# 多个节点汇聚到一个节点（Fan-in）
builder.add_edge("branch_a", "merge_node")
builder.add_edge("branch_b", "merge_node")
builder.add_edge("branch_c", "merge_node")

完全な条件付きルーティングの例

実行結果の例：

ユーザー: 北京今天天气怎么样？
アシスタント: 很抱歉，我没有实时天气数据。不过北京现在是春季，建议您出门时关注天气预报...
--------------------------------------------------

ユーザー: 帮我写一个 Python 快速排序
アシスタント: 好的，下面是一个 Python 实现的快速排序算法：

```python
def quicksort(arr):
    if len(arr) <= 1:
        return arr
    ...
--------------------------------------------------

ユーザー: 你好，介绍一下你自己
アシスタント: 你好！我是一个 AI 助手，很高兴为你服务。我可以帮你回答各种问题...
--------------------------------------------------

ユーザー: 再见啦！
アシスタント: 再见！期待下次与你交流。
--------------------------------------------------

チャットボットの構築

ここで学んだ知識を使って、マルチターン対話をサポートするボットを構築しましょう。このボットは会話のコンテキストを記憶し、連続した対話体験を実現できます。

実行結果の例：

あなた: 你好，我叫小明
アシスタント: 你好，小明！很高兴认识你。有什么我可以帮助你的吗？

あなた: 我想学习 Python 编程
アシスタント: 太好了！Python 是一门很适合初学者的编程语言。我可以帮你从基础开始：
1. 首先了解变量和数据类型
2. 学习条件语句和循环
3. 掌握函数的定义和使用
你想从哪个部分开始呢？

あなた: 你还记得我叫什么吗？
アシスタント: 当然记得，你叫小明！你刚才说想学习 Python 编程，我们可以继续这个话题。

あなた: 退出
再见！

ツール呼び出し - ReAct Agent

Agent に外部ツールを使用する能力を持たせることは、LangGraph の最も強力な機能の1つです。ReAct（Reason + Act）は最も一般的な Agent パターンです：LLM が思考し -> ツールを選択し -> ツールを実行し -> 結果を観察し -> 思考を続けます。

ツールの定義

まず Agent が使用できるツールを定義します：

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息。
    
    Args:
        query: 搜索关键词
    
    Returns:
        搜索结果摘要
    """
    # 实际项目中替换为真实搜索 API
    return f"关于 '{query}' 的搜索结果：这是模拟的搜索结果..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。
    
    Args:
        expression: 数学表达式，如 '2 + 2' 或 '100 * 0.8'
    
    Returns:
        计算结果
    """
    import ast
    import operator
    
    # 安全的运算符映射
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }
    
    def safe_eval(node):
        if isinstance(node, ast.Expression):
            return safe_eval(node.body)
        elif isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = safe_eval(node.left)
            right = safe_eval(node.right)
            return ops[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = safe_eval(node.operand)
            return ops[type(node.op)](operand)
        else:
            raise ValueError(f"不支持的表达式类型: {type(node)}")
    
    try:
        tree = ast.parse(expression, mode='eval')
        result = safe_eval(tree)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称
    
    Returns:
        天气信息
    """
    # 实际项目中替换为真实天气 API
    return f"{city} 今日天气：晴，温度 22C，湿度 60%"

tools = [search_web, calculate, get_weather]

ReAct Agentの構築

ツール呼び出しプロセスのストリーミング出力

# 流式观察 Agent 的每一步
for chunk in graph.stream(
    {"messages": [HumanMessage(content="搜索 LangGraph 的最新特性")]},
    stream_mode="updates"
):
    for node_name, updates in chunk.items():
        print(f"\n=== 节点: {node_name} ===")
        for msg in updates.get("messages", []):
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"  -> 调用工具: {tc['name']}({tc['args']})")
            else:
                print(f"  -> 输出: {msg.content[:200] if msg.content else '(无内容)'}")

Human-in-the-Loop（人間とAIの協働）

LangGraph はワークフロー実行中に一時停止して、人間の承認や入力を待つことをネイティブにサポートしています。これは人間の確認が必要な機密操作に非常に役立ちます。

interruptを使用して実行を一時停止

from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver

def sensitive_action_node(state: MessagesState) -> dict:
    """执行敏感操作前请求人工审批"""
    last_msg = state["messages"][-1].content
    
    # 暂停图的执行，等待人工决策
    human_decision = interrupt({
        "question": "是否批准执行以下操作？",
        "action": last_msg,
        "risk_level": "中等"
    })
    
    if human_decision == "approve":
        return {"messages": [{"role": "assistant", "content": "操作已批准并执行完毕。"}]}
    else:
        return {"messages": [{"role": "assistant", "content": "操作已取消。"}]}

# 必须使用 checkpointer 才能支持 interrupt
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

完全な承認ワークフロー

エッジにブレークポイントを設定

# 另一种方式：在编译时指定断点
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_node"],   # 执行该节点前暂停
    # interrupt_after=["review_node"],     # 执行该节点后暂停
)

永続メモリ

LangGraph は組み込みの状態永続化メカニズムを提供し、Agent がセッションをまたいで会話履歴を記憶できるようにします。

メモリストレージ（開発テスト向け）

from langgraph.checkpoint.memory import MemorySaver

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

# 使用 thread_id 区分不同会话
config_user_a = {"configurable": {"thread_id": "user-alice"}}
config_user_b = {"configurable": {"thread_id": "user-bob"}}

# Alice 的对话
graph.invoke({"messages": [HumanMessage(content="我叫 Alice")]}, config=config_user_a)
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_a)
# Agent 能记住：你叫 Alice

# Bob 的对话完全独立
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_b)
# Agent 不知道 Bob 的名字（不同 thread_id）

SQLite永続ストレージ（ローカルプロジェクト向け）

# 安装依赖
# pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

# 数据持久化到文件，程序重启后对话历史仍存在
with SqliteSaver.from_conn_string("./chat_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "persistent-chat"}}
    
    # 第一次运行
    graph.invoke({"messages": [HumanMessage(content="我叫张三")]}, config=config)
    
    # 程序重启后再次运行，记忆仍然存在
    result = graph.invoke(
        {"messages": [HumanMessage(content="你还记得我叫什么吗？")]},
        config=config
    )

会話履歴の表示

# 获取某个 thread 的完整状态历史
history = list(graph.get_state_history(config))

for snapshot in history:
    print(f"时间: {snapshot.created_at}")
    print(f"消息数: {len(snapshot.values['messages'])}")
    print("---")

# 获取当前状态
current_state = graph.get_state(config)
print(f"当前消息数: {len(current_state.values['messages'])}")

マルチAgentシステム

LangGraph は、複数の専門化された Agent が連携して動作するよう調整するのが得意です。複雑なタスクを異なる専門 Agent に分割することで、より強力な問題解決能力を実現できます。

主従アーキテクチャ（Supervisorパターン）

サブグラフ（Subgraph）

複雑なサブプロセスをサブグラフとしてカプセル化し、メイングラフで再利用します：

# 将复杂子流程封装为子图，在主图中复用
sub_builder = StateGraph(MessagesState)
sub_builder.add_node("step1", step1_node)
sub_builder.add_node("step2", step2_node)
sub_builder.add_edge(START, "step1")
sub_builder.add_edge("step1", "step2")
sub_builder.add_edge("step2", END)
sub_graph = sub_builder.compile()

# 在主图中使用子图
main_builder = StateGraph(MessagesState)
main_builder.add_node("preprocessing", preprocess_node)
main_builder.add_node("sub_workflow", sub_graph)  # 直接使用编译好的子图
main_builder.add_node("postprocessing", postprocess_node)

main_builder.add_edge(START, "preprocessing")
main_builder.add_edge("preprocessing", "sub_workflow")
main_builder.add_edge("sub_workflow", "postprocessing")
main_builder.add_edge("postprocessing", END)

main_graph = main_builder.compile()

LangGraph Studioでの可視化デバッグ

LangGraph Studio は公式提供のビジュアル開発環境であり、Agent の実行プロセスをリアルタイムで確認でき、開発とデバッグの効率を大幅に向上させます。

LangGraph CLIのインストール

pip install langgraph-cli -i https://mirrors.aliyun.com/pypi/simple/

プロジェクト設定ファイルの作成

プロジェクトのルートディレクトリに作成langgraph.json設定ファイル：

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./my_agent.py:graph"
  },
  "env": ".env"
}

開発サーバーの起動

langgraph dev

起動後にアクセスhttp://localhost:8123ブラウザで LangGraph Studio を使用できます。

Studioの主な機能

• リアルタイム可視化：ノードの実行プロセスをグラフィカルに表示し、ワークフローの状態を直感的に把握
• 状態チェック：任意のノードで一時停止して現在の State を確認でき、問題の調査に便利
• タイムトラベル：過去の実行ステップをリプレイし、各ステップの状態変化を追跡
• ホットリロード：コードを変更するとグラフ構造が自動的に更新され、サービスの再起動は不要

ベストプラクティスとよくある質問

ベストプラクティス

状態設計のポイント

• State を簡潔に保ち、必要なフィールドのみを含める
• 複雑なフィールドには明確な reducer を定義する（例：add_messages）
• 本番環境では Pydantic モデルを使用して状態タイプを検証する
• State に過度に大きなオブジェクトを保存しないようにし、外部ストレージの使用を検討する

ノード設計のポイント

• 各ノードは単一の責務を持たせ、テストと再利用を容易にする
• ノード関数は冪等であるべき（同じ入力から同じ出力を生成する）
• ノード内で渡された state を直接変更せず、新しい値を返す
• I/O 集約型操作を処理するために非同期ノードを適切に使用する

エラーハンドリング

def robust_node(state: AgentState) -> dict:
    try:
        result = risky_operation(state)
        return {"messages": [result], "error": None}
    except Exception as e:
        return {
            "error": str(e),
            "messages": [{"role": "assistant", "content": f"操作失败: {e}"}]
        }

無限ループを避ける

def route_with_limit(state: AgentState) -> str:
    # 设置最大重试次数，防止无限循环
    if state.get("retry_count", 0) >= 3:
        return END
    
    if needs_retry(state):
        return "retry_node"
    return END

よくある質問（FAQ）

Q1：ノードの戻り値の形式が正しくない場合はどうすればよいですか？

# 错误：直接修改 state 对象
def bad_node(state):
    state["messages"].append(...)  # 不要直接修改
    return state

# 正确：返回需要更新的字段
def good_node(state):
    return {"messages": [new_message]}  # 只返回变更字段

Q2：ノード間で一時データを渡すにはどうすればよいですか？

一時データを State 定義に追加するか、アンダースコアプレフィックス規約を使用して内部フィールドとして扱います：

from typing import TypedDict

class PublicState(TypedDict):
    messages: list  # 对外暴露

class PrivateState(TypedDict):
    messages: list
    _internal_cache: dict  # 以下划线开头约定为内部使用

Q3：ノード実行プロセスをデバッグするには？

# 使用 stream 模式观察每个节点的输出
for event in graph.stream(initial_state, stream_mode="updates"):
    for node_name, state_update in event.items():
        print(f"\n[节点: {node_name}]")
        print(f"更新: {state_update}")

Q4：StateGraph と MessageGraph の違いは？

MessageGraphは初期バージョンのAPIであり、機能が比較的制限されています。現在は統一して以下を使用することを推奨しますStateGraph。これはより柔軟で、機能がより充実しています。メッセージを処理する必要がある場合は、以下を使用してくださいStateGraph(MessagesState)、または、以下を含むカスタムmessagesフィールドの State を使用してください。

まとめ

本記事では、LangGraph フレームワークの核心概念と実践的な応用について体系的に紹介します。LangGraph はグラフ構造のワークフローオーケストレーションを通じて、開発者がループ、分岐、状態永続化の機能を備えた複雑な AI Agent を構築できるようにします。従来の線形 LLM Chain と比較して、LangGraph は多段階の推論、人機協働、マルチ Agent 連携が必要なシナリオにおいて顕著な利点があります。

推奨学習パス