本記事の目標
本シリーズでは、LangGraphを使って技術記事の校正エージェントを構築する方法を解説します。第1回の本記事では、LangGraphの基本的な概念と、今回作成するエージェントの全体像を紹介します。
LangGraphとは
LangGraphは、LangChainチームが開発したLLMアプリケーションのためのフレームワークです。グラフベースのワークフローを定義することで、複雑なエージェントの挙動を制御できます。
従来のLLMアプリケーションでは、単一のプロンプトでタスクを完了させるか、シンプルなチェーンで処理を繋げることが一般的でした。しかし、複雑なタスクでは以下のような課題があります。
- 複数の処理を並列で実行したい
- 条件分岐によって異なる処理パスを選択したい
- 中間状態を保持しながら複数ステップを実行したい
LangGraphは、これらの課題を状態機械（State Machine） の概念で解決します。
LangGraphの基本コンセプト
LangGraphには3つの重要な概念があります。
1. State（状態）
Stateは、グラフ全体で共有されるデータを定義します。TypedDictを使って型安全に定義できます。
from typing import Annotated, NotRequired, TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage
class ProofreadState(TypedDict):
# 入力
target_article: str
article_type: str
# 処理結果
issues: NotRequired[list]
# 最終出力
final_report: NotRequired[str]
messages: NotRequired[Annotated[list[BaseMessage], add_messages]]
Annotatedを使うことで、状態の更新方法を指定できます。例えばadd_messagesは、新しいメッセージを既存のリストに追加する動作を定義しています。
2. Node（ノード）
Nodeは、グラフ内の処理単位です。状態を受け取り、更新された状態（の一部）を返す関数として定義します。
async def parse_article_node(state: ProofreadState) -> dict:
"""記事を解析するノード"""
target_article = state["target_article"]
# 記事のパース処理
parsed = parse_article(target_article)
return {"parsed_article": parsed}
3. Edge（エッジ）
Edgeは、ノード間の遷移を定義します。単純な遷移の他に、条件分岐も可能です。
from langgraph.graph import END, START, StateGraph
builder = StateGraph(ProofreadState)
# ノードの追加
builder.add_node("parse", parse_article_node)
builder.add_node("check", check_article_node)
builder.add_node("report", generate_report_node)
# エッジの定義
builder.add_edge(START, "parse")
builder.add_edge("parse", "check")
builder.add_edge("check", "report")
builder.add_edge("report", END)
# グラフのコンパイル
graph = builder.compile()
校正エージェントの全体像
今回作成する校正エージェントは、以下のようなフローで動作します。
START
↓
Parser（記事解析）
↓
─────────────────────────────────────────
並列実行
├─ Markdown Checker
├─ Link Checker
├─ Structure Checker
├─ Privacy Checker
├─ Language Quality Checker
├─ Technical Accuracy Checker（MCP連携）
└─ Code Quality Checker
─────────────────────────────────────────
↓
Issue Aggregator（結果統合）
↓
Report Generator
↓
END
このフローでは、記事の解析後に7つのチェッカーが並列で実行されます。LangGraphでは、同じノードから複数のノードにエッジを張ることで、並列実行を実現できます。
# 並列実行の定義
builder.add_edge("parse", "check_markdown_format")
builder.add_edge("parse", "check_link")
builder.add_edge("parse", "check_structure")
builder.add_edge("parse", "check_language_quality")
builder.add_edge("parse", "check_privacy")
builder.add_edge("parse", "check_technical_accuracy")
builder.add_edge("parse", "check_code_quality")
# 並列処理の合流
builder.add_edge(
[
"check_markdown_format",
"check_link",
"check_structure",
"check_language_quality",
"check_privacy",
"check_technical_accuracy",
"check_code_quality",
],
"aggregate",
)
状態のReducer関数
複数のノードが並列で実行される場合、状態の更新をどのようにマージするかが重要です。LangGraphではReducer関数を使ってこの挙動を定義します。
def add_values(left: list | None, right: list | None) -> list:
"""リストを結合するReducer"""
if left is None:
left = []
if right is None:
right = []
return left + right
def keep_value(left, right):
"""最初の値を保持するReducer"""
if left is not None:
return left
return right
class ProofreadState(TypedDict):
# 記事タイプは最初に設定された値を保持
article_type: Annotated[str, keep_value]
# issuesは各チェッカーの結果を結合
issues: Annotated[list[Issue], add_values]
add_valuesは、各チェッカーが検出したissueを一つのリストにまとめます。これにより、並列実行の結果を自然に統合できます。
非同期実行
LangGraphは非同期実行をネイティブにサポートしています。async/awaitを使うことで、I/O待ちの間に他の処理を進めることができます。
async def main():
# グラフの構築（非同期）
graph = await create_proofread_graph()
# 初期状態の定義
initial_state = ProofreadState(
target_article=article_content,
article_type="zenn",
)
# グラフの実行（非同期）
result = await graph.ainvoke(initial_state)
print(result["final_report"])
asyncio.run(main())
今後の応用
本記事では、LangGraphの基本概念と校正エージェントの全体像を紹介しました。次回以降の記事では、以下のトピックを解説していきます。
- ABCを使ったチェッカーの抽象化
- MCPを使った外部ツール連携
- LangGraph Studioでのデバッグ方法
- サブグラフによるサブエージェント設計
まとめ
本記事では、LangGraphの基本的な概念（State、Node、Edge）と、今回作成する校正エージェントの全体像を紹介しました。
LangGraphを使うことで、複数の処理を並列実行し、その結果を自然に統合できます。また、非同期実行のサポートにより、効率的なエージェントの実装が可能です。
次回は、ABCを使ったチェッカーの抽象化について解説します。