LangGraph

LangGraph のエキスパート - ステートフルで複数のアクターを持つAIアプリケーションを構築するための本番レベルのフレームワーク。グラフ構築、状態管理、サイクルと分岐、チェックポインターによる永続性、ヒューマンインザループパターン、ReActエージェントパターンをカバーしています。LinkedIn、Uber、400以上の企業で本番環境で使用されています。これは LangChain が推奨するエージェント構築アプローチです。

Role: LangGraph Agent Architect

LangGraph を使って本番レベルの AI エージェントを構築するエキスパートです。エージェントは明確な構造が必要であり、グラフはフローを可視化しデバッグ可能にします。状態を慎重に設計し、リデューサーを適切に使用し、本番環境での永続性を常に考慮します。サイクルが必要な場合と無限ループを防ぐ方法を理解しています。

Expertise

• グラフトポロジーの設計
• 状態スキーマパターン
• 条件付き分岐
• 永続性戦略
• ヒューマンインザループ
• ツール統合
• エラーハンドリングとリカバリー

Capabilities

• グラフ構築（StateGraph）
• 状態管理とリデューサー
• ノードとエッジの定義
• 条件付きルーティング
• チェックポインターと永続性
• ヒューマンインザループパターン
• ツール統合
• ストリーミングと非同期実行

Prerequisites

• 0: Python プロフィシエンシー
• 1: LLM API の基礎
• 2: 非同期プログラミングの概念
• 3: グラフ理論の基礎
• 必須スキル: Python 3.9+、langgraph パッケージ、LLM API アクセス（OpenAI、Anthropic など）、グラフの概念の理解

Scope

• 0: Python のみ（TypeScript は初期段階）
• 1: グラフ概念の学習曲線
• 2: 状態管理の複雑さ
• 3: デバッグは困難な場合がある

Ecosystem

Primary

• LangGraph
• LangChain
• LangSmith（可観測性）

Common_integrations

• OpenAI / Anthropic / Google
• Tavily（検索）
• SQLite / PostgreSQL（永続性）
• Redis（状態ストア）

Platforms

• Python アプリケーション
• FastAPI / Flask バックエンド
• クラウドデプロイ

Patterns

Basic Agent Graph

シンプルな ReAct スタイルのツール付きエージェント

使用場面: 単一エージェントでツール呼び出しを行う場合

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. Define State
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages reducer appends, doesn't overwrite

# 2. Define Tools
@tool
def search(query: str) -> str:
    """Search the web for information."""
    # Implementation here
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

tools = [search, calculator]

# 3. Create LLM with tools
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 4. Define Nodes
def agent(state: AgentState) -> dict:
    """The agent node - calls LLM."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Tool node handles tool execution
tool_node = ToolNode(tools)

# 5. Define Routing
def should_continue(state: AgentState) -> str:
    """Route based on whether tools were called."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 6. Build Graph
graph = StateGraph(AgentState)

# Add nodes
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

# Add edges
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # Loop back

# Compile
app = graph.compile()

# 7. Run
result = app.invoke({
    "messages": [("user", "What is 25 * 4?")]
})

State with Reducers

カスタムリデューサーを使用した複雑な状態管理

使用場面: 複数エージェントが共有状態を更新する場合

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph

# Custom reducer for merging dictionaries
def merge_dicts(left: dict, right: dict) -> dict:
    return {**left, **right}

# State with multiple reducers
class ResearchState(TypedDict):
    # Messages append (don't overwrite)
    messages: Annotated[list, add_messages]

    # Research findings merge
    findings: Annotated[dict, merge_dicts]

    # Sources accumulate
    sources: Annotated[list[str], add]

    # Current step (overwrites - no reducer)
    current_step: str

    # Error count (custom reducer)
    errors: Annotated[int, lambda a, b: a + b]

# Nodes return partial state updates
def researcher(state: ResearchState) -> dict:
    # Only return fields being updated
    return {
        "findings": {"topic_a": "New finding"},
        "sources": ["source1.com"],
        "current_step": "researching"
    }

def writer(state: ResearchState) -> dict:
    # Access accumulated state
    all_findings = state["findings"]
    all_sources = state["sources"]

    return {
        "messages": [("assistant", f"Report based on {len(all_sources)} sources")],
        "current_step": "writing"
    }

# Build graph
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
# ... add edges

Conditional Branching

状態に基づいてさまざまなパスにルートする

使用場面: 複数の可能なワークフローがある場合

from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    query: str
    query_type: str
    result: str

def classifier(state: RouterState) -> dict:
    """Classify the query type."""
    query = state["query"].lower()
    if "code" in query or "program" in query:
        return {"query_type": "coding"}
    elif "search" in query or "find" in query:
        return {"query_type": "search"}
    else:
        return {"query_type": "chat"}

def coding_agent(state: RouterState) -> dict:
    return {"result": "Here's your code..."}

def search_agent(state: RouterState) -> dict:
    return {"result": "Search results..."}

def chat_agent(state: RouterState) -> dict:
    return {"result": "Let me help..."}

# Routing function
def route_query(state: RouterState) -> str:
    """Route to appropriate agent."""
    query_type = state["query_type"]
    return query_type  # Returns node name

# Build graph
graph = StateGraph(RouterState)

graph.add_node("classifier", classifier)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)

graph.add_edge(START, "classifier")

# Conditional edges from classifier
graph.add_conditional_edges(
    "classifier",
    route_query,
    {
        "coding": "coding",
        "search": "search",
        "chat": "chat"
    }
)

# All agents lead to END
graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)

app = graph.compile()

Persistence with Checkpointer

エージェント状態を保存して再開する

使用場面: マルチターン会話、長時間実行エージェント

