langchain-agents

LangGraph で、基本的なプリミティブから高度なコンテキスト管理まで、本番環境に対応したエージェントを構築します。

<quick_start> 重要： 最新の抽象化を使用してください。create_sql_agent、create_tool_calling_agent、create_react_agent などの古いヘルパーは非推奨です。

シンプルなツール呼び出しエージェントが必要ですか？ → create_agent

from langchain.agents import create_agent
graph = create_agent(model="anthropic:claude-sonnet-4-5", tools=[search], system_prompt="...")

用途： 基本的な ReAct ループ、ツール呼び出しエージェント、シンプルな Q&A ボット。

計画機能＋ファイルシステム＋サブエージェントが必要ですか？ → create_deep_agent

from deepagents import create_deep_agent
agent = create_deep_agent(model=model, tools=tools, backend=FilesystemBackend())

用途： リサーチエージェント、複雑なワークフロー、複数ステップの計画。

カスタム制御フロー／マルチエージェント／高度なコンテキストが必要ですか？ → LangGraph（以下を参照） 用途： カスタムルーティングロジック、スーパーバイザーパターン、専用の状態管理、標準外のワークフロー。

シンプルに始めましょう： まず基本的な ReAct ループで構築してください。ユースケースで必要な場合にのみ、複雑さを追加します。 </quick_start>

<create_agent>

create_agent の使用（推奨）

from langchain_anthropic import ChatAnthropic
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def my_tool(query: str) -> str:
    """Tool description that the model sees."""
    return perform_operation(query)

model = ChatAnthropic(model="claude-sonnet-4-5")
agent = create_agent(
    model=model,
    tools=[my_tool],
    system_prompt="Your agent behavior and guidelines."
)

result = agent.invoke({"messages": [("user", "Your question")]})

このパターンの用途： SQL エージェント、検索エージェント、Q&A ボット、ツール呼び出しワークフロー。

例：計算エージェント

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        allowed = set('0123456789+-*/(). ')
        if not all(c in allowed for c in expression):
            return "Error: Invalid characters"
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

@tool
def convert_units(value: float, from_unit: str, to_unit: str) -> str:
    """Convert between common units."""
    conversions = {
        ("km", "miles"): 0.621371,
        ("miles", "km"): 1.60934,
    }
    factor = conversions.get((from_unit, to_unit), None)
    return f"{value * factor:.2f} {to_unit}" if factor else "Conversion not supported"

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5"),
    tools=[calculate, convert_units],
    system_prompt="You are a helpful calculator assistant."
)

クイックリファレンス

from langchain.agents import create_agent
agent = create_agent(model=model, tools=[my_tool], system_prompt="...")
result = agent.invoke({"messages": [("user", "question")]})

</create_agent>

### ゼロから作る基本的なエージェント

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

tools = [search_tool]
tool_node = ToolNode(tools)

def agent(state: State):
    return {"messages": [model.bind_tools(tools).invoke(state["messages"])]}

def route(state: State):
    return "tools" if state["messages"][-1].tool_calls else END

workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", route)
workflow.add_edge("tools", "agent")
app = workflow.compile()

ループ： エージェント → ツール → エージェント → END

ToolMessages：重要な詳細

カスタムのツール実行を実装する場合、ツール呼び出しごとに ToolMessage を必ず作成してください。

from langchain_core.messages import ToolMessage

def custom_tool_node(state: State) -> dict:
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        result = execute_tool(tool_call["name"], tool_call["args"])
        # CRITICAL: tool_call_id must match!
        tool_messages.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": tool_messages}

Commands：更新によるルーティング

from langgraph.types import Command
from typing import Literal

def router(state: State) -> Command[Literal["research", "write", END]]:
    if needs_more_context(state):
        return Command(update={"notes": "Starting research"}, goto="research")
    return Command(goto=END)

# Human-in-loop
def ask_user(state: State) -> Command:
    response = interrupt("Please clarify:")
    return Command(update={"messages": [HumanMessage(content=response)]}, goto="continue")

<context_management>

戦略 1：サブエージェントへの委任

パターン： 作業をサブエージェントに委任し、要約のみを返します。

