この記事の対象読者
- Pythonの基本文法（関数、クラス、デコレータ）を理解している方
- LLM（大規模言語モデル）を使ったアプリ開発に興味がある方
- AIエージェントを使って複雑なクリエイティブワークフローを構築したい方
この記事で得られること
- LangGraphの核心概念と他のフレームワークとの違いの理解
- クリエイティブエージェントの設計パターンと実装方法
- 開発・本番・テスト環境向けの設定テンプレート
- 実際に動作するマルチエージェントワークフローのコード
この記事で扱わないこと
- LLM API（OpenAI, Claude等）の基本的な使い方
- Pythonの環境構築方法
- LangChainの詳細な解説（必要最低限のみ触れます）
1. LangGraphとの出会い
「このAIワークフロー、なんか思い通りに動かないんだよな...」
CrewAIで簡単なマルチエージェントシステムを組んでいたとき、私はそう感じた。アイデア出し→執筆→編集→校正という流れを作りたかっただけなのに、状態管理がブラックボックスすぎて、どこで何が起きているのかさっぱりわからない。
そんなとき出会ったのがLangGraphだった。
LangGraphは、LangChain社が開発した「グラフベースのエージェントオーケストレーションフレームワーク」だ。人間で言えば、プロジェクトマネージャーが各メンバーの作業状況を常に把握しながら、次に誰が何をすべきかを指示する——そんなイメージだ。
最初にLangGraphのグラフ構造を見たとき、「ああ、これが欲しかった」と膝を打った。状態が可視化され、フローが明示的で、どこで何が起きているか一目瞭然。今回は、その感動をお伝えしたい。
ここまでで、LangGraphがどんなものか、なんとなくイメージできただろうか。次は、この技術で使われる基本用語を押さえていこう。
2. 前提知識の確認
本題に入る前に、この記事で使う用語を整理しておく。
2.1 AIエージェントとは
AIエージェントとは、LLM（Large Language Model、大規模言語モデル）を使って自律的にタスクを実行するプログラムのことだ。単なるチャットボットとの違いは、「ツール」を使えること。Web検索したり、コードを実行したり、外部APIを叩いたりできる。
2.2 オーケストレーションとは
オーケストレーションとは、複数のエージェントやタスクを協調させて動かすことを指す。オーケストラの指揮者が各楽器を統率するように、AIシステムでも誰かが「次はお前の番だ」と指示を出す必要がある。
2.3 状態管理（ステートマネジメント）とは
状態管理とは、ワークフローの「今どこにいるか」「これまで何をしたか」を追跡する仕組みだ。人間のプロジェクトでも、タスクの進捗を管理するのと同じ発想だ。
2.4 MCP（Model Context Protocol）とは
MCPとは、Anthropic社が提唱し、現在はLinux Foundationが管理するLLMとツール間の標準接続プロトコルだ。月間9700万以上のダウンロードがあり、Claude、OpenAI、Googleなど主要プレイヤーが採用している業界標準となっている。
2.5 HITL（Human-in-the-Loop）とは
HITLとは、AIワークフローに人間の確認・介入を組み込む設計のことだ。「ここで一旦止めて、人間がチェックしてから次に進む」というポイントを設定できる。
これらの用語が押さえられたら、次に進もう。
3. LangGraphが生まれた背景
3.1 LangChainの限界とLangGraphの誕生
LangGraphは2024年初頭、LangChain社によって独立したライブラリとしてリリースされた。LangChainは直線的なワークフロー（DAG: Directed Acyclic Graph、有向非巡回グラフ）には強かったが、「ループ」や「条件分岐」を含む複雑なエージェント動作には限界があった。
現実のエージェントは、単純に「A→B→C」と進むわけではない。「Aをやって、結果が悪ければAに戻る」「BとCを並行で走らせる」といった複雑なフローが必要になる。LangGraphはこの課題を解決するために生まれた。
3.2 2025年現在の重要性
2025年の今、AIエージェント開発は「研究」から「本番運用」のフェーズに移行している。Uber、LinkedIn、Replit、Elastic、Klarnaといった大企業がLangGraphを本番環境で採用している。
GitHub Stars 21,000以上、月間数百万ダウンロードという数字が、その支持を物語っている。特にクリエイティブ領域では、コンテンツ生成→レビュー→修正というイテレーティブなワークフローが不可欠であり、LangGraphの「サイクル（ループ）対応」が真価を発揮する。
背景がわかったところで、抽象的な概念から順に、具体的な仕組みを見ていこう。
4. LangGraphの基本概念
4.1 グラフ構造：ノードとエッジ
LangGraphの核心は「有向グラフ」だ。ワークフローを「ノード」（処理単位）と「エッジ」（接続）で表現する。
[アイデア出し] → [執筆] → [レビュー] → [完成]
↑ ↓
└───[修正]←─┘
この構造により、「レビューで問題があれば修正に回し、修正が終わったら再度レビューする」というループが自然に表現できる。
4.2 ステート（状態）：共有されるコンテキスト
LangGraphの特徴的な機能がTypedDictベースの明示的な状態管理だ。
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class CreativeState(TypedDict):
"""クリエイティブワークフローの状態定義"""
topic: str # テーマ
draft: str # 下書き
feedback: str # フィードバック
final_content: str # 最終成果物
revision_count: Annotated[int, lambda x, y: y] # 修正回数
messages: Annotated[list, add_messages] # メッセージ履歴
この状態が全ノード間で共有され、各ノードは状態を読み取り・更新しながら処理を進める。
4.3 条件分岐：動的なルーティング
LangGraphの強力な機能が「条件付きエッジ」だ。実行時の状態に応じて、次にどのノードに進むかを動的に決定できる。
def should_continue(state: CreativeState) -> str:
"""レビュー結果に基づいてルーティングを決定"""
if "approved" in state["feedback"].lower():
return "finalize"
elif state["revision_count"] >= 3:
return "human_review" # 3回修正しても通らなければ人間に委ねる
else:
return "revise"
この仕組みにより、「品質が基準に達するまで自動でイテレーション」「手に負えなければ人間にエスカレーション」という現実的なワークフローが実現する。
4.4 チェックポイント：永続化と復旧
LangGraphには組み込みのチェックポイント機能がある。これにより、処理の途中で失敗しても、最後のチェックポイントから再開できる。長時間実行されるクリエイティブワークフローでは必須の機能だ。
基本概念が理解できたところで、これらの抽象的な概念を具体的なコードで実装していこう。
5. 実際に使ってみよう
5.1 環境構築
# 必要なパッケージのインストール
pip install langgraph langchain langchain-openai python-dotenv
# オプション: LangSmith（デバッグ・可視化用）
pip install langsmith
5.2 設定ファイルの準備
以下の3種類の設定ファイルを用意している。用途に応じて選択してほしい。
開発環境用（config.yaml）
# config.yaml - 開発環境用（このままコピーして使える）
# LangGraphクリエイティブエージェントの設定
llm:
provider: "openai"
model: "gpt-4o-mini" # 開発時はコスト重視
temperature: 0.7 # クリエイティブタスクなのでやや高め
max_tokens: 2000
workflow:
max_revisions: 3 # 最大修正回数
timeout_seconds: 120 # ノードごとのタイムアウト
checkpoint_enabled: true
logging:
level: "DEBUG" # 開発時は詳細ログ
output: "console"
langsmith:
enabled: true # 開発時はトレース有効
project: "creative-agent-dev"
本番環境用（config.production.yaml）
# config.production.yaml - 本番環境用（このままコピーして使える）
# 高品質・高信頼性向けの設定
llm:
provider: "openai"
model: "gpt-4o" # 本番は高品質モデル
temperature: 0.5 # 安定性重視
max_tokens: 4000
retry_count: 3
retry_delay: 1.0
workflow:
max_revisions: 5 # 品質担保のため多めに
timeout_seconds: 300
checkpoint_enabled: true
checkpoint_backend: "postgres" # 本番はDB永続化
logging:
level: "INFO"
output: "file"
file_path: "/var/log/creative-agent/app.log"
langsmith:
enabled: true
project: "creative-agent-prod"
monitoring:
metrics_enabled: true
alert_on_failure: true
テスト環境用（config.test.yaml）
# config.test.yaml - テスト/CI用（このままコピーして使える）
# 高速・低コスト・再現性重視の設定
llm:
provider: "openai"
model: "gpt-4o-mini" # コスト削減
temperature: 0.0 # 再現性のため固定
max_tokens: 500 # テストは短い出力で十分
mock_enabled: true # モックモード（APIを叩かない）
workflow:
max_revisions: 1 # テストは1回で十分
timeout_seconds: 30
checkpoint_enabled: false # テストでは不要
logging:
level: "WARNING" # ノイズ軽減
output: "console"
langsmith:
enabled: false # CIではトレース不要
test:
seed: 42 # 再現性のための乱数シード
sample_size: 3 # 小さいサンプルで高速化
デバッグ環境用（config.debug.yaml）
# config.debug.yaml - デバッグ特化（このままコピーして使える）
# 問題調査時の詳細情報取得向け
llm:
provider: "openai"
model: "gpt-4o-mini"
temperature: 0.0 # 再現性確保
max_tokens: 2000
workflow:
max_revisions: 2
timeout_seconds: 600 # デバッグ時は長めに
checkpoint_enabled: true
checkpoint_backend: "sqlite"
step_by_step: true # ステップ実行モード
logging:
level: "DEBUG"
output: "both" # コンソールとファイル両方
file_path: "./debug.log"
include_state: true # 状態の詳細を含める
include_timestamps: true
langsmith:
enabled: true
project: "creative-agent-debug"
trace_all: true # 全ての呼び出しをトレース
5.3 基本的な使い方
"""
LangGraph クリエイティブエージェント - 基本実装
使い方: python creative_agent.py
必要なパッケージ: pip install langgraph langchain langchain-openai python-dotenv
"""
import os
from typing import TypedDict, Annotated, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
# 環境変数の読み込み
load_dotenv()
# =============================================================================
# 状態定義
# =============================================================================
class CreativeState(TypedDict):
"""クリエイティブワークフローの状態"""
topic: str # 執筆テーマ
draft: str # 下書き
feedback: str # レビューフィードバック
final_content: str # 最終成果物
revision_count: int # 修正回数
# =============================================================================
# ノード定義
# =============================================================================
def ideation_node(state: CreativeState) -> dict:
"""アイデア出しノード: テーマを受け取り、構成案を作成"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.8)
messages = [
SystemMessage(content="あなたは創造的なコンテンツプランナーです。"),
HumanMessage(content=f"""
テーマ「{state['topic']}」について、魅力的な記事の構成案を作成してください。
以下を含めてください：
- キャッチーなタイトル案
- 3-5つの主要セクション
- 各セクションの要点
""")
]
response = llm.invoke(messages)
return {"draft": response.content, "revision_count": 0}
def writing_node(state: CreativeState) -> dict:
"""執筆ノード: 構成案をもとに本文を執筆"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
messages = [
SystemMessage(content="あなたはプロのテクニカルライターです。"),
HumanMessage(content=f"""
以下の構成案をもとに、読みやすい技術記事を執筆してください。
構成案:
{state['draft']}
要件:
- 各セクションは具体的な例を含める
- 専門用語には簡潔な説明を付ける
- 読者に語りかける文体で
""")
]
response = llm.invoke(messages)
return {"draft": response.content}
def review_node(state: CreativeState) -> dict:
"""レビューノード: 執筆された内容を評価"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
messages = [
SystemMessage(content="あなたは厳格な編集者です。"),
HumanMessage(content=f"""
以下の記事をレビューしてください。
記事:
{state['draft']}
以下の観点で評価し、フィードバックを提供してください：
1. 論理的な構成
2. 読みやすさ
3. 具体例の適切さ
4. 専門用語の説明
問題がなければ「APPROVED」、修正が必要なら具体的な改善点を記載してください。
""")
]
response = llm.invoke(messages)
return {"feedback": response.content}
def revision_node(state: CreativeState) -> dict:
"""修正ノード: フィードバックを反映して修正"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
messages = [
SystemMessage(content="あなたはプロのテクニカルライターです。"),
HumanMessage(content=f"""
以下のフィードバックを反映して、記事を修正してください。
現在の記事:
{state['draft']}
フィードバック:
{state['feedback']}
修正版を出力してください。
""")
]
response = llm.invoke(messages)
return {
"draft": response.content,
"revision_count": state["revision_count"] + 1
}
def finalize_node(state: CreativeState) -> dict:
"""完成ノード: 最終成果物として確定"""
return {"final_content": state["draft"]}
# =============================================================================
# ルーティング関数
# =============================================================================
def should_continue(state: CreativeState) -> Literal["revision", "finalize"]:
"""レビュー結果に基づいてルーティングを決定"""
feedback = state.get("feedback", "").lower()
revision_count = state.get("revision_count", 0)
# APPROVEDならば完成へ
if "approved" in feedback:
return "finalize"
# 3回修正してもダメなら諦めて完成へ
if revision_count >= 3:
print(f"[WARNING] 修正上限({revision_count}回)に達しました。強制完了します。")
return "finalize"
# それ以外は修正へ
return "revision"
# =============================================================================
# グラフ構築
# =============================================================================
def build_creative_graph():
"""クリエイティブワークフローのグラフを構築"""
# StateGraphを初期化
workflow = StateGraph(CreativeState)
# ノードを追加
workflow.add_node("ideation", ideation_node)
workflow.add_node("writing", writing_node)
workflow.add_node("review", review_node)
workflow.add_node("revision", revision_node)
workflow.add_node("finalize", finalize_node)
# エッジを追加
workflow.add_edge("ideation", "writing")
workflow.add_edge("writing", "review")
# 条件付きエッジ: レビュー後のルーティング
workflow.add_conditional_edges(
"review",
should_continue,
{
"revision": "revision",
"finalize": "finalize"
}
)
# 修正後は再度レビューへ
workflow.add_edge("revision", "review")
# 完成したら終了
workflow.add_edge("finalize", END)
# エントリーポイントを設定
workflow.set_entry_point("ideation")
# チェックポイント（メモリ永続化）を設定してコンパイル
memory = MemorySaver()
return workflow.compile(checkpointer=memory)
# =============================================================================
# 実行
# =============================================================================
def main():
"""メイン実行関数"""
# グラフを構築
graph = build_creative_graph()
# 初期状態を設定
initial_state = {
"topic": "LangGraphでAIクリエイティブエージェントを作る方法",
"draft": "",
"feedback": "",
"final_content": "",
"revision_count": 0
}
# スレッドIDを設定（チェックポイント用）
config = {"configurable": {"thread_id": "creative-session-001"}}
print("=" * 60)
print("クリエイティブエージェント起動")
print("=" * 60)
print(f"テーマ: {initial_state['topic']}")
print("-" * 60)
# グラフを実行
final_state = graph.invoke(initial_state, config)
print("\n" + "=" * 60)
print("完成コンテンツ")
print("=" * 60)
print(final_state["final_content"])
print("-" * 60)
print(f"修正回数: {final_state['revision_count']}")
if __name__ == "__main__":
main()
5.4 実行結果
上記のコードを実行すると、以下のような出力が得られる：
============================================================
クリエイティブエージェント起動
============================================================
テーマ: LangGraphでAIクリエイティブエージェントを作る方法
------------------------------------------------------------
============================================================
完成コンテンツ
============================================================
# LangGraphでAIクリエイティブエージェントを作る方法
## はじめに
AIエージェント開発の世界で、LangGraphは今最も注目されている...
[以下、生成されたコンテンツが出力される]
------------------------------------------------------------
修正回数: 1
5.5 よくあるエラーと対処法
| エラー | 原因 | 対処法 |
|---|---|---|
ImportError: cannot import name 'StateGraph' |
langgraphのバージョンが古い |
pip install -U langgraph で最新版に更新 |
openai.AuthenticationError |
APIキー未設定または無効 |
.envファイルにOPENAI_API_KEY=sk-xxxを設定 |
RecursionError: maximum recursion depth exceeded |
ルーティングループが無限に続いている |
should_continue関数で必ず終了条件を設定 |
KeyError: 'draft' |
状態の初期値が未設定 |
initial_stateで全フィールドを初期化 |
TypeError: 'NoneType' object is not subscriptable |
ノードがNoneを返している | ノード関数で必ずdictをreturnする |
基本的な使い方をマスターしたので、次は応用例を見ていこう。
6. ユースケース別ガイド
6.1 ユースケース1: ブログ記事自動生成システム
- 想定読者: テックブログを運営しているエンジニア
- 推奨構成: アイデア → リサーチ → 執筆 → SEO最適化 → 公開
- サンプルコード:
"""
ブログ記事自動生成エージェント
想定: 週次でテックブログを更新したいエンジニア向け
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class BlogState(TypedDict):
topic: str
keywords: list[str]
outline: str
content: str
seo_optimized: str
def research_node(state: BlogState) -> dict:
"""キーワードリサーチを行う"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
response = llm.invoke([
SystemMessage(content="あなたはSEOスペシャリストです。"),
HumanMessage(content=f"""
テーマ「{state['topic']}」に関連する検索キーワードを5つ提案してください。
JSON配列形式で出力: ["keyword1", "keyword2", ...]
""")
])
# 簡易的なパース（本番ではstructured outputを使う）
import json
try:
keywords = json.loads(response.content)
except:
keywords = [state['topic']]
return {"keywords": keywords}
def outline_node(state: BlogState) -> dict:
"""記事の構成案を作成"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
response = llm.invoke([
SystemMessage(content="あなたはコンテンツストラテジストです。"),
HumanMessage(content=f"""
テーマ: {state['topic']}
キーワード: {', '.join(state['keywords'])}
SEOに強い記事構成を作成してください。
""")
])
return {"outline": response.content}
def write_node(state: BlogState) -> dict:
"""本文を執筆"""
llm = ChatOpenAI(model="gpt-4o", temperature=0.7) # 高品質モデル
response = llm.invoke([
SystemMessage(content="あなたはテクニカルライターです。"),
HumanMessage(content=f"""
以下の構成で記事を執筆してください。
構成:
{state['outline']}
キーワードを自然に含めてください: {', '.join(state['keywords'])}
""")
])
return {"content": response.content}
def seo_optimize_node(state: BlogState) -> dict:
"""SEO最適化"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
response = llm.invoke([
SystemMessage(content="あなたはSEOエディターです。"),
HumanMessage(content=f"""
以下の記事をSEO最適化してください。
記事:
{state['content']}
最適化ポイント:
- メタディスクリプションを追加
- 見出しタグの最適化
- 内部リンクの提案
""")
])
return {"seo_optimized": response.content}
def build_blog_graph():
workflow = StateGraph(BlogState)
workflow.add_node("research", research_node)
workflow.add_node("outline", outline_node)
workflow.add_node("write", write_node)
workflow.add_node("seo_optimize", seo_optimize_node)
workflow.add_edge("research", "outline")
workflow.add_edge("outline", "write")
workflow.add_edge("write", "seo_optimize")
workflow.add_edge("seo_optimize", END)
workflow.set_entry_point("research")
return workflow.compile()
# 使用例
if __name__ == "__main__":
graph = build_blog_graph()
result = graph.invoke({
"topic": "Pythonの非同期処理入門",
"keywords": [],
"outline": "",
"content": "",
"seo_optimized": ""
})
print(result["seo_optimized"])
6.2 ユースケース2: マルチエージェント脚本生成
- 想定読者: 動画コンテンツクリエイター
- 推奨構成: プロット → キャラクター設計 → シーン執筆 → ダイアログ生成
- サンプルコード:
"""
マルチエージェント脚本生成システム
想定: YouTube動画や短編アニメの脚本を自動生成したいクリエイター向け
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class ScriptState(TypedDict):
premise: str # 前提・設定
genre: str # ジャンル
plot: str # プロット
characters: str # キャラクター設定
scenes: list[str] # シーン一覧
dialogue: str # ダイアログ
final_script: str # 最終脚本
def genre_detector(state: ScriptState) -> dict:
"""ジャンルを自動検出"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
response = llm.invoke([
SystemMessage(content="入力から最適なジャンルを1つ選んでください。"),
HumanMessage(content=f"""
前提: {state['premise']}
選択肢: comedy, drama, action, fantasy, sci-fi, horror
ジャンル名のみ出力:
""")
])
return {"genre": response.content.strip().lower()}
def plot_generator(state: ScriptState) -> dict:
"""プロットを生成"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.8)
response = llm.invoke([
SystemMessage(content=f"あなたは{state['genre']}ジャンルの脚本家です。"),
HumanMessage(content=f"""
前提: {state['premise']}
3幕構成のプロットを作成してください:
- 第1幕: 設定と導入
- 第2幕: 対立と展開
- 第3幕: クライマックスと解決
""")
])
return {"plot": response.content}
def character_designer(state: ScriptState) -> dict:
"""キャラクターを設計"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
response = llm.invoke([
SystemMessage(content="あなたはキャラクターデザイナーです。"),
HumanMessage(content=f"""
プロット: {state['plot']}
主要キャラクター（3名まで）を設計してください:
- 名前
- 性格
- 動機
- 特徴的な話し方
""")
])
return {"characters": response.content}
def scene_writer(state: ScriptState) -> dict:
"""シーンを執筆"""
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
response = llm.invoke([
SystemMessage(content="あなたはシーン構成の専門家です。"),
HumanMessage(content=f"""
プロット: {state['plot']}
キャラクター: {state['characters']}
5つのキーシーンを作成してください。
各シーンには場所、時間、登場人物、アクションを含めてください。
""")
])
scenes = response.content.split("\n\n")
return {"scenes": scenes}
def dialogue_writer(state: ScriptState) -> dict:
"""ダイアログを生成"""
llm = ChatOpenAI(model="gpt-4o", temperature=0.8)
scenes_text = "\n\n".join(state['scenes'])
response = llm.invoke([
SystemMessage(content="あなたはダイアログライターです。自然で魅力的な会話を書きます。"),
HumanMessage(content=f"""
キャラクター: {state['characters']}
シーン: {scenes_text}
各シーンのダイアログを脚本形式で書いてください。
キャラクターの個性が出るように注意してください。
""")
])
return {"dialogue": response.content}
def script_finalizer(state: ScriptState) -> dict:
"""最終脚本にまとめる"""
final = f"""
# 脚本: {state['premise']}
## ジャンル: {state['genre']}
---
## プロット
{state['plot']}
---
## キャラクター
{state['characters']}
---
## 本編
{state['dialogue']}
"""
return {"final_script": final}
def build_script_graph():
workflow = StateGraph(ScriptState)
workflow.add_node("detect_genre", genre_detector)
workflow.add_node("generate_plot", plot_generator)
workflow.add_node("design_characters", character_designer)
workflow.add_node("write_scenes", scene_writer)
workflow.add_node("write_dialogue", dialogue_writer)
workflow.add_node("finalize", script_finalizer)
# 直線的なフロー
workflow.add_edge("detect_genre", "generate_plot")
workflow.add_edge("generate_plot", "design_characters")
workflow.add_edge("design_characters", "write_scenes")
workflow.add_edge("write_scenes", "write_dialogue")
workflow.add_edge("write_dialogue", "finalize")
workflow.add_edge("finalize", END)
workflow.set_entry_point("detect_genre")
return workflow.compile()
# 使用例
if __name__ == "__main__":
graph = build_script_graph()
result = graph.invoke({
"premise": "迷子になったドラゴンが家族を見つける、子供向けファンタジー",
"genre": "",
"plot": "",
"characters": "",
"scenes": [],
"dialogue": "",
"final_script": ""
})
print(result["final_script"])
6.3 ユースケース3: レビュー→修正ループ付きコピーライティング
- 想定読者: マーケター、広告運用担当者
- 推奨構成: ドラフト → 評価 → 修正（ループ）→ 完成
- サンプルコード:
"""
レビューループ付きコピーライティングエージェント
想定: 広告コピーやランディングページの文言を繰り返し改善したいマーケター向け
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class CopyState(TypedDict):
product: str # 商品・サービス名
target_audience: str # ターゲット層
copy: str # 現在のコピー
score: int # 評価スコア (0-100)
feedback: str # 改善フィードバック
iteration: int # イテレーション回数
final_copy: str # 最終コピー
def draft_writer(state: CopyState) -> dict:
"""初稿を作成"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.8)
response = llm.invoke([
SystemMessage(content="あなたは一流のコピーライターです。"),
HumanMessage(content=f"""
商品: {state['product']}
ターゲット: {state['target_audience']}
心に刺さる広告コピーを3パターン作成してください。
各パターンは:
- ヘッドライン（20文字以内）
- ボディコピー（100文字以内）
- CTA（行動喚起）
""")
])
return {"copy": response.content, "iteration": 0}
def evaluator(state: CopyState) -> dict:
"""コピーを評価"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
response = llm.invoke([
SystemMessage(content="""
あなたは広告効果の専門家です。
コピーを0-100のスコアで評価し、改善点を指摘してください。
出力形式:
SCORE: [数字]
FEEDBACK: [改善点]
"""),
HumanMessage(content=f"""
商品: {state['product']}
ターゲット: {state['target_audience']}
評価対象コピー:
{state['copy']}
""")
])
# スコアを抽出（簡易パース）
content = response.content
score = 70 # デフォルト
feedback = content
if "SCORE:" in content:
try:
score_line = [l for l in content.split("\n") if "SCORE:" in l][0]
score = int(''.join(filter(str.isdigit, score_line)))
except:
pass
if "FEEDBACK:" in content:
feedback = content.split("FEEDBACK:")[-1].strip()
return {"score": score, "feedback": feedback}
def reviser(state: CopyState) -> dict:
"""フィードバックを反映して修正"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6)
response = llm.invoke([
SystemMessage(content="あなたはコピーライターです。フィードバックを反映して改善してください。"),
HumanMessage(content=f"""
現在のコピー:
{state['copy']}
フィードバック:
{state['feedback']}
改善版を出力してください。
""")
])
return {"copy": response.content, "iteration": state["iteration"] + 1}
def finalizer(state: CopyState) -> dict:
"""最終コピーとして確定"""
return {"final_copy": state["copy"]}
def should_revise(state: CopyState) -> Literal["revise", "finalize"]:
"""修正が必要か判定"""
# スコア80以上、または3回修正したら完了
if state["score"] >= 80:
print(f"[OK] スコア {state['score']} 達成。完了します。")
return "finalize"
if state["iteration"] >= 3:
print(f"[WARN] 修正上限に達しました。スコア {state['score']} で完了します。")
return "finalize"
print(f"[INFO] スコア {state['score']}。修正を続けます。（{state['iteration'] + 1}回目）")
return "revise"
def build_copy_graph():
workflow = StateGraph(CopyState)
workflow.add_node("draft", draft_writer)
workflow.add_node("evaluate", evaluator)
workflow.add_node("revise", reviser)
workflow.add_node("finalize", finalizer)
workflow.add_edge("draft", "evaluate")
workflow.add_conditional_edges(
"evaluate",
should_revise,
{
"revise": "revise",
"finalize": "finalize"
}
)
workflow.add_edge("revise", "evaluate") # 修正後は再評価
workflow.add_edge("finalize", END)
workflow.set_entry_point("draft")
return workflow.compile()
# 使用例
if __name__ == "__main__":
graph = build_copy_graph()
result = graph.invoke({
"product": "AI搭載スマート家計簿アプリ",
"target_audience": "20-30代の共働き夫婦",
"copy": "",
"score": 0,
"feedback": "",
"iteration": 0,
"final_copy": ""
})
print("\n=== 最終コピー ===")
print(result["final_copy"])
print(f"\n最終スコア: {result['score']}, イテレーション: {result['iteration']}")
ユースケースが把握できたところで、この記事を読んだ後の学習パスを確認しよう。
7. 学習ロードマップ
この記事を読んだ後、次のステップとして以下をおすすめする。
初級者向け（まずはここから）
- LangChain Academy - Introduction to LangGraph: 無料のオンラインコースで基礎を固める
- LangGraph公式クイックスタート: 最小限のコードで動くエージェントを作る
- LangGraph Templates: 事前構築されたテンプレートから学ぶ
中級者向け（実践に進む）
- LangGraph Guides: ストリーミング、永続化、デザインパターンなど
- LangSmith Integration: デバッグとオブザーバビリティを導入
- MCP対応の実装: 外部ツール連携を標準化
上級者向け（さらに深く）
- LangGraph Platform: 本番デプロイと運用
- Human-in-the-Loop設計: 人間の介入を組み込む高度な設計
- LangGraph GitHub: ソースコードを読んで内部実装を理解
8. 他のフレームワークとの比較
最後に、主要なAIエージェントフレームワークとLangGraphを比較しておこう。
| 機能 | LangGraph | CrewAI | AutoGen | Google ADK | OpenAI SDK |
|---|---|---|---|---|---|
| マルチエージェント | ◎ | ◎ | ◎ | ◎ | ○ |
| 状態管理 | ◎ 明示的 | ○ 暗黙的 | ○ | ○ | △ |
| Human-in-the-loop | ◎ | ○ | ○ | ◎ | ○ |
| MCP対応 | ○ | ○ | ○ | ○ | ○ |
| ビジュアルUI | △ Studio | △ | ◎ Studio | ◎ Designer | × |
| 学習曲線 | 中 | 低 | 中 | 中 | 低 |
| コミュニティ | ◎ 21k+ stars | ○ 25k+ | ○ 40k+ | ○ 10k+ | ○ 15k+ |
LangGraphの強みは「明示的な状態管理」と「グラフベースの制御フロー」だ。CrewAIは簡単に始められるが、複雑なワークフローでは制御が難しくなる。AutoGenは研究用途には良いが本番運用には課題がある。LangGraphは「本番で使えるエージェント」を作るための最適解だと言える。
9. まとめ
この記事では、LangGraphについて以下を解説した：
- LangGraphはグラフベースのエージェントオーケストレーションフレームワークである
- 明示的な状態管理とサイクル対応が、クリエイティブワークフローに最適
- 条件分岐、チェックポイント、Human-in-the-loopなど本番運用に必要な機能が揃っている
私の所感
正直に言うと、LangGraphは学習曲線がやや高い。CrewAIのように「ロールを定義してGo」とはいかない。しかし、その分だけ「何が起きているか」が明確になる。
クリエイティブワークフローは、単純な直線フローでは済まない。「品質が基準に達するまでループ」「人間のレビューを待って再開」——こうした要件に、LangGraphは真正面から応えてくれる。
もしあなたが「本番で使えるAIクリエイティブエージェント」を作りたいなら、LangGraphは最有力の選択肢だ。この記事がその第一歩になれば幸いである。