AIエージェント多ターン記憶実践：長期記憶とコンテキスト圧縮システムの構築

AI与大数据

概要

  • AIエージェントの記憶システムは3層に分かれています：ワーキングメモリ（コンテキストウィンドウ）、短期記憶（セッションレベル）、長期記憶（セッション横断の永続化）
  • 会話コンテキスト圧縮は長時間会話シナリオにおいて必須であり、LLM要約圧縮により100ターンの会話を2Kトークン以内に圧縮できます
  • 長期記憶の3つのストレージモード：ベクトル検索記憶、ナレッジグラフ記憶、構造化キーバリュー記憶
  • 記憶検索のRAGアプローチ：過去の会話をベクトルデータベースに埋め込み、関連性に基づいて重要な断片を再呼出します
  • 本記事では、記憶アーキテクチャ設計から本番デプロイまでの完全なソリューションを提供します（Redis+Milvus永続化実装を含む）

目次

AIエージェント記憶システムの3層アーキテクチャ

┌──────────────────────────────────────────────────────────────┐ │ AI Agent 3層記憶アーキテクチャ │ │ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 1: ワーキングメモリ (Working Memory) │ │ │ │ コンテキストウィンドウ内の会話履歴 │ │ │ │ 容量: 4K-128K tokens │ │ │ │ レイテンシ: 0ms (メモリ内にあり) │ │ │ │ ライフサイクル: 単一リクエスト │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ オーバーフロー圧縮 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 2: 短期記憶 (Short-Term Memory) │ │ │ │ 現在のセッションの完全な会話状態 │ │ │ │ ストレージ: Redis / インメモリキャッシュ │ │ │ │ 容量: 無制限 │ │ │ │ ライフサイクル: 単一セッション (ブラウザ終了後に期限切れ) │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ 重要情報抽出 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 3: 長期記憶 (Long-Term Memory) │ │ │ │ セッション横断のユーザー設定、知識、履歴 │ │ │ │ ストレージ: MilvusベクトルDB + PostgreSQL構造化 │ │ │ │ 容量: 無制限 │ │ │ │ ライフサイクル: 永続 │ │ │ └──────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

3層記憶の比較

項目 ワーキングメモリ 短期記憶 長期記憶
ストレージ LLMコンテキストウィンドウ Redis/メモリ Milvus+PostgreSQL
容量 4K-128K tokens 無制限 無制限
レイテンシ 0ms 1-5ms 10-30ms
ライフサイクル 単一リクエスト 単一セッション 永続
検索方式 直接アクセス Key-Value ベクトル+構造化
代表的な内容 現在の会話 セッション履歴 ユーザー設定/知識

ワーキングメモリ：コンテキストウィンドウ管理

トークン予算配分

┌──────────────────────────────────────────────────────────┐ │ 8Kコンテキストウィンドウのトークン予算配分 │ │ │ │ System Prompt: 500 tokens (6%) │ │ 長期記憶再呼出: 1000 tokens (12%) │ │ 会話履歴: 5000 tokens (62%) │ │ ツール呼出結果: 1000 tokens (12%) │ │ 出力予約領域: 500 tokens (6%) │ │ ───────────────────────────────── │ │ 合計: 8000 tokens │ └──────────────────────────────────────────────────────────┘

コンテキストウィンドウマネージャー

`python from dataclasses import dataclass, field from typing import Optional

@dataclass class Message: role: str content: str token_count: int = 0 metadata: dict = field(default_factory=dict)

class ContextWindowManager: def init( self, max_tokens: int = 8192, system_prompt_tokens: int = 500, reserved_output_tokens: int = 500, memory_recall_tokens: int = 1000, ): self.max_tokens = max_tokens self.system_prompt_tokens = system_prompt_tokens self.reserved_output_tokens = reserved_output_tokens self.memory_recall_tokens = memory_recall_tokens self.available_for_history = ( max_tokens - system_prompt_tokens - reserved_output_tokens - memory_recall_tokens )

def build_context(
    self,
    system_prompt: str,
    memory_facts: list[str],
    conversation_history: list[Message],
    tool_results: list[str] = None,
) -> list[dict]:
    messages = [{"role": "system", "content": system_prompt}]

    if memory_facts:
        memory_text = "ユーザー関連記憶：\n" + "\n".join(f"- {f}" for f in memory_facts)
        messages.append({"role": "system", "content": memory_text})

    budget = self.available_for_history
    selected_history = []
    for msg in reversed(conversation_history):
        if budget <= 0:
            break
        if msg.token_count <= budget:
            selected_history.insert(0, msg)
            budget -= msg.token_count
        else:
            break

    for msg in selected_history:
        messages.append({"role": msg.role, "content": msg.content})

    if tool_results:
        for result in tool_results:
            messages.append({"role": "tool", "content": result})

    return messages

`

