LangGraphとは？StateGraphの書き方・人間承認・永続化【2026年】

# 1. Stateを定義する（グラフ全体で共有されるデータ）
class State(TypedDict):
question: str
answer: str

# 2. ノードを関数として書く（Stateを受け取り、更新分を返す）
def think(state: State) -> dict:
return {“answer”: f”「{state[‘question’]}」について考えました”}

def polish(state: State) -> dict:
return {“answer”: state[“answer”] + “（整形済み）”}

# 3. グラフを組み立てる
builder = StateGraph(State)
builder.add_node(“think”, think)
builder.add_node(“polish”, polish)

builder.add_edge(START, “think”)
builder.add_edge(“think”, “polish”)
builder.add_edge(“polish”, END)

graph = builder.compile()

# 4. 実行
result = graph.invoke({“question”: “LangGraphとは？”})
print(result[“answer”])

class State(TypedDict):
question: str
# operator.add で、返り値がリストに追加されていく
logs: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
return {“logs”: [“step_a を実行”]}

def step_b(state: State) -> dict:
return {“logs”: [“step_b を実行”]}
# 実行後、state[“logs”] は [“step_a を実行”, “step_b を実行”] になる

# 判定結果に応じて行き先を返す関数
def route(state: State) -> str:
if state[“is_relevant”]:
return “generate” # 十分 → 回答生成へ
return “web_search” # 不十分 → Web検索へ

builder.add_node(“grade”, grade)
builder.add_node(“generate”, generate)
builder.add_node(“web_search”, web_search)

# 条件付きエッジ：routeの返り値でノードを切り替える
builder.add_conditional_edges(
“grade”,
route,
{
“generate”: “generate”,
“web_search”: “web_search”,
},
)

# Web検索したら、もう一度判定に戻す（ループ）
builder.add_edge(“web_search”, “grade”)

def send_email(state: State) -> dict:
# 送信前に人間の承認を要求して停止する
approval = interrupt({
“action”: “メール送信”,
“to”: state[“recipient”],
“body”: state[“draft”],
})

if approval == “approve”:
# 実際の送信処理
return {“status”: “送信しました”}
return {“status”: “キャンセルしました”}

# Checkpointerがないと中断・再開できない点に注意
graph = builder.compile(checkpointer=MemorySaver())

config = {“configurable”: {“thread_id”: “user-123”}}

# 1回目の実行 → interrupt で停止する
graph.invoke({“recipient”: “client@example.com”}, config=config)

# 人間が内容を確認した後、承認して再開する
graph.invoke(Command(resume=”approve”), config=config)

DB_URI = “postgresql://user:password@localhost:5432/langgraph”

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # 初回のみ：必要なテーブルを作成
graph = builder.compile(checkpointer=checkpointer)

config = {“configurable”: {“thread_id”: “session-001”}}
graph.invoke({“question”: “…”}, config=config)

# 過去の実行履歴を取得できる
for snapshot in graph.get_state_history(config):
print(snapshot.values)

class RagState(TypedDict):
question: str
documents: list
answer: str
retry_count: Annotated[int, operator.add]

def retrieve(state: RagState) -> dict:
docs = retriever.invoke(state[“question”])
return {“documents”: docs}

def grade(state: RagState) -> dict:
# LLMに「この文書で質問に答えられるか」をYes/Noで判定させる
context = “\n”.join(d.page_content for d in state[“documents”])
verdict = model.invoke(
f”以下の資料で質問に答えられますか。yes か no のみで答えてください。\n”
f”【質問】{state[‘question’]}\n【資料】{context[:2000]}”
).content.strip().lower()
return {“is_relevant”: verdict.startswith(“yes”)}

def rewrite(state: RagState) -> dict:
# 検索にヒットしやすい表現へ質問を言い換える
new_q = model.invoke(
f”次の質問を、社内文書の検索にヒットしやすい表現へ言い換えてください。”
f”言い換えた質問のみを出力してください。\n{state[‘question’]}”
).content
return {“question”: new_q, “retry_count”: 1}

def generate(state: RagState) -> dict:
context = “\n”.join(d.page_content for d in state[“documents”])
ans = model.invoke(
f”以下の資料のみを根拠に答えてください。記載がなければ「資料に記載がありません」と答えてください。\n”
f”【資料】{context}\n【質問】{state[‘question’]}”
).content
return {“answer”: ans}

def route(state: RagState) -> str:
if state.get(“is_relevant”):
return “generate”
if state.get(“retry_count”, 0) >= 3: # 無限ループ防止
return “generate”
return “rewrite”

builder = StateGraph(RagState)
builder.add_node(“retrieve”, retrieve)
builder.add_node(“grade”, grade)
builder.add_node(“rewrite”, rewrite)
builder.add_node(“generate”, generate)

builder.add_edge(START, “retrieve”)
builder.add_edge(“retrieve”, “grade”)
builder.add_conditional_edges(“grade”, route,
{“generate”: “generate”, “rewrite”: “rewrite”})
builder.add_edge(“rewrite”, “retrieve”) # 言い換えて再検索
builder.add_edge(“generate”, END)

graph = builder.compile()

# プロジェクトルートで実行するとStudioが起動する
langgraph dev