AIエージェントが長時間稼働するようになって、改めて問われている問題がある。「状態をどこに持つか」だ。
単発の処理ならステートレスで十分だった。しかし人間の代わりに何時間も、何日も動き続けるエージェントでは、文脈の維持・障害からの回復・並行エージェント間の同期が全て「状態設計」にかかってくる。フロントエンドで長年議論されてきたRedux的なアプローチと、近年注目が集まるイベントソーシング——両者の実像を、実際のAIエージェント実装コードとともに検証する。
AIエージェントのステート管理に正解はない。ただし、選択を誤るとデバッグ地獄と障害が待っている。この記事では、具体的なトレードオフをコードで示しながら「どのケースでどちらを選ぶか」を明確にする。
スペック比較：Redux的集中管理 vs イベントソーシング
| 比較軸 | Redux的集中管理（集中ストア） | イベントソーシング（イベントログ） |
|---|---|---|
| 状態の形 | 現在のスナップショット（単一） | イベントの累積ログ（追記型） |
| デバッグ | タイムトラベルデバッグが容易 | イベントリプレイで完全再現可能 |
| 書き込み | 上書き（ミューテーション） | 追記のみ（イミュータブル） |
| スケーラビリティ | 書き込みボトルネック発生リスク | 高スループット（追記のみ） |
| 監査ログ | 別途実装が必要 | ログがそのまま監査証跡 |
| コンプレキシティ | 低〜中（フロントエンド技術転用） | 高（CQRS、プロジェクション設計が必要） |
| 長期実行向き | ストア肥大化・一貫性維持が課題 | 非常に高い（設計次第） |
| フレームワーク対応 | LangGraph, CrewAI, OpenAI Agents SDK | Apache Kafka, EventStoreDB + LangGraph |
Redux的集中管理で比較する
LangGraphはRedux的な思想に最も近い実装を提供している。グラフの各ノードが「ディスパッチャー」として動作し、共有ステートを更新する。フロントエンドエンジニアなら「useReducer + Context」に近いと考えると理解しやすい。
以下は、LangGraphでカスタマーサポートエージェントのステートを設計した例だ。
# 動作環境: Python 3.11+, langgraph>=0.2.0, langchain-openai>=0.1.0
# pip install langgraph langchain-openai
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
# ステート定義（Reduxのstoreに相当）
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add] # 追記式
current_intent: str # 分類されたユーザー意図
resolved: bool # 解決済みフラグ
escalation_count: int # エスカレーション回数
session_id: str # セッション識別子
def classify_intent(state: AgentState) -> dict:
"""意図分類ノード — stateを受け取り差分dictを返す（Reducerに相当）"""
last_message = state["messages"][-1].content
# 実際はLLM呼び出し
intent = "billing" if "料金" in last_message else "technical"
return {"current_intent": intent} # ストアのbilling sliceのみ更新
def handle_billing(state: AgentState) -> dict:
"""料金対応ノード"""
response = AIMessage(content="料金についてお調べします。しばらくお待ちください。")
return {"messages": [response]}
def should_escalate(state: AgentState) -> str:
"""エッジ関数 — 条件分岐（Reduxのmiddlewareに相当）"""
if state["escalation_count"] >= 2:
return "human_handoff"
return "respond"
# グラフ構築（Reduxのstore configに相当）
builder = StateGraph(AgentState)
builder.add_node("classify", classify_intent)
builder.add_node("billing", handle_billing)
builder.add_conditional_edges("classify", should_escalate, {
"respond": "billing",
"human_handoff": END
})
graph = builder.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
ポイントは各ノードが「差分dictを返す」設計だ。stateオブジェクトを直接ミューテーションしない。LangGraphはこの差分をReducerのように適用してnew stateを生成する。Reduxを知っていれば、自然に理解できる。
イベントソーシングで比較する
イベントソーシングの本質は「状態（what）を保存するのではなく、起きたこと（why/how）を保存する」発想だ。状態は「イベントの累積ログを再生した結果」として導出される。
AIエージェントにイベントソーシングを適用すると、こんな設計になる。
# 動作環境: Python 3.11+
# pip install langgraph aiofiles
import asyncio
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal
# イベント定義（不変の事実を表す）
@dataclass
class AgentEvent:
event_id: str
event_type: Literal[
"SESSION_STARTED",
"USER_MESSAGE_RECEIVED",
"INTENT_CLASSIFIED",
"TOOL_CALLED",
"TOOL_RESULT_RECEIVED",
"RESPONSE_GENERATED",
"SESSION_ENDED"
]
payload: dict
timestamp: str
agent_id: str
class EventStore:
"""イベントを追記専用ログに保存（絶対に上書きしない）"""
def __init__(self, storage_path: str):
self.path = Path(storage_path)
self.path.mkdir(parents=True, exist_ok=True)
async def append(self, session_id: str, event: AgentEvent) -> None:
log_file = self.path / f"{session_id}.jsonl"
async with aiofiles.open(log_file, "a") as f:
await f.write(json.dumps(asdict(event)) + "n")
async def replay(self, session_id: str) -> dict:
"""イベントをリプレイして現在の状態を導出（プロジェクション）"""
log_file = self.path / f"{session_id}.jsonl"
state = {"messages": [], "current_intent": None, "tool_calls": []}
async with aiofiles.open(log_file, "r") as f:
async for line in f:
event = AgentEvent(**json.loads(line))
# 各イベントタイプに応じてstateを更新（純粋関数）
if event.event_type == "USER_MESSAGE_RECEIVED":
state["messages"].append(event.payload)
elif event.event_type == "INTENT_CLASSIFIED":
state["current_intent"] = event.payload["intent"]
elif event.event_type == "TOOL_CALLED":
state["tool_calls"].append(event.payload)
return state
# 使用例
store = EventStore("/var/agent_events")
event = AgentEvent(
event_id="evt_01j...",
event_type="USER_MESSAGE_RECEIVED",
payload={"content": "請求書の確認をお願いします"},
timestamp=datetime.now(timezone.utc).isoformat(),
agent_id="cs-agent-001"
)
asyncio.run(store.append("session_abc123", event))
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
筆者のおすすめ
正直、「どちらが正解」という話ではない。ユースケースと運用体制で選ぶべきだ。
Redux的集中管理を選ぶべき場合: チームにフロントエンドエンジニアが多い、LangGraph / OpenAI Agents SDKをそのまま使いたい、1セッション数時間程度の中規模エージェント、デバッグ速度を最優先したい——こういった状況では集中管理が圧倒的にやりやすい。LangGraphのチェックポイント機能（AsyncPostgresSaver）と組み合わせれば、pause/resumeも「タイムトラベルデバッグ」も標準機能で賄える。
イベントソーシングを選ぶべき場合: 金融・医療など完全な監査証跡が必要、週単位・月単位で動く長時間エージェント、複数エージェントが並行して同一ドメインを操作する、LLMの意思決定プロセスを事後に完全再現したい——こうした要件があるなら、イベントソーシングの初期コストを払う価値がある。全てのエージェント行動がイベントログに残るため、コンプライアンス対応が格段に楽になる。
ハイブリッドも現実的な選択肢だ。LangGraphをロジック実行エンジンとして使いながら、外部のApache KafkaやEventStoreDBにドメインイベントを書き出す構成は、実運用で増えている。コンプレキシティは上がるが、それぞれの強みを引き出せる。
【要注意】よくある失敗パターンと回避策
失敗1: 全てをメモリ内Dictで管理する
❌ state = {"messages": [], "context": {}} をプロセス内Dictで持つ
⭕ LangGraphのチェックポインター（PostgresSaver/RedisSaver）で外部に永続化する
なぜ重要か: エージェントプロセスが落ちた瞬間、セッション状態が全消失する。長時間エージェントでこれは致命的だ。
失敗2: ミュータブルなステートオブジェクトを複数ノードで共有する
❌ state["messages"].append(msg) をノード内で直接実行する
⭕ 各ノードは差分dictを返し、フレームワークに更新を委ねる
なぜ重要か: 並行実行時に競合状態（race condition）が発生し、メッセージの欠落や重複が起きる。LangGraphがこのパターンを強制しているのには理由がある。
失敗3: イベントログを「ステートのバックアップ」として使う
❌ 定期的にステートをシリアライズしてイベントログに書き込む
⭕ 「何が起きたか（意図・行動・結果）」をドメインイベントとして記録する
なぜ重要か: ステートのスナップショットをイベントログに入れてしまうと、イベントソーシングのメリット（リプレイ・監査・分析）が失われる。イベントは「UserMessageReceived」「ToolCalled」といったビジネス上の事実であるべきだ。
失敗4: 永遠に増え続けるイベントログの設計をしない
❌ 1年間のイベントを1ファイルに蓄積し続ける
⭕ スナップショットを定期的に作成し、古いイベントをアーカイブする
なぜ重要か: 1000万件のイベントをリプレイしてカレントステートを導出しようとすると、起動時間が分単位になる。スナップショット+差分イベントの組み合わせが実用的だ。
参考・出典
- Event-Driven Architecture for AI Agent Systems — Zylos Research（参照日: 2026-04-11）
- Event-Driven Architecture for AI Agents: Patterns and Benefits — Atlan（参照日: 2026-04-11）
- Four Design Patterns for Event-Driven, Multi-Agent Systems — Confluent（参照日: 2026-04-11）
- AI Agent Workflow State Persistence: Best Practices 2026 — Fastio（参照日: 2026-04-11）
AIエージェントのアーキテクチャについては、AIエージェント構築完全ガイドで基本から体系的にまとめています。状態設計が決まったら次はAIエージェント構築ツール徹底比較でフレームワーク選定に進んでください。
AIエージェントの設計・導入に関するご相談は 株式会社Uravation（uravation.com） までお気軽にどうぞ。
この記事はAIgent Lab編集部がお届けしました。