短期記憶：セッションレベルの状態永続化

Redisセッションストレージ

`python import json import redis from datetime import timedelta

class SessionMemory: def init(self, redis_url: str = "redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.session_ttl = timedelta(hours=24)

def _key(self, session_id: str) -> str:
    return f"session:{session_id}"

async def save_message(self, session_id: str, role: str, content: str):
    key = self._key(session_id)
    message = {"role": role, "content": content, "timestamp": int(time.time())}
    await self.redis.rpush(key, json.dumps(message, ensure_ascii=False))
    await self.redis.expire(key, self.session_ttl)

async def get_history(self, session_id: str, limit: int = 50) -> list[dict]:
    key = self._key(session_id)
    raw_messages = await self.redis.lrange(key, -limit, -1)
    return [json.loads(m) for m in raw_messages]

async def get_session_summary(self, session_id: str) -> Optional[str]:
    key = f"session_summary:{session_id}"
    summary = await self.redis.get(key)
    return summary.decode() if summary else None

async def save_session_summary(self, session_id: str, summary: str):
    key = f"session_summary:{session_id}"
    await self.redis.set(key, summary, ex=self.session_ttl)

async def clear_session(self, session_id: str):
    await self.redis.delete(self._key(session_id))
    await self.redis.delete(f"session_summary:{session_id}")

`

長期記憶：セッション横断の知識蓄積

3つの長期記憶モード

モード ストレージ 検索方式 適用シナリオ
ベクトル検索記憶 Milvus 意味的類似度 オープンな知識の再呼出
ナレッジグラフ記憶 Neo4j 関係トラバーサル 構造化関係推論
構造化キーバリュー記憶 PostgreSQL 完全一致 ユーザー設定/構成

ベクトル検索記憶の実装

`python from pymilvus import MilvusClient, DataType from openai import OpenAI import hashlib

class VectorLongTermMemory: def init(self, milvus_uri: str, embedding_model: str = "text-embedding-3-small"): self.client = MilvusClient(uri=milvus_uri) self.embedding_client = OpenAI() self.embedding_model = embedding_model self._ensure_collection()

def _ensure_collection(self):
    if self.client.has_collection("agent_memory"):
        return
    schema = self.client.create_schema(auto_id=True, enable_dynamic_field=True)
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="user_id", datatype=DataType.VARCHAR, max_length=128)
    schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535)
    schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=1536)
    schema.add_field(field_name="category", datatype=DataType.VARCHAR, max_length=64)
    schema.add_field(field_name="timestamp", datatype=DataType.INT64)

    index_params = self.client.prepare_index_params()
    index_params.add_index(field_name="embedding", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 200})

    self.client.create_collection("agent_memory", schema=schema, index_params=index_params)

def _embed(self, text: str) -> list[float]:
    response = self.embedding_client.embeddings.create(input=text, model=self.embedding_model)
    return response.data[0].embedding

async def store(self, user_id: str, content: str, category: str = "general"):
    embedding = self._embed(content)
    self.client.insert("agent_memory", [{
        "user_id": user_id,
        "content": content,
        "embedding": embedding,
        "category": category,
        "timestamp": int(time.time()),
    }])

async def recall(self, user_id: str, query: str, top_k: int = 5) -> list[str]:
    query_embedding = self._embed(query)
    results = self.client.search(
        "agent_memory",
        data=[query_embedding],
        filter=f'user_id == "{user_id}"',
        limit=top_k,
        output_fields=["content", "category"],
        search_params={"metric_type": "COSINE", "params": {"ef": 64}},
    )
    return [r["entity"]["content"] for r in results[0]]

`

構造化キーバリュー記憶の実装

