LangGraph入門2026｜Pythonでエージェントを構築する全手順

pip install langgraph langchain-anthropic langchain-community

# Claude (Anthropic) の場合
export ANTHROPIC_API_KEY="sk-ant-..."

# OpenAI の場合
export OPENAI_API_KEY="sk-..."

from typing import TypedDict

class Essay(TypedDict):
    topic: str
    content: str | None
    score: float | None

def write_essay(state: Essay):
    # topic を受け取って content を生成
    return {"content": f"{state['topic']}に関するエッセイ本文..."}

def score_essay(state: Essay):
    # content を受け取って score を付与
    return {"score": 8.5}

from langgraph.graph import StateGraph, START, END

builder = StateGraph(Essay)
builder.add_node(write_essay)
builder.add_node(score_essay)

builder.add_edge(START, "write_essay")
builder.add_edge("write_essay", "score_essay")
builder.add_edge("score_essay", END)

graph = builder.compile()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class Essay(TypedDict):
    topic: str
    content: str | None
    score: float | None


def write_essay(state: Essay):
    return {"content": f"{state['topic']}に関する800字のエッセイ"}


def score_essay(state: Essay):
    word_count = len(state["content"] or "")
    score = min(10.0, word_count / 20)
    return {"score": score}


builder = StateGraph(Essay)
builder.add_node(write_essay)
builder.add_node(score_essay)

builder.add_edge(START, "write_essay")
builder.add_edge("write_essay", "score_essay")
builder.add_edge("score_essay", END)

graph = builder.compile()

# 実行
result = graph.invoke({"topic": "LangGraphの設計思想", "content": None, "score": None})
print(result)
# {'topic': 'LangGraphの設計思想', 'content': 'LangGraphの設計思想に関する800字のエッセイ', 'score': 1.05}

from typing import Literal
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.graph import MessagesState, StateGraph, START, END


# --- 1. ツール定義 ---
@tool
def add(a: int, b: int) -> int:
    """2つの整数を足す"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """2つの整数を掛ける"""
    return a * b

tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}


# --- 2. LLMにツールをバインド ---
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
llm_with_tools = llm.bind_tools(tools)


# --- 3. ノード定義 ---
def llm_call(state: MessagesState):
    """LLMを呼び出し、ツール呼び出しの有無を含む応答を返す"""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}


def tool_executor(state: MessagesState):
    """LLMが要求したツールを実行し、結果をToolMessageで返す"""
    results = []
    for tc in state["messages"][-1].tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}


# --- 4. 条件分岐 ---
def should_continue(state: MessagesState) -> Literal["tool_executor", "__end__"]:
    last = state["messages"][-1]
    return "tool_executor" if last.tool_calls else END


# --- グラフ組み立て ---
builder = StateGraph(MessagesState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_executor", tool_executor)

builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", should_continue, ["tool_executor", END])
builder.add_edge("tool_executor", "llm_call")  # ツール実行後、LLMに戻る

agent = builder.compile()

# 実行
result = agent.invoke({"messages": [HumanMessage(content="3と7を足して、その結果に5を掛けて")]})
print(result["messages"][-1].content)

import sqlite3
from typing import TypedDict
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command, interrupt


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信する（送信前に人間の承認が必要）"""

    # ここでグラフが一時停止する
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "このメールを送信しますか？",
    })

    if response.get("action") == "approve":
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        return f"送信完了: {final_to} / {final_subject}"

    return "送信キャンセル"


llm = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email])


def agent_node(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}


def tool_node(state: MessagesState):
    results = []
    for tc in state["messages"][-1].tool_calls:
        result = send_email.invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}


def should_continue(state: MessagesState):
    last = state["messages"][-1]
    return "tool_node" if getattr(last, "tool_calls", None) else END


builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tool_node", END])
builder.add_edge("tool_node", "agent")

# Checkpointer付きでコンパイル（これがないとinterruptが使えない）
checkpointer = SqliteSaver(sqlite3.connect("email_approval.db"))
agent = builder.compile(checkpointer=checkpointer)


# --- 実行 ---
config = {"configurable": {"thread_id": "email-001"}}

# Step 1: エージェントが動く → send_email内のinterruptで停止
result = agent.invoke(
    {"messages": [HumanMessage(content="田中さん（[email protected]）に会議の件でメールして")]},
    config=config,
)
print("承認待ち:", result["__interrupt__"])

# Step 2: 人間が内容を確認し、承認（件名を修正して再開）
resumed = agent.invoke(
    Command(resume={"action": "approve", "subject": "【4/10】定例会議のご案内"}),
    config=config,
)
print(resumed["messages"][-1].content)

from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, BaseMessage
from langchain_anthropic import ChatAnthropic
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task


@tool
def add(a: int, b: int) -> int:
    """2つの整数を足す"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """2つの整数を掛ける"""
    return a * b

tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0).bind_tools(tools)


@task
def call_llm(messages: list[BaseMessage]):
    return llm.invoke(messages)


@task
def call_tool(tool_call):
    return tools_by_name[tool_call["name"]].invoke(tool_call)


@entrypoint()
def agent(messages: list[BaseMessage]):
    llm_response = call_llm(messages).result()

    while True:
        if not llm_response.tool_calls:
            break
        # ツール呼び出しを並列実行
        tool_futures = [call_tool(tc) for tc in llm_response.tool_calls]
        tool_results = [f.result() for f in tool_futures]
        messages = add_messages(messages, [llm_response, *tool_results])
        llm_response = call_llm(messages).result()

    return add_messages(messages, llm_response)


# 実行
result = agent.invoke([HumanMessage(content="3と7を足して、その結果に5を掛けて")])
print(result[-1].content)