from langgraph.graph import StateGraph
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# SQLite for development
memory = SqliteSaver.from_conn_string(":memory:")
# Or persistent file
memory = SqliteSaver.from_conn_string("agent_state.db")

# PostgreSQL for production
# memory = PostgresSaver.from_conn_string(DATABASE_URL)

# Compile with checkpointer
app = graph.compile(checkpointer=memory)

# Run with thread_id for conversation continuity
config = {"configurable": {"thread_id": "user-123-session-1"}}

# First message
result1 = app.invoke(
    {"messages": [("user", "My name is Alice")]},
    config=config
)

# Second message - agent remembers context
result2 = app.invoke(
    {"messages": [("user", "What's my name?")]},
    config=config
)
# Agent knows name is Alice!

# Get conversation history
state = app.get_state(config)
print(state.values["messages"])

# List all checkpoints
for checkpoint in app.get_state_history(config):
    print(checkpoint.config, checkpoint.values)

Human-in-the-Loop

アクション前に人間の承認を待つ

使用場面: 機密性の高い操作、実行前のレビューが必要な場合

from langgraph.graph import StateGraph, START, END

class ApprovalState(TypedDict):
    messages: Annotated[list, add_messages]
    pending_action: dict | None
    approved: bool

def agent(state: ApprovalState) -> dict:
    # Agent decides on action
    action = {"type": "send_email", "to": "user@example.com"}
    return {
        "pending_action": action,
        "messages": [("assistant", f"I want to: {action}")]
    }

def execute_action(state: ApprovalState) -> dict:
    action = state["pending_action"]
    # Execute the approved action
    result = f"Executed: {action['type']}"
    return {
        "messages": [("assistant", result)],
        "pending_action": None
    }

def should_execute(state: ApprovalState) -> str:
    if state.get("approved"):
        return "execute"
    return END  # Wait for approval

# Build graph
graph = StateGraph(ApprovalState)
graph.add_node("agent", agent)
graph.add_node("execute", execute_action)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_execute, ["execute", END])
graph.add_edge("execute", END)

# Compile with interrupt_before for human review
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["execute"]  # Pause before execution
)

# Run until interrupt
config = {"configurable": {"thread_id": "approval-flow"}}
result = app.invoke({"messages": [("user", "Send report")]}, config)

# Agent paused - get pending state
state = app.get_state(config)
pending = state.values["pending_action"]
print(f"Pending: {pending}")  # Human reviews

# Human approves - update state and continue
app.update_state(config, {"approved": True})
result = app.invoke(None, config)  # Resume

Parallel Execution (Map-Reduce)

複数のブランチを並列実行する

使用場面: 並列リサーチ、バッチ処理

from langgraph.graph import StateGraph, START, END, Send
from langgraph.constants import Send

class ParallelState(TypedDict):
    topics: list[str]
    results: Annotated[list[str], add]
    summary: str

def research_topic(state: dict) -> dict:
    """Research a single topic."""
    topic = state["topic"]
    result = f"Research on {topic}..."
    return {"results": [result]}

def summarize(state: ParallelState) -> dict:
    """Combine all research results."""
    all_results = state["results"]
    summary = f"Summary of {len(all_results)} topics"
    return {"summary": summary}

def fanout_topics(state: ParallelState) -> list[Send]:
    """Create parallel tasks for each topic."""
    return [
        Send("research", {"topic": topic})
        for topic in state["topics"]
    ]

# Build graph
graph = StateGraph(ParallelState)
graph.add_node("research", research_topic)
graph.add_node("summarize", summarize)

# Fan out to parallel research
graph.add_conditional_edges(START, fanout_topics, ["research"])
# All research nodes lead to summarize
graph.add_edge("research", "summarize")
graph.add_edge("summarize", END)

app = graph.compile()

result = app.invoke({
    "topics": ["AI", "Climate", "Space"],
    "results": []
})
# Research runs in parallel, then summarizes

Collaboration

Delegation Triggers

• crewai|role-based|crew -> crewai（ロールベースのマルチエージェントアプローチが必要な場合）
• observability|tracing|langsmith -> langfuse（LLM 可観測性が必要な場合）
• structured output|json schema -> structured-output（構造化された LLM レスポンスが必要な場合）
• evaluate|benchmark|test agent -> agent-evaluation（エージェントパフォーマンス評価が必要な場合）

Production Agent Stack

スキル: langgraph, langfuse, structured-output

ワークフロー:

1. LangGraph でエージェントグラフを設計
2. ツールレスポンス用の構造化出力を追加
3. 可観測性のために Langfuse を統合
4. 本番環境でテストおよび監視

Multi-Agent System

スキル: langgraph, crewai, agent-communication

ワークフロー:

1. エージェントロールを設計（CrewAI パターン）
2. LangGraph のサブグラフとして実装
3. エージェント間通信を追加
4. スーパーバイザーパターンで調整

Evaluated Agent

スキル: langgraph, agent-evaluation, langfuse

ワークフロー:

1. LangGraph でエージェントを構築
2. 評価スイートを作成
3. Langfuse で監視
4. メトリクスに基づいて反復

Related Skills

相性が良いスキル: crewai, autonomous-agents, langfuse, structured-output

When to Use

• ユーザーが langgraph について言及または暗示
• ユーザーが langchain agent について言及または暗示
• ユーザーが stateful agent について言及または暗示
• ユーザーが agent graph について言及または暗示
• ユーザーが react agent について言及または暗示
• ユーザーが agent workflow について言及または暗示
• ユーザーが multi-step agent について言及または暗示

Limitations

• 上記で説明されたスコープに明確に合致する場合にのみこのスキルを使用してください。
• 出力を環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、セキュリティ境界、または成功基準が不足している場合は、停止して明確化を求めてください。