`python import asyncpg

class StructuredLongTermMemory: def init(self, db_url: str): self.db_url = db_url self.pool = None

async def init(self):
    self.pool = await asyncpg.create_pool(self.db_url, min_size=2, max_size=10)
    async with self.pool.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS user_memory (
                id SERIAL PRIMARY KEY,
                user_id VARCHAR(128) NOT NULL,
                memory_key VARCHAR(256) NOT NULL,
                memory_value TEXT NOT NULL,
                category VARCHAR(64) DEFAULT 'general',
                updated_at TIMESTAMP DEFAULT NOW(),
                UNIQUE(user_id, memory_key)
            )
        """)

async def set_memory(self, user_id: str, key: str, value: str, category: str = "general"):
    async with self.pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO user_memory (user_id, memory_key, memory_value, category)
            VALUES (, , , )
            ON CONFLICT (user_id, memory_key)
            DO UPDATE SET memory_value = , category = , updated_at = NOW()
        """, user_id, key, value, category)

async def get_memory(self, user_id: str, key: str) -> Optional[str]:
    async with self.pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT memory_value FROM user_memory WHERE user_id =  AND memory_key = ",
            user_id, key
        )
        return row["memory_value"] if row else None

async def get_all_memories(self, user_id: str, category: str = None) -> list[dict]:
    async with self.pool.acquire() as conn:
        if category:
            rows = await conn.fetch(
                "SELECT memory_key, memory_value, category FROM user_memory WHERE user_id =  AND category = ",
                user_id, category
            )
        else:
            rows = await conn.fetch(
                "SELECT memory_key, memory_value, category FROM user_memory WHERE user_id = ",
                user_id
            )
        return [{"key": r["memory_key"], "value": r["memory_value"], "category": r["category"]} for r in rows]

`

会話圧縮：長時間会話シナリオの必須要件

3つの圧縮戦略

戦略 圧縮率 情報損失 レイテンシ 適用シナリオ
スライディングウィンドウ切り捨て 高 大 0ms 単純な会話
LLM要約圧縮 中 小 200-500ms 長時間会話
構造化抽出 低 極小 100-300ms タスク指向型会話

LLM要約圧縮

`python class ConversationCompressor: def init(self, llm_client, max_summary_tokens: int = 512): self.llm = llm_client self.max_summary_tokens = max_summary_tokens

async def compress(self, messages: list[dict], keep_recent: int = 4) -> list[dict]:
    if len(messages) <= keep_recent + 2:
        return messages

    to_compress = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    conversation_text = "\n".join(
        f"{m['role']}: {m['content'][:200]}" for m in to_compress
    )

    response = self.llm.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[{
            "role": "system",
            "content": "以下の会話履歴を簡潔な要約に圧縮してください。重要な事実、決定、ユーザー設定を保持してください。要約内容のみを出力し、説明は不要です。"
        }, {
            "role": "user",
            "content": conversation_text
        }],
        max_tokens=self.max_summary_tokens,
        temperature=0.0,
    )

    summary = response.choices[0].message.content
    return [
        {"role": "system", "content": f"以前の会話要約：{summary}"},
        *recent,
    ]

`

構造化情報抽出

`python class StructuredMemoryExtractor: def init(self, llm_client): self.llm = llm_client

async def extract(self, user_message: str, assistant_message: str) -> list[dict]:
    response = self.llm.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[{
            "role": "system",
            "content": """会話から長期記憶すべき構造化情報を抽出してください。

JSON配列を出力してください。各要素には key（記憶キー名）、value（記憶値）、category（分類：preference/fact/decision/task）を含めてください。

出力例： [{"key":"preferred_language","value":"Python","category":"preference"}]

記憶すべき情報がない場合は、空配列 [] を出力してください""" }, { "role": "user", "content": f"ユーザー: {user_message}\nアシスタント: {assistant_message}" }], max_tokens=256, temperature=0.0, response_format={"type": "json_object"}, )

    try:
        data = json.loads(response.choices[0].message.content)
        return data.get("memories", data.get("items", []))
    except json.JSONDecodeError:
        return []

`

本番デプロイ：Redis+Milvus記憶サービス

記憶サービス統合インターフェース

