メモリと状態管理：長期実行エージェント

エージェントの実装で最も難しい問題が メモリ管理。コンテキストウィンドウは無限ではないし、コストとレイテンシは履歴量に比例します。

メモリの3階層設計

直近コンテキストの管理

# 履歴を「最新N件」で打ち切るシンプル実装

def trim_history(messages, max_tokens=50000):
    # システムプロンプトは保持
    system = messages[0]
    history = messages[1:]

    # トークン数で計算（簡易）
    while estimate_tokens(history) > max_tokens:
        # 最古を削る
        history.pop(0)

    return [system] + history

セッション要約による圧縮

履歴が長くなったら、古い部分をサマリに圧縮 します。

# 古い履歴を要約して圧縮

def compress_history(messages, threshold=80000):
    if estimate_tokens(messages) < threshold:
        return messages

    # 古い履歴（先頭N件）を抜き出す
    old = messages[1:30]  # 最初の30件
    recent = messages[30:]

    # サマリ生成
    summary = call_claude(
        system="次の対話履歴を要点だけ簡潔にまとめてください。",
        messages=old,
    )

    # サマリで置換
    return [
        messages[0],  # システムプロンプト
        {"role": "user", "content": f"【これまでの経緯のサマリ】n{summary}"},
        {"role": "assistant", "content": "了解しました。続きを処理します。"},
    ] + recent

サマリの注意点

• 重要事実は省略しない：ユーザーIDやエラーコードは要約しても残す
• 未解決のタスクは強調：「Aは未完了、Bは完了」と明確に
• ツール呼び出し結果の数値：要約で消えやすい、再現可能なように残す

永続メモリの設計

セッションを跨いで覚えたい情報は 外部DB に保存します。

# メモリ・ストアの実装例（Postgres + JSON）

CREATE TABLE agent_memory (
    user_id TEXT,
    key TEXT,
    value JSONB,
    created_at TIMESTAMP,
    last_accessed_at TIMESTAMP,
    access_count INTEGER,
    importance_score FLOAT
);

# メモリ書き込みツール
{
  "name": "remember",
  "description": "ユーザーに関する事実を記憶する",
  "input_schema": {
    "type": "object",
    "properties": {
      "key": {"type": "string", "description": "メモリのキー"},
      "value": {"type": "string", "description": "記憶する内容"},
      "importance": {"type": "number", "description": "重要度 0.0〜1.0"}
    }
  }
}

# メモリ読み込みツール
{
  "name": "recall",
  "description": "記憶した事実を呼び出す",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": {"type": "string", "description": "検索クエリ"},
      "top_k": {"type": "integer", "description": "取得件数", "default": 5}
    }
  }
}

セマンティック検索によるメモリ呼び出し

メモリ件数が増えたら、キーマッチではなく 埋め込みベース検索。

# 埋め込み + ベクトルDB での実装

# 書き込み時：
embedding = create_embedding(value)
store_vector(user_id, key, value, embedding)

# 読み込み時：
query_embedding = create_embedding(query)
results = vector_db.search(user_id, query_embedding, top_k=5)
return [r.value for r in results]

メモリの忘却戦略

すべてを永続化するとコストとプライバシーの問題が出ます。

# 自動忘却ルール

1. 時間ベース：30日アクセスがないメモリは削除
2. 重要度ベース：importance < 0.3 のメモリは7日後に削除
3. ユーザー要請：「忘れて」と言われたら即削除
4. プライバシー：機密情報フラグ付きメモリは2時間で自動削除

# 実装
DELETE FROM agent_memory
WHERE last_accessed_at < NOW() - INTERVAL '30 days'
   OR (importance_score < 0.3 AND created_at < NOW() - INTERVAL '7 days')

状態管理のアーキテクチャ

大規模エージェントでは、状態管理を 独立したコンポーネント にします。

class AgentState:
    def __init__(self, user_id, session_id):
        self.user_id = user_id
        self.session_id = session_id
        self.short_term = []  # messages配列
        self.summary = ""     # 圧縮されたサマリ
        self.facts = {}       # このセッション中に確定した事実
        self.pending_actions = []  # 実行待ちアクション
        self.completed_tasks = []  # 完了したタスク

    def to_messages(self):
        # APIに渡す形式に変換
        ...

    def update(self, message, tool_result):
        # 状態を進める
        ...

    def persist(self):
        # DBに永続化
        ...

    @classmethod
    def restore(cls, user_id, session_id):
        # DBから復元
        ...

多ユーザー対応の注意

このレッスンのまとめ

「直近コンテキスト → セッション要約 → 永続メモリ」の3階層と、忘却戦略・状態管理の仕組みでエージェントは長期実行に耐えます。次のレッスンでは、プランニングと自己修正のパターンを学びます。