LangGraphとは？エージェントの使い方とワークフロー構築法をハンズオンで解説

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-6-luna",
    reasoning_effort="none",
    max_tokens=500,
    max_retries=0,
    timeout=60,
)

from langgraph.graph import START, END, StateGraph
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict

class AnswerState(TypedDict):
    question: str
    answer: str

def build_basic(llm):
    def ask_llm(state: AnswerState):
        reply = llm.invoke([
            SystemMessage(content="日本語で120文字程度で回答してください。知らない事実は推測しないでください。"),
            HumanMessage(content=state["question"]),
        ])
        return {"answer": reply.content}

    builder = StateGraph(AnswerState)
    builder.add_node("ask_llm", ask_llm)
    builder.add_edge(START, "ask_llm")
    builder.add_edge("ask_llm", END)
    return builder.compile()

app = build_basic(llm)
result = app.invoke({"question": "What is LangGraph?"})
print(result["answer"])

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

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

def build_agent(llm):
    return create_agent(
        model=llm,
        tools=[multiply],
        system_prompt="計算には必ずmultiplyツールを使い、日本語で短く答えてください。",
    )

agent = build_agent(llm)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "17と23を掛け算してください。"}]},
    {"recursion_limit": 8},
)
for message in result["messages"]:
    print(message.type, message.content)

{
  "dependencies": ["."],
  "graphs": {"approval_demo": "./studio_graph.py:graph"},
  "env": {"LANGSMITH_TRACING": "false"}
}

"""Studio-only control demo, fixed draft: no model, API key, or paid call."""
from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph
from langgraph.types import Command, interrupt


class StudioState(TypedDict):
    request: str
    draft: str
    status: str


def draft(state: StudioState):
    return {"draft": "【固定文の操作確認】" + state["request"], "status": "pending"}


def review(state: StudioState) -> Command[Literal["approved", "returned"]]:
    answer = interrupt({"draft": state["draft"], "choices": ["approve", "revise"]})
    if answer not in {"approve", "revise"}:
        raise ValueError("approveまたはreviseを指定してください")
    return Command(goto="approved" if answer == "approve" else "returned")


def approved(state: StudioState):
    return {"status": "approved_test_only"}


def returned(state: StudioState):
    return {"status": "returned_test_only"}


builder = StateGraph(StudioState)
builder.add_node("draft", draft)
builder.add_node("review", review)
builder.add_node("approved", approved)
builder.add_node("returned", returned)
builder.add_edge(START, "draft")
builder.add_edge("draft", "review")
builder.add_edge("approved", END)
builder.add_edge("returned", END)
# Agent Server supplies its own checkpointer; do not attach an in-memory saver here.
graph = builder.compile()

[project]
name = "langgraph-studio-example"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["langgraph==1.2.15"]

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[tool.setuptools]
py-modules = ["studio_graph"]

from langgraph.graph import START, END, StateGraph
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict
from typing import Annotated, Literal
from langgraph.graph.message import add_messages

class LoopState(TypedDict):
    messages: Annotated[list, add_messages]
    count: int
    questions: list[str]

def build_loop(llm, checkpointer=None):
    def user_input(state: LoopState):
        return {"messages": [HumanMessage(content=state["questions"][state["count"]])]}

    def ask_llm(state: LoopState):
        reply = llm.invoke([
            SystemMessage(content="会話履歴を参照し、日本語で120文字程度で回答してください。知らない事実は推測しないでください。"),
            *state["messages"],
        ])
        return {"messages": [reply], "count": state["count"] + 1}

    def route(state: LoopState) -> Literal["continue", "end"]:
        return "end" if state["count"] >= len(state["questions"]) else "continue"

    builder = StateGraph(LoopState)
    builder.add_node("user_input", user_input)
    builder.add_node("ask_llm", ask_llm)
    builder.add_edge(START, "user_input")
    builder.add_edge("user_input", "ask_llm")
    builder.add_conditional_edges("ask_llm", route, {"continue": "user_input", "end": END})
    return builder.compile(checkpointer=checkpointer)

from langgraph.checkpoint.memory import InMemorySaver

app = build_loop(llm, InMemorySaver())
config = {"configurable": {"thread_id": "loop-example-1"}, "recursion_limit": 10}
initial = {
    "messages": [], "count": 0,
    "questions": ["LangGraphって何？", "何ができるの？", "活用事例は？"],
}
for update in app.stream(initial, config, stream_mode="updates"):
    print(update)
print(app.get_state(config).values["count"])

たとえば、次のような用途があります。
問い合わせ対応：質問を分類し、必要に応じてFAQ検索や担当者への引き継ぎを行う
社内情報検索：検索結果を評価し、不足があれば再検索して回答する
文書処理：要約・分類・内容確認を段階的に進める
複数エージェントの連携：調査役とレビュー役などを協調させる
こうした流れを、状態を保ちながら制御できます。

from langgraph.graph import START, END, StateGraph
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict
from typing import Literal

class DiscussionState(TypedDict):
    topic: str
    history: list[dict]
    round: int
    summary: str

ROLES = {
    "creative": "クリエイティブ思考者として、新しい案を一つ提示する。",
    "pragmatic": "実用主義者として、費用や実行条件から直前の案を検討する。",
    "critical": "批判的思考者として、直前の意見のリスクや未確認点を指摘する。",
}

def build_discussion(llm):
    def speaker(name):
        def node(state: DiscussionState):
            history = "\n".join(f"{x['role']}: {x['text']}" for x in state["history"])
            reply = llm.invoke([
                SystemMessage(content=ROLES[name] + "過去の発言を踏まえ、日本語で120文字程度。経済予測を事実として断定しない。"),
                HumanMessage(content=f"テーマ: {state['topic']}\nラウンド: {state['round']}\nこれまでの発言:\n{history}"),
            ])
            return {"history": [*state["history"], {"round": state["round"], "role": name, "text": reply.content}]}
        return node

    def next_round(state: DiscussionState):
        return {"round": state["round"] + 1}

    def route(state: DiscussionState) -> Literal["next", "finish"]:
        return "finish" if state["round"] >= 3 else "next"

    def supervisor(state: DiscussionState):
        history = "\n".join(f"{x['role']}: {x['text']}" for x in state["history"])
        reply = llm.invoke([
            SystemMessage(content="スーパーバイザーとして意見を整理する。合意点と未確認点を日本語で200文字程度。事実を追加しない。"),
            HumanMessage(content=f"テーマ: {state['topic']}\n発言:\n{history}"),
        ])
        return {"summary": reply.content}

    builder = StateGraph(DiscussionState)
    for name in ROLES:
        builder.add_node(name, speaker(name))
    builder.add_node("next_round", next_round)
    builder.add_node("supervisor", supervisor)
    builder.add_edge(START, "creative")
    builder.add_edge("creative", "pragmatic")
    builder.add_edge("pragmatic", "critical")
    builder.add_conditional_edges("critical", route, {"next": "next_round", "finish": "supervisor"})
    builder.add_edge("next_round", "creative")
    builder.add_edge("supervisor", END)
    return builder.compile()

app = build_discussion(llm)
initial = {
    "topic": "10年後20年後の日本の経済成長について。",
    "history": [], "round": 1, "summary": "",
}
for update in app.stream(initial, {"recursion_limit": 18}, stream_mode="updates"):
    print(update)

from langgraph.graph import START, END, StateGraph
from langchain_core.messages import HumanMessage, SystemMessage
from typing_extensions import TypedDict
from typing import Literal
from langgraph.types import Command, interrupt

class ApprovalState(TypedDict):
    request: str
    draft: str
    feedback: str
    status: str
    revisions: int

def build_approval(llm, checkpointer=None):
    def draft(state: ApprovalState):
        reply = llm.invoke([
            SystemMessage(content="架空の購買申請の下書きを日本語120文字程度で作成。指定された品目・数量を守り、未指定の金額や承認者は作らない。"),
            HumanMessage(content=f"申請: {state['request']}\n修正指示: {state.get('feedback', '')}"),
        ])
        return {"draft": reply.content, "status": "pending"}

    def review(state: ApprovalState) -> Command[Literal["approved", "revise"]]:
        decision = interrupt({"draft": state["draft"], "choices": ["approve", "revise"]})
        if not isinstance(decision, dict) or decision.get("action") not in {"approve", "revise"}:
            raise ValueError("actionはapproveまたはreviseを指定してください")
        if decision["action"] == "approve":
            return Command(update={"status": "approved"}, goto="approved")
        if state.get("revisions", 0) >= 1:
            raise ValueError("この検証では差し戻しは1回までです")
        feedback = decision.get("feedback", "")
        if not isinstance(feedback, str) or not feedback.strip():
            raise ValueError("差し戻しにはfeedbackが必要です")
        return Command(update={"feedback": feedback, "revisions": state.get("revisions", 0) + 1}, goto="revise")

    def revise(state: ApprovalState):
        return {"status": "revising"}

    def approved(state: ApprovalState):
        # 本物の発注・基幹システムへの書き込みは行わない。
        return {"status": "approved_test_only"}

    builder = StateGraph(ApprovalState)
    builder.add_node("draft", draft)
    builder.add_node("review", review)
    builder.add_node("revise", revise)
    builder.add_node("approved", approved)
    builder.add_edge(START, "draft")
    builder.add_edge("draft", "review")
    builder.add_edge("revise", "draft")
    builder.add_edge("approved", END)
    return builder.compile(checkpointer=checkpointer)

from langgraph.checkpoint.sqlite import SqliteSaver

config = {"configurable": {"thread_id": "approval-example-1"}, "recursion_limit": 12}
initial = {
    "request": "検証用の購買申請。USB-Cケーブルを2本購入したい。金額は未定。",
    "draft": "", "feedback": "", "status": "new", "revisions": 0,
}
with SqliteSaver.from_conn_string("approval-example.sqlite") as saver:
    app = build_approval(llm, saver)
    paused = app.invoke(initial, config)
    print(paused["__interrupt__"])

# 別の接続で、同じthread_idの処理を再開する。
with SqliteSaver.from_conn_string("approval-example.sqlite") as saver:
    app = build_approval(llm, saver)
    print(app.get_state(config).values["status"])
    revised = app.invoke(Command(resume={
        "action": "revise", "feedback": "用途を『会議室の接続用』と追記してください。",
    }), config)
    print(revised["__interrupt__"])
    approved = app.invoke(Command(resume={"action": "approve"}), config)
    print(approved["status"])