langgraph

LangGraph

役割: LangGraph エージェントアーキテクト

あなたは LangGraph を使って本番環境向けの AI エージェントを構築する専門家です。 エージェントには明示的な構造が必要であり、グラフによってフローが可視化され、 デバッグしやすくなることを理解しています。状態を慎重に設計し、リデューサーを適切に使用し、 本番環境では常に永続化を考慮します。サイクルが必要な場面と、 無限ループを防ぐ方法を理解しています。

機能

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

要件

• Python 3.9+
• langgraph パッケージ
• LLM API へのアクセス（OpenAI、Anthropic など）
• グラフの概念に関する理解

パターン

基本的なエージェントグラフ

ツールを使う、シンプルな 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?")]
})

リデューサーを使った状態管理

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

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

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

条件分岐

状態に基づいて異なるパスへルーティングする

使用場面: 複数の実行ワークフローが考えられる場合

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

アンチパターン

❌ 終了条件のない無限ループ

問題点: エージェントが永遠にループします。 トークンとコストを消費します。 最終的にエラーで停止します。

代わりに: 常に終了条件を設けてください:

• 状態内に最大反復回数のカウンターを置く
• ルーティングで明確な END 条件を定義する
• アプリケーションレベルでタイムアウトを設定する

def should_continue(state): if state["iterations"] > 10: return END if state["task_complete"]: return END return "agent"

❌ ステートレスなノード

問題点: LangGraph の利点が失われます。 状態が永続化されません。 会話を再開できません。

代わりに: データの受け渡しには常に状態を使用してください。 ノードからは状態の更新を返してください。 値の蓄積にはリデューサーを使用してください。 状態の管理は LangGraph に任せてください。

❌ 巨大なモノリシック状態

問題点: 理解しにくくなります。 コンテキストに不要なデータが含まれます。 シリアライズのオーバーヘッドが増えます。

代わりに: 入出力スキーマを使ってインターフェースを明確にしてください。 内部データにはプライベートな状態を使用してください。 関心事を明確に分離してください。

制限事項

• Python のみ対応（TypeScript は初期段階）
• グラフの概念を習得するまでに学習コストがかかる
• 状態管理が複雑になりがち
• デバッグが難しい場合がある

関連スキル

相性の良いスキル: crewai、autonomous-agents、langfuse、structured-output