researcher_subgraph = build_researcher_graph().compile()

def main_agent(state: State) -> Command:
    if needs_research(state["messages"][-1]):
        result = researcher_subgraph.invoke({"query": extract_query(state)})
        return Command(
            update={"context": state["context"] + f"\n{result['summary']}"},
            goto="respond"
        )
    return Command(goto="respond")

戦略 2：メッセージを段階的にトリミング

パターン： システムメッセージと最近のコンテキストを保持しながら、古いメッセージを削除します。

def trim_messages(messages: list, max_messages: int = 20) -> list:
    system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    conversation = [m for m in messages if not isinstance(m, SystemMessage)]
    return system_msgs + conversation[-max_messages:]

def agent_with_trimming(state: State) -> dict:
    trimmed = trim_messages(state["messages"], max_messages=15)
    return {"messages": [model.invoke(trimmed)]}

戦略 3：要約による圧縮

パターン： 古いコンテキストを要約し、最近のメッセージはそのまま保持します。

def compress_history(state: State) -> dict:
    messages = state["messages"]
    if len(messages) > 30:
        old, recent = messages[:-10], messages[-10:]
        summary = model.invoke([HumanMessage(content=f"Summarize:\n{format_messages(old)}")])
        return {"messages": [SystemMessage(content=f"Previous:\n{summary.content}")] + recent}
    return {"messages": messages}

</context_management>

<multi_agent>

スーパーバイザーパターン

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str

def supervisor(state: AgentState) -> Command[Literal["billing", "technical", END]]:
    last_msg = state["messages"][-1].content.lower()
    if "invoice" in last_msg or "payment" in last_msg:
        return Command(goto="billing")
    elif "error" in last_msg or "not working" in last_msg:
        return Command(goto="technical")
    return Command(goto=END)

def billing_agent(state: AgentState) -> dict:
    return {"messages": [billing_model.invoke(state["messages"])]}

def technical_agent(state: AgentState) -> dict:
    return {"messages": [tech_model.invoke(state["messages"])]}

workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("billing", billing_agent)
workflow.add_node("technical", technical_agent)
workflow.add_edge(START, "supervisor")
workflow.add_edge("billing", END)
workflow.add_edge("technical", END)
app = workflow.compile()

</multi_agent>

### Checkpointer + Store による永続化

from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = MemorySaver()  # Thread-level state
store = InMemoryStore()       # Cross-thread memory

app = graph.compile(checkpointer=checkpointer, store=store)

app.invoke(
    {"messages": [HumanMessage("Hello")]},
    config={"configurable": {"thread_id": "user-123"}}
)

構造化出力

from pydantic import BaseModel, Field

class ResearchOutput(BaseModel):
    summary: str = Field(description="3-sentence summary")
    sources: list[str] = Field(description="Source URLs")
    confidence: float = Field(description="0-1 confidence score")

model_with_structure = model.with_structured_output(ResearchOutput)

def structured_research(state: State) -> dict:
    result = model_with_structure.invoke(state["messages"])
    return {"research": result.model_dump()}

DeepAgents：必要な機能を標準搭載

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, FilesystemBackend, StoreBackend

backend = CompositeBackend({
    "/workspace/": FilesystemBackend("./workspace"),
    "/memories/": StoreBackend(store)
})

agent = create_deep_agent(
    model=model,
    tools=[search, scrape],
    subagents=[researcher_agent, analyst_agent],
    backend=backend
)

DeepAgents が提供する機能： ファイルシステム（コンテキストファイルを自動生成）、計画（タスクの分解）、サブエージェント（委任）、メモリ（永続化）。

- [LangGraph ドキュメント](https://docs.langchain.com/langgraph) - [create_agent](https://docs.langchain.com/oss/python/langchain/agents) - [DeepAgents](https://docs.langchain.com/oss/python/deepagents/overview) - [LangGraph 101 マルチエージェント](https://github.com/langchain-ai/langgraph-101/blob/main/notebooks/LG201/multi_agent.ipynb) - [ディープリサーチの例](https://github.com/langchain-samples/deep_research_101)