`python class AgentMemoryService: def init(self, redis_url: str, milvus_uri: str, pg_url: str, llm_client): self.session_memory = SessionMemory(redis_url) self.vector_memory = VectorLongTermMemory(milvus_uri) self.structured_memory = StructuredLongTermMemory(pg_url) self.compressor = ConversationCompressor(llm_client) self.extractor = StructuredMemoryExtractor(llm_client)

async def on_user_message(self, session_id: str, user_id: str, message: str) -> list[str]:
    vector_facts = await self.vector_memory.recall(user_id, message, top_k=3)
    structured = await self.structured_memory.get_all_memories(user_id)
    preference_facts = [f"{m['key']}: {m['value']}" for m in structured if m["category"] == "preference"]
    return vector_facts + preference_facts

async def on_assistant_message(self, session_id: str, user_id: str, user_msg: str, assistant_msg: str):
    await self.session_memory.save_message(session_id, "user", user_msg)
    await self.session_memory.save_message(session_id, "assistant", assistant_msg)

    memories = await self.extractor.extract(user_msg, assistant_msg)
    for mem in memories:
        await self.structured_memory.set_memory(user_id, mem["key"], mem["value"], mem["category"])

    await self.vector_memory.store(user_id, f"ユーザーの発言: {user_msg}", category="conversation")

async def build_prompt(self, session_id: str, user_id: str, current_message: str) -> list[dict]:
    memory_facts = await self.on_user_message(session_id, user_id, current_message)
    history = await self.session_memory.get_history(session_id, limit=50)
    messages = self.session_memory.build_context(
        system_prompt="あなたは記憶を持つAIアシスタントであり、ユーザーの設定や会話履歴を記憶できます。",
        memory_facts=memory_facts,
        conversation_history=history,
    )
    messages.append({"role": "user", "content": current_message})
    return messages

`

K8sデプロイ

yaml apiVersion: apps/v1 kind: Deployment metadata: name: agent-memory-service namespace: ai-agent spec: replicas: 2 selector: matchLabels: app: agent-memory-service template: spec: containers: - name: memory-service image: myregistry/agent-memory-service:v1.0 ports: - containerPort: 8080 resources: requests: cpu: "1" memory: 512Mi limits: cpu: "2" memory: 1Gi env: - name: REDIS_URL value: "redis://redis:6379" - name: MILVUS_URI value: "http://milvus-svc:19530" - name: PG_URL value: "postgresql://postgres:password@postgres:5432/agent_memory"

記憶システムパフォーマンスベンチマーク

操作 レイテンシ(P50) レイテンシ(P99) QPS
メッセージ保存(Redis) 1.2ms 3ms 8000
履歴取得(Redis) 2ms 5ms 6000
ベクトル保存(Milvus) 15ms 30ms 500
ベクトル再呼出(Milvus) 8ms 15ms 1000
構造化保存(PG) 3ms 8ms 3000
構造化クエリ(PG) 2ms 5ms 5000
会話圧縮(LLM) 350ms 800ms 50

まとめと関連記事

AIエージェントの記憶システムは、エージェント知能の中核インフラです。3層アーキテクチャ（ワーキングメモリ→短期記憶→長期記憶）は、コンテキストウィンドウの制限、セッション状態の喪失、セッション横断の知識忘却という3つの大きな問題を解決します。会話圧縮と構造化抽出は、長時間会話シナリオにおけるキーテクノロジーです。

開発のポイント：

  1. 3層記憶アーキテクチャ：ワーキングメモリ（コンテキストウィンドウ）→短期記憶（Redis）→長期記憶（Milvus+PG）
  2. トークン予算配分：System 6% + 記憶再呼出 12% + 会話履歴 62% + ツール 12% + 出力 6%
  3. 会話圧縮はLLM要約を優先し、タスク指向型会話では構造化抽出を使用
  4. 長期記憶の3つのモード：ベクトル検索（オープン型）、ナレッジグラフ（関係型）、構造化キーバリュー（完全一致）
  5. 本番デプロイはRedis+Milvus+PostgreSQLの3点セット

関連記事：

権威ある参考文献：

ブラウザローカルツールを無料で試す →

#AI Agent记忆管理#多轮对话记忆#长期记忆架构#Agent状态持久化#对话上下文压缩#2026