session-compression

AI セッション圧縮テクニック

概要

重要な情報を保持しながら、長い AI の会話をコンテキストウィンドウに収まるよう圧縮します。

セッション圧縮を利用すると、本番環境の AI アプリケーションで複数ターンの会話を効率的に管理できます。要約、埋め込みベースの検索、インテリジェントなコンテキスト管理によってトークン使用量を 70～95％削減します。パフォーマンスの低下を最小限に抑えながら、3～20 倍の圧縮率を実現できます。

主なメリット：

• コスト削減： 階層型メモリによりトークンコストを 80～90％削減
• パフォーマンス： 圧縮したコンテキストで応答速度が 2 倍に向上
• スケーラビリティ： 100 万トークンを超える会話に対応
• 品質： 精度の低下を 2％未満に抑えながら重要な情報を保持

使用するタイミング

セッション圧縮を使用するケース：

• 複数ターンの会話がコンテキストウィンドウの上限に近づいている（容量の 50％超）
• 長時間にわたるチャットセッション（カスタマーサポート、個別指導、コードアシスタント）
• トークンコストが無視できない規模になっている（大量利用のアプリケーション）
• 大きなコンテキストによって応答の遅延が増えている
• 複数のセッションにわたる会話履歴を管理している

使用しないケース：

• コンテキストに余裕で収まる短い会話（10 ターン未満）
• あらゆる詳細を原文どおりに保持する必要がある（法務、コンプライアンス）
• 単一ターンまたはステートレスなやり取り
• コンテキストウィンドウの使用率が 30％未満

最適な利用シーン：

• 50 ターン以上の会話を行うチャットボット
• 長時間の開発セッションを追跡する AI コードアシスタント
• 複数セッションにわたるチケット履歴を扱うカスタマーサポート
• 学習者の進捗を追跡する教育用チューター
• 複数日にわたる共同 AI ワークフロー

クイックスタート

LangChain での基本設定

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic
from anthropic import Anthropic

# Initialize Claude client
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    api_key="your-api-key"
)

# Setup memory with automatic summarization
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # Summarize when exceeding this
    return_messages=True
)

# Add conversation turns
memory.save_context(
    {"input": "What's session compression?"},
    {"output": "Session compression reduces conversation token usage..."}
)

# Retrieve compressed context
context = memory.load_memory_variables({})

段階的な圧縮パターン

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

class ProgressiveCompressor:
    def __init__(self, thresholds=[0.70, 0.85, 0.95]):
        self.thresholds = thresholds
        self.messages = []
        self.max_tokens = 200000  # Claude context window

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # Check if compression needed
        current_usage = self._estimate_tokens()
        usage_ratio = current_usage / self.max_tokens

        if usage_ratio >= self.thresholds[0]:
            self._compress(level=self._get_compression_level(usage_ratio))

    def _estimate_tokens(self):
        return sum(len(m["content"]) // 4 for m in self.messages)

    def _get_compression_level(self, ratio):
        for i, threshold in enumerate(self.thresholds):
            if ratio < threshold:
                return i
        return len(self.thresholds)

    def _compress(self, level: int):
        """Apply compression based on severity level."""
        if level == 1:  # 70% threshold: Light compression
            self._remove_redundant_messages()
        elif level == 2:  # 85% threshold: Medium compression
            self._summarize_old_messages(keep_recent=10)
        else:  # 95% threshold: Aggressive compression
            self._summarize_old_messages(keep_recent=5)

    def _remove_redundant_messages(self):
        """Remove duplicate or low-value messages."""
        # Implementation: Use semantic deduplication
        pass

    def _summarize_old_messages(self, keep_recent: int):
        """Summarize older messages, keep recent ones verbatim."""
        if len(self.messages) <= keep_recent:
            return

        # Messages to summarize
        to_summarize = self.messages[:-keep_recent]
        recent = self.messages[-keep_recent:]

        # Generate summary
        conversation_text = "\n\n".join([
            f"{m['role'].upper()}: {m['content']}"
            for m in to_summarize
        ])

        response = client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"Summarize this conversation:\n\n{conversation_text}"
            }]
        )

        # Replace old messages with summary
        summary = {
            "role": "system",
            "content": f"[Summary]\n{response.content[0].text}"
        }
        self.messages = [summary] + recent

# Usage
compressor = ProgressiveCompressor()

for i in range(100):
    compressor.add_message("user", f"Message {i}")
    compressor.add_message("assistant", f"Response {i}")

Anthropic Prompt Caching の使用（コストを 90％削減）

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# Build context with cache control
messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "Long conversation context here...",
                "cache_control": {"type": "ephemeral"}  # Cache this
            }
        ]
    },
    {
        "role": "assistant",
        "content": "Previous response..."
    },
    {
        "role": "user",
        "content": "New question"  # Not cached, changes frequently
    }
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=messages
)

# Cache hit reduces costs by 90% for cached content

---

主な概念

コンテキストウィンドウとトークン上限

コンテキストウィンドウ： LLM が 1 回のリクエストで処理できる最大トークン数（入力＋出力）。

現在の上限（2025 年）：

• Claude 3.5 Sonnet：200K トークン（約 15 万語、約 600 ページ）
• GPT-4 Turbo：128K トークン（約 9 万 6 千語、約 384 ページ）
• Gemini 1.5 Pro：2M トークン（約 150 万語、約 6000 ページ）

トークン数の目安：

• 英語：1 トークンあたり約 4 文字
• コード：1 トークンあたり約 3 文字
• 目安：1 トークン ≈ 0.75 語

圧縮が重要な理由：

• コスト： Claude Sonnet の料金は、入力／出力 100 万トークンあたり $3/$15
• レイテンシ： コンテキストが大きくなると処理時間が増加
• 品質： コンテキストが過剰だと、関連情報への注意が分散することがある

圧縮率

圧縮率 = 元のトークン数 / 圧縮後のトークン数

業界ベンチマーク：

• 抽出的要約：2～3 倍
• 抽象的要約：5～10 倍
• 階層型要約：20 倍以上
• LLMLingua（プロンプト圧縮）：精度の低下 1.5％で 20 倍
• KVzip（KV キャッシュ圧縮）：速度を 2 倍に向上させながら 3～4 倍

ユースケース別の目標圧縮率：

• カスタマーサポート：5～7 倍（詳細を保持）
• 一般的なチャット：8～12 倍（品質と効率のバランス）
• コードアシスタント：3～5 倍（技術的な正確性を保持）
• 長文書：15～20 倍（重要な知見を抽出）

段階的な圧縮しきい値

業界標準のパターン：

Context Usage    Action                     Technique
─────────────────────────────────────────────────────────
0-70%           No compression             Store verbatim
70-85%          Light compression          Remove redundancy
85-95%          Medium compression         Summarize old messages
95-100%         Aggressive compression     Hierarchical + RAG

実装ガイドライン：

• しきい値 70％： 重複・冗長なメッセージを削除し、意味的に重複を排除
• しきい値 85％： 20 ターンより前のメッセージを要約し、直近 10～15 ターンを保持
• しきい値 95％： 多段階の階層型要約とベクトルストアへのアーカイブ
• 緊急時（100％）： 重要度が最も低いメッセージを削除し、積極的に要約

---

圧縮テクニック

1. 要約テクニック

1.1 抽出的要約

主要な文やフレーズを変更せずに選び出します。

メリット： ハルシネーションがなく、高速で決定論的 デメリット： 圧縮率が限られる（2～3 倍）、つながりが不自然になることがある 最適な用途： 法務・コンプライアンス、短期的な圧縮

from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

def extractive_compress(messages: list, compression_ratio: float = 0.3):
    """Extract most important messages using TF-IDF scoring."""
    texts = [msg['content'] for msg in messages]

    # Calculate TF-IDF scores
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform(texts)
    scores = np.array(tfidf_matrix.sum(axis=1)).flatten()

    # Select top messages
    n_keep = max(1, int(len(messages) * compression_ratio))
    top_indices = sorted(np.argsort(scores)[-n_keep:])

    return [messages[i] for i in top_indices]

1.2 要約型要約

LLM を使用して、会話履歴を意味を保ちながら圧縮します。

メリット： 圧縮率が高い（5～10倍）、一貫性がある、情報を統合できる デメリット： ハルシネーションのリスク、コストが高い、決定性が低い 最適な用途： 一般的なチャット、カスタマーサポート、複数セッションにわたる継続性

from anthropic import Anthropic

def abstractive_compress(messages: list, client: Anthropic):
    """Generate semantic summary using Claude."""
    conversation_text = "\n\n".join([
        f"{msg['role'].upper()}: {msg['content']}"
        for msg in messages
    ])

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Summarize this conversation, preserving:
1. Key decisions made
2. Important context and facts
3. Unresolved questions
4. Action items

Conversation:
{conversation_text}

Summary (aim for 1/5 the original length):"""
        }]
    )

    return {
        "role": "assistant",
        "content": f"[Summary]\n{response.content[0].text}"
    }

1.3 階層型要約（マルチレベル）

要約をツリー構造でさらに要約します。

メリット： 非常に高い圧縮率（20倍以上）、100万トークンを超える会話に対応 デメリット： 実装が複雑、複数回の LLM 呼び出しが必要、情報の損失が蓄積する 最適な用途： 長期にわたる会話、複数セッションのアプリケーション

アーキテクチャ：

Level 0 (Raw):    [Msg1][Msg2][Msg3][Msg4][Msg5][Msg6][Msg7][Msg8]
Level 1 (Chunk):  [Summary1-2]  [Summary3-4]  [Summary5-6]  [Summary7-8]
Level 2 (Group):  [Summary1-4]              [Summary5-8]
Level 3 (Session): [Overall Session Summary]

from anthropic import Anthropic
from typing import List, Dict

class HierarchicalMemory:
    def __init__(self, client: Anthropic, chunk_size: int = 10):
        self.client = client
        self.chunk_size = chunk_size
        self.levels: List[List[Dict]] = [[]]  # Level 0 = raw messages

    def add_message(self, message: Dict):
        """Add message and trigger summarization if needed."""
        self.levels[0].append(message)

        if len(self.levels[0]) >= self.chunk_size * 2:
            self._summarize_level(0)

    def _summarize_level(self, level: int):
        """Summarize a level into the next higher level."""
        messages = self.levels[level]

        # Ensure next level exists
        while len(self.levels) <= level + 1:
            self.levels.append([])

        # Summarize first chunk
        chunk = messages[:self.chunk_size]
        summary = self._generate_summary(chunk, level)

        # Move to next level
        self.levels[level + 1].append(summary)
        self.levels[level] = messages[self.chunk_size:]

        # Recursively check if next level needs summarization
        if len(self.levels[level + 1]) >= self.chunk_size * 2:
            self._summarize_level(level + 1)

    def _generate_summary(self, messages: List[Dict], level: int) -> Dict:
        """Generate summary for a chunk."""
        conversation_text = "\n\n".join([
            f"{msg['role'].upper()}: {msg['content']}"
            for msg in messages
        ])

        response = self.client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"Summarize this Level {level} conversation chunk:\n\n{conversation_text}"
            }]
        )

        return {
            "role": "system",
            "content": f"[L{level+1} Summary] {response.content[0].text}",
            "level": level + 1
        }

    def get_context(self, max_tokens: int = 4000) -> List[Dict]:
        """Retrieve context within token budget."""
        context = []
        token_count = 0

        # Prioritize recent raw messages
        for msg in reversed(self.levels[0]):
            msg_tokens = len(msg['content']) // 4
            if token_count + msg_tokens > max_tokens * 0.6:
                break
            context.insert(0, msg)
            token_count += msg_tokens

        # Add summaries from higher levels
        for level in range(1, len(self.levels)):
            for summary in self.levels[level]:
                summary_tokens = len(summary['content']) // 4
                if token_count + summary_tokens > max_tokens:
                    break
                context.insert(0, summary)
                token_count += summary_tokens

        return context

学術論文：「Recursively Summarizing Enables Long-Term Dialogue Memory in Large Language Models」（arXiv:2308.15022）

1.4 ローリング要約（継続的）

スライディングウィンドウを使って、会話を継続的に圧縮します。

メリット： レイテンシが低い、トークン使用量が予測可能、シンプル デメリット： 初期の詳細が過度に圧縮される、情報を復元できない 最適な用途： リアルタイムチャット、ストリーミング会話

from anthropic import Anthropic

class RollingMemory:
    def __init__(self, client: Anthropic, window_size: int = 10, compress_threshold: int = 15):
        self.client = client
        self.window_size = window_size
        self.compress_threshold = compress_threshold
        self.rolling_summary = None
        self.recent_messages = []

    def add_message(self, message: dict):
        self.recent_messages.append(message)

        if len(self.recent_messages) >= self.compress_threshold:
            self._compress()

    def _compress(self):
        """Compress older messages into rolling summary."""
        messages_to_compress = self.recent_messages[:-self.window_size]

        parts = []
        if self.rolling_summary:
            parts.append(f"Existing summary:\n{self.rolling_summary}")

        parts.append("\nNew messages:\n" + "\n\n".join([
            f"{msg['role']}: {msg['content']}"
            for msg in messages_to_compress
        ]))

        response = self.client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": "\n".join(parts) + "\n\nUpdate the summary:"
            }]
        )

        self.rolling_summary = response.content[0].text
        self.recent_messages = self.recent_messages[-self.window_size:]

    def get_context(self):
        context = []
        if self.rolling_summary:
            context.append({
                "role": "system",
                "content": f"[Summary]\n{self.rolling_summary}"
            })
        context.extend(self.recent_messages)
        return context

2. 埋め込みベースのアプローチ

2.1 RAG（検索拡張生成）

会話全体をベクトルデータベースに保存し、関連するチャンクのみを取得します。

メリット： 非常に高い拡張性、情報の損失がない、高い関連性 デメリット： ベクトル DB インフラストラクチャが必要、検索レイテンシが発生する 最適な用途： ナレッジベース、大量の履歴を扱うカスタマーサポート

from anthropic import Anthropic
from openai import OpenAI
import chromadb

class RAGMemory:
    def __init__(self, anthropic_client: Anthropic, openai_client: OpenAI):
        self.anthropic = anthropic_client
        self.openai = openai_client

        # Initialize vector store
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection(
            name="conversation",
            metadata={"hnsw:space": "cosine"}
        )

        self.recent_messages = []
        self.recent_window = 5
        self.message_counter = 0

    def add_message(self, message: dict):
        """Add to recent memory and vector store."""
        self.recent_messages.append(message)

        if len(self.recent_messages) > self.recent_window:
            old_msg = self.recent_messages.pop(0)
            self._store_in_vectordb(old_msg)

    def _store_in_vectordb(self, message: dict):
        """Archive to vector database."""
        # Generate embedding
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=message['content']
        )

        self.collection.add(
            embeddings=[response.data[0].embedding],
            documents=[message['content']],
            metadatas=[{"role": message['role']}],
            ids=[f"msg_{self.message_counter}"]
        )
        self.message_counter += 1

    def retrieve_context(self, query: str, max_tokens: int = 4000):
        """Retrieve relevant context using RAG."""
        context = []
        token_count = 0

        # 1. Recent messages (short-term memory)
        for msg in self.recent_messages:
            context.append(msg)
            token_count += len(msg['content']) // 4

        # 2. Retrieve relevant historical context
        if token_count < max_tokens:
            query_embedding = self.openai.embeddings.create(
                model="text-embedding-3-small",
                input=query
            )

            n_results = min(10, (max_tokens - token_count) // 100)
            results = self.collection.query(
                query_embeddings=[query_embedding.data[0].embedding],
                n_results=n_results
            )

            for i, doc in enumerate(results['documents'][0]):
                if token_count + len(doc) // 4 > max_tokens:
                    break

                metadata = results['metadatas'][0][i]
                context.insert(0, {
                    "role": metadata['role'],
                    "content": f"[Retrieved] {doc}"
                })
                token_count += len(doc) // 4

        return context

ベクトルデータベースの選択肢：

• ChromaDB： 組み込み型、ローカル開発が容易
• Pinecone： マネージド型、p95 レイテンシは 50ms
• Weaviate： オープンソース、ハイブリッド検索
• Qdrant： 高パフォーマンス、ペイロードフィルタリング

2.2 ベクトル検索とクラスタリング

類似したメッセージをクラスタにまとめ、重心で表現します。

メリット： 冗長性を減らす、テーマを特定する、複数トピックに対応 デメリット： 十分なデータが必要、ニュアンスが失われる可能性がある 最適な用途： 複数トピックの会話、会議の要約

from sklearn.cluster import KMeans
from openai import OpenAI
import numpy as np

class ClusteredMemory:
    def __init__(self, openai_client: OpenAI, n_clusters: int = 5):
        self.client = openai_client
        self.n_clusters = n_clusters
        self.messages = []
        self.embeddings = []

    def add_messages(self, messages: list):
        for msg in messages:
            self.messages.append(msg)

            response = self.client.embeddings.create(
                model="text-embedding-3-small",
                input=msg['content']
            )
            self.embeddings.append(response.data[0].embedding)

    def compress_by_clustering(self):
        """Cluster messages and return representatives."""
        if len(self.messages) < self.n_clusters:
            return self.messages

        embeddings_array = np.array(self.embeddings)
        kmeans = KMeans(n_clusters=self.n_clusters, random_state=42)
        labels = kmeans.fit_predict(embeddings_array)

        # Select message closest to each centroid
        compressed = []
        for cluster_id in range(self.n_clusters):
            cluster_indices = np.where(labels == cluster_id)[0]
            centroid = kmeans.cluster_centers_[cluster_id]
            cluster_embeddings = embeddings_array[cluster_indices]
            distances = np.linalg.norm(cluster_embeddings - centroid, axis=1)
            closest_idx = cluster_indices[np.argmin(distances)]

            compressed.append({
                **self.messages[closest_idx],
                "cluster_id": int(cluster_id),
                "cluster_size": len(cluster_indices)
            })

        return compressed

2.3 セマンティック重複排除

冗長な情報を伝える、意味的に類似したメッセージを削除します。

メリット： 独自の内容を失わずに冗長性を減らせる デメリット： しきい値の調整が必要、計算量は O(n²) 最適な用途： FAQ システム、繰り返しの多い会話

from openai import OpenAI
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticDeduplicator:
    def __init__(self, openai_client: OpenAI, similarity_threshold: float = 0.85):
        self.client = openai_client
        self.threshold = similarity_threshold

    def deduplicate(self, messages: list):
        """Remove semantically similar messages."""
        if len(messages) <= 1:
            return messages

        # Generate embeddings
        embeddings = []
        for msg in messages:
            response = self.client.embeddings.create(
                model="text-embedding-3-small",
                input=msg['content']
            )
            embeddings.append(response.data[0].embedding)

        embeddings_array = np.array(embeddings)
        similarity_matrix = cosine_similarity(embeddings_array)

        # Mark unique messages
        keep_indices = []
        for i in range(len(messages)):
            is_unique = True
            for j in keep_indices:
                if similarity_matrix[i][j] > self.threshold:
                    is_unique = False
                    break

            if is_unique:
                keep_indices.append(i)

        return [messages[i] for i in keep_indices]

3. トークン効率化の戦略

3.1 メッセージの優先順位付け

重要度スコアを割り当て、優先度の高い内容のみを保持します。

メリット： 最も重要な情報を保持できる、基準が柔軟 デメリット： スコア付けがヒューリスティックベース、会話の流れが崩れる可能性がある 最適な用途： 重要度が混在する会話、ノイズのフィルタリング

import re

class MessagePrioritizer:
    def score_message(self, msg: dict, index: int, total: int) -> float:
        """Calculate composite importance score."""
        scores = []

        # Length score (longer = more info)
        scores.append(min(len(msg['content']) / 500, 1.0))

        # Question score
        if msg['role'] == 'user':
            scores.append(min(msg['content'].count('?') * 0.5, 1.0))

        # Entity score (capitalized words)
        entities = len(re.findall(r'\b[A-Z][a-z]+', msg['content']))
        scores.append(min(entities / 10, 1.0))

        # Recency score (linear decay)
        scores.append(index / max(total - 1, 1))

        # Role score
        scores.append(0.6 if msg['role'] == 'user' else 0.4)

        return sum(scores) / len(scores)

    def prioritize(self, messages: list, target_count: int):
        """Select top N messages by priority."""
        scored = [
            (msg, self.score_message(msg, i, len(messages)), i)
            for i, msg in enumerate(messages)
        ]

        scored.sort(key=lambda x: x[1], reverse=True)
        top_messages = scored[:target_count]
        top_messages.sort(key=lambda x: x[2])  # Restore chronological order

        return [msg for msg, score, idx in top_messages]

3.2 差分圧縮

連続するメッセージ間の変更点のみを保存します。

メリット： 増分的な変更に対して高い効率性 デメリット： 再構築のオーバーヘッド、すべてのコンテンツに適しているわけではない 最適な用途： 増分編集に対応するコードアシスタント

import difflib

class DeltaCompressor:
    def __init__(self):
        self.base_messages = []
        self.deltas = []

    def add_message(self, message: dict):
        if not self.base_messages:
            self.base_messages.append(message)
            return

        # Find most similar previous message
        last_msg = self.base_messages[-1]

        if last_msg['role'] == message['role']:
            # Calculate delta
            diff = list(difflib.unified_diff(
                last_msg['content'].splitlines(),
                message['content'].splitlines(),
                lineterm=''
            ))

            if len('\n'.join(diff)) < len(message['content']) * 0.7:
                # Store as delta if compression achieved
                self.deltas.append({
                    'base_index': len(self.base_messages) - 1,
                    'delta': diff,
                    'role': message['role']
                })
                return

        # Store as new base message
        self.base_messages.append(message)

    def reconstruct(self):
        """Reconstruct full conversation from bases + deltas."""
        messages = self.base_messages.copy()

        for delta_info in self.deltas:
            base_content = messages[delta_info['base_index']]['content']
            # Apply diff to reconstruct (simplified)
            reconstructed = base_content  # Full implementation would apply diff
            messages.append({
                'role': delta_info['role'],
                'content': reconstructed
            })

        return messages

4. LangChain のメモリタイプ

4.1 ConversationSummaryMemory

会話の進行に合わせて、自動的に要約します。

from langchain.memory import ConversationSummaryMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

memory = ConversationSummaryMemory(llm=llm)

# Add conversation
memory.save_context(
    {"input": "Hi, I'm working on a Python project"},
    {"output": "Great! How can I help with your Python project?"}
)

# Get summary
summary = memory.load_memory_variables({})
print(summary['history'])

メリット： 自動要約、シンプルな API デメリット： ターンごとに LLM の呼び出しが発生する 最適な用途： 中程度の長さの会話（20～50ターン）

4.2 ConversationSummaryBufferMemory

ハイブリッド方式：直近のメッセージはそのまま保持し、古いメッセージは要約します。

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # Summarize when exceeding
    return_messages=True
)

# Add conversation
for i in range(50):
    memory.save_context(
        {"input": f"Question {i}"},
        {"output": f"Answer {i}"}
    )

# Automatically keeps recent messages + summary of old
context = memory.load_memory_variables({})

メリット： 詳細さと圧縮率のバランスが最適 デメリット： トークン上限の調整が必要 最適な用途： ほとんどの本番アプリケーション

4.3 ConversationTokenBufferMemory

固定のトークン予算を維持し、上限を超えると最も古いメッセージを削除します。

from langchain.memory import ConversationTokenBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=2000
)

# Simple FIFO when token limit exceeded

メリット： トークン使用量が予測可能、シンプル デメリット： 古い情報が完全に失われる 最適な用途： 厳しい制限のあるリアルタイムチャット

4.4 VectorStoreRetrieverMemory

すべてのメッセージをベクトルデータベースに保存し、関連するものを取得します。

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# Automatically retrieves most relevant context

メリット： 会話の長さが無制限、意味検索 デメリット： ベクトル DB が必要、検索のオーバーヘッド 最適な用途： 長期にわたる会話、ナレッジベース

5. Anthropic 固有のパターン

5.1 プロンプトキャッシュ（コストを90％削減）

静的コンテキストをキャッシュして、トークンコストを削減します。

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# Long conversation context
conversation_history = [
    {"role": "user", "content": "Message 1"},
    {"role": "assistant", "content": "Response 1"},
    # ... many more messages
]

# Mark context for caching
messages = []
for i, msg in enumerate(conversation_history[:-1]):
    content = msg['content']

    # Add cache control to last context message
    if i == len(conversation_history) - 2:
        messages.append({
            "role": msg['role'],
            "content": [
                {
                    "type": "text",
                    "text": content,
                    "cache_control": {"type": "ephemeral"}
                }
            ]
        })
    else:
        messages.append(msg)

# Add new user message (not cached)
messages.append(conversation_history[-1])

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=messages
)

# Subsequent calls with same cached context cost 90% less

キャッシュのTTL： 5分 削減効果： キャッシュされたトークンのコストを90％削減 制限： リクエストあたり最大4つのキャッシュブレークポイント ベストプラクティス：

• 現在のクエリではなく、会話履歴をキャッシュします
• コンテキストが大きく変わったら、キャッシュを更新します
• 最大限の効率を得るため、要約と組み合わせます

5.2 圧縮計画のための拡張思考

拡張思考を使って、最適な圧縮戦略を計画します。

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    messages=[{
        "role": "user",
        "content": f"""Analyze this conversation and recommend compression:

{conversation_text}

Current token count: {current_tokens}
Target: {target_tokens}
Required compression: {compression_ratio}x

Recommend optimal strategy."""
    }]
)

# Access thinking process
thinking_content = [
    block for block in response.content
    if block.type == "thinking"
]

# Get compression recommendation
recommendation = response.content[-1].text

---

本番環境でのパターン

チェックポイントと永続化

復旧と再開に備えて、圧縮状態を保存します。

import json
import pickle
from pathlib import Path

class PersistentMemory:
    def __init__(self, checkpoint_dir: str = "./checkpoints"):
        self.checkpoint_dir = Path(checkpoint_dir)
        self.checkpoint_dir.mkdir(exist_ok=True)
        self.memory = []
        self.summary = None

    def save_checkpoint(self, session_id: str):
        """Save current memory state."""
        checkpoint = {
            'messages': self.memory,
            'summary': self.summary,
            'timestamp': time.time()
        }

        checkpoint_file = self.checkpoint_dir / f"{session_id}.json"
        with open(checkpoint_file, 'w') as f:
            json.dump(checkpoint, f, indent=2)

    def load_checkpoint(self, session_id: str):
        """Load memory state from checkpoint."""
        checkpoint_file = self.checkpoint_dir / f"{session_id}.json"

        if checkpoint_file.exists():
            with open(checkpoint_file, 'r') as f:
                checkpoint = json.load(f)

            self.memory = checkpoint['messages']
            self.summary = checkpoint.get('summary')
            return True

        return False

    def auto_checkpoint(self, session_id: str, interval: int = 10):
        """Automatically save every N messages."""
        if len(self.memory) % interval == 0:
            self.save_checkpoint(session_id)

ワークフローの再開

セッションをまたいで会話を続けます。

from anthropic import Anthropic
import json

class ResumableConversation:
    def __init__(self, client: Anthropic, session_id: str):
        self.client = client
        self.session_id = session_id
        self.memory = self._load_or_create()

    def _load_or_create(self):
        """Load existing session or create new."""
        try:
            with open(f'sessions/{self.session_id}.json', 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {
                'messages': [],
                'summary': None,
                'created_at': time.time()
            }

    def add_turn(self, user_message: str):
        """Add user message and get response."""
        # Add user message
        self.memory['messages'].append({
            'role': 'user',
            'content': user_message
        })

        # Build context (with compression)
        context = self._build_context()

        # Get response
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=context + [{
                'role': 'user',
                'content': user_message
            }]
        )

        # Save response
        assistant_message = response.content[0].text
        self.memory['messages'].append({
            'role': 'assistant',
            'content': assistant_message
        })

        # Compress if needed
        if len(self.memory['messages']) > 20:
            self._compress()

        # Save state
        self._save()

        return assistant_message

    def _build_context(self):
        """Build context with compression."""
        context = []

        # Add summary if exists
        if self.memory['summary']:
            context.append({
                'role': 'system',
                'content': f"[Previous conversation summary]\n{self.memory['summary']}"
            })

        # Add recent messages
        context.extend(self.memory['messages'][-10:])

        return context

    def _compress(self):
        """Compress older messages."""
        if len(self.memory['messages']) < 15:
            return

        # Messages to summarize
        to_summarize = self.memory['messages'][:-10]

        # Generate summary
        conversation_text = "\n\n".join([
            f"{msg['role']}: {msg['content']}"
            for msg in to_summarize
        ])

        response = self.client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=500,
            messages=[{
                'role': 'user',
                'content': f"Summarize this conversation:\n\n{conversation_text}"
            }]
        )

        # Update memory
        self.memory['summary'] = response.content[0].text
        self.memory['messages'] = self.memory['messages'][-10:]

    def _save(self):
        """Save session to disk."""
        with open(f'sessions/{self.session_id}.json', 'w') as f:
            json.dump(self.memory, f, indent=2)

# Usage
client = Anthropic(api_key="your-api-key")
conversation = ResumableConversation(client, session_id="user123_session1")

# Continue across multiple sessions
response1 = conversation.add_turn("What's Python?")
# ... later session
response2 = conversation.add_turn("Show me an example")  # Remembers context

ハイブリッドアプローチ（ベストプラクティス）

最適な結果を得るため、複数の手法を組み合わせます。

from anthropic import Anthropic
from openai import OpenAI
import chromadb

class HybridMemorySystem:
    """
    Combines:
    - Rolling summarization (short-term compression)
    - RAG retrieval (long-term memory)
    - Prompt caching (cost optimization)
    - Progressive compression (adaptive behavior)
    """

    def __init__(self, anthropic_client: Anthropic, openai_client: OpenAI):
        self.anthropic = anthropic_client
        self.openai = openai_client

        # Recent messages (verbatim)
        self.recent_messages = []
        self.recent_window = 10

        # Rolling summary
        self.rolling_summary = None

        # Vector store (long-term)
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection(name="memory")
        self.message_counter = 0

        # Compression thresholds
        self.thresholds = {
            'light': 0.70,    # Start basic compression
            'medium': 0.85,   # Aggressive summarization
            'heavy': 0.95     # Emergency measures
        }

    def add_message(self, message: dict):
        """Add message with intelligent compression."""
        self.recent_messages.append(message)

        # Check compression needs
        usage_ratio = self._estimate_usage()

        if usage_ratio >= self.thresholds['heavy']:
            self._emergency_compress()
        elif usage_ratio >= self.thresholds['medium']:
            self._medium_compress()
        elif usage_ratio >= self.thresholds['light']:
            self._light_compress()

    def _light_compress(self):
        """Remove redundancy, archive to vector store."""
        if len(self.recent_messages) > self.recent_window * 1.5:
            # Archive oldest to vector store
            to_archive = self.recent_messages[:5]
            for msg in to_archive:
                self._archive_to_vectorstore(msg)

            self.recent_messages = self.recent_messages[5:]

    def _medium_compress(self):
        """Generate rolling summary, aggressive archival."""
        if len(self.recent_messages) > self.recent_window:
            # Summarize older messages
            to_summarize = self.recent_messages[:-self.recent_window]

            summary_text = "\n\n".join([
                f"{msg['role']}: {msg['content']}"
                for msg in to_summarize
            ])

            if self.rolling_summary:
                summary_text = f"Existing: {self.rolling_summary}\n\nNew: {summary_text}"

            response = self.anthropic.messages.create(
                model="claude-3-5-haiku-20241022",
                max_tokens=400,
                messages=[{
                    'role': 'user',
                    'content': f"Update summary:\n{summary_text}"
                }]
            )

            self.rolling_summary = response.content[0].text

            # Archive all summarized messages
            for msg in to_summarize:
                self._archive_to_vectorstore(msg)

            self.recent_messages = self.recent_messages[-self.recent_window:]

    def _emergency_compress(self):
        """Extreme compression for near-limit situations."""
        # Keep only 5 most recent messages
        to_archive = self.recent_messages[:-5]
        for msg in to_archive:
            self._archive_to_vectorstore(msg)

        self.recent_messages = self.recent_messages[-5:]

        # Compress summary further if needed
        if self.rolling_summary and len(self.rolling_summary) > 1000:
            response = self.anthropic.messages.create(
                model="claude-3-5-haiku-20241022",
                max_tokens=200,
                messages=[{
                    'role': 'user',
                    'content': f"Create ultra-concise summary:\n{self.rolling_summary}"
                }]
            )
            self.rolling_summary = response.content[0].text

    def _archive_to_vectorstore(self, message: dict):
        """Store in vector database for retrieval."""
        embedding_response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=message['content']
        )

        self.collection.add(
            embeddings=[embedding_response.data[0].embedding],
            documents=[message['content']],
            metadatas=[{'role': message['role']}],
            ids=[f"msg_{self.message_counter}"]
        )
        self.message_counter += 1

    def get_context(self, current_query: str, max_tokens: int = 8000):
        """Build optimal context for current query."""
        context = []
        token_count = 0

        # 1. Add rolling summary (if exists)
        if self.rolling_summary:
            summary_msg = {
                'role': 'system',
                'content': [
                    {
                        'type': 'text',
                        'text': f"[Conversation Summary]\n{self.rolling_summary}",
                        'cache_control': {'type': 'ephemeral'}  # Cache it
                    }
                ]
            }
            context.append(summary_msg)
            token_count += len(self.rolling_summary) // 4

        # 2. Retrieve relevant historical context (RAG)
        if token_count < max_tokens * 0.3:
            query_embedding = self.openai.embeddings.create(
                model="text-embedding-3-small",
                input=current_query
            )

            results = self.collection.query(
                query_embeddings=[query_embedding.data[0].embedding],
                n_results=5
            )

            for i, doc in enumerate(results['documents'][0]):
                if token_count + len(doc) // 4 > max_tokens * 0.3:
                    break

                metadata = results['metadatas'][0][i]
                context.append({
                    'role': metadata['role'],
                    'content': f"[Retrieved] {doc}"
                })
                token_count += len(doc) // 4

        # 3. Add recent messages verbatim
        for msg in self.recent_messages:
            if token_count + len(msg['content']) // 4 > max_tokens * 0.8:
                break
            context.append(msg)
            token_count += len(msg['content']) // 4

        return context

    def _estimate_usage(self):
        """Estimate current context window usage."""
        total_tokens = 0

        if self.rolling_summary:
            total_tokens += len(self.rolling_summary) // 4

        for msg in self.recent_messages:
            total_tokens += len(msg['content']) // 4

        return total_tokens / 200000  # Claude Sonnet context window

# Usage
anthropic_client = Anthropic(api_key="your-anthropic-key")
openai_client = OpenAI(api_key="your-openai-key")

memory = HybridMemorySystem(anthropic_client, openai_client)

# Add messages over time
for i in range(1000):
    memory.add_message({
        'role': 'user' if i % 2 == 0 else 'assistant',
        'content': f"Message {i} with some content..."
    })

# Retrieve optimized context
current_query = "What did we discuss about pricing?"
context = memory.get_context(current_query)

# Use with Claude
response = anthropic_client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=context + [{
        'role': 'user',
        'content': current_query
    }]
)

---

パフォーマンスベンチマーク

圧縮効率

手法	圧縮率	品質の低下	レイテンシ	コストへの影響
抽出型	2-3x	<1%	<10ms	なし
抽象型	5-10x	2-5%	1-2s	+$0.001/turn
階層型	20x+	5-8%	2-5s	+$0.003/turn
LLMLingua	20x	1.5%	500ms	なし
RAG	可変	<1%	100-300ms	+$0.0005/turn
プロンプトキャッシュ	該当なし	0%	0ms	-90%

ユースケース別のトークン削減量

カスタマーサポート（50ターンの会話）：

• 圧縮なし：約8,000トークン／リクエスト
• ローリング要約：約2,000トークン／リクエスト（75％削減）
• ハイブリッド（RAG＋要約）：約1,500トークン／リクエスト（81％削減）

コードアシスタント（100ターンのセッション）：

• 圧縮なし：約25,000トークン／リクエスト
• 階層型：約5,000トークン／リクエスト（80％削減）
• ハイブリッド＋キャッシュ：実効約1,000トークン／リクエスト（コストを96％削減）

教育チューター（複数セッション）：

• 圧縮なし：コンテキストウィンドウを超過
• RAG＋要約：約3,000トークン／リクエスト
• セッション長を無制限にできます

コスト分析

例：Claude Sonnet の料金（入力100万トークンあたり$3、出力100万トークンあたり$15）

1,000件の会話、各50ターン：

• 圧縮なし：

  • リクエストあたり平均8Kトークン × 50Kリクエスト = 400Mトークン
  • コスト：$1,200

• ローリング要約を使用：

  • リクエストあたり平均2Kトークン × 50Kリクエスト = 100Mトークン
  • 要約のオーバーヘッド：+10Mトークン
  • コスト：$330（72％削減）

• ハイブリッドシステム＋キャッシュを使用：

  • 最初のターン：2Kトークン（キャッシュなし）
  • それ以降：実効200トークン（キャッシュヒット率90％）
  • 合計：約15M実効トークン
  • コスト：$45（96％削減）

---

ツールのおすすめ

メモリ管理ツール

Mem0（本番環境向けのおすすめ）

最適な用途： 最小限のコードで実現するハイブリッドメモリシステム

from mem0 import MemoryClient

client = MemoryClient(api_key="your-mem0-key")

# Automatically handles compression, summarization, RAG
memory = client.create_memory(
    user_id="user123",
    messages=[
        {"role": "user", "content": "I'm working on a Python project"},
        {"role": "assistant", "content": "Great! What kind of project?"}
    ]
)

# Retrieve relevant context
context = client.get_memory(
    user_id="user123",
    query="What programming language am I using?"
)

機能：

• 階層型要約の自動化
• RAG 検索機能を内蔵
• マルチユーザーセッション管理
• 分析ダッシュボード

料金： メモリ操作1K回あたり$0.40

Zep

最適な用途： 低レイテンシの本番環境へのデプロイ**

from zep_python import ZepClient

client = ZepClient(api_key="your-zep-key")

# Add to session
client.memory.add_memory(
    session_id="session123",
    messages=[
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ]
)

# Auto-summarized retrieval
memory = client.memory.get_memory(session_id="session123")

機能：

• 検索レイテンシ <100ms
• 事実の自動抽出
• エンティティ認識
• セッション管理

料金： オープンソース（セルフホスト可能）または操作1K回あたり$0.50（クラウド）

ChromaDB

最適な用途： セルフホスト型のベクトルストレージ**

import chromadb

client = chromadb.Client()
collection = client.create_collection("conversations")

# Store embeddings
collection.add(
    documents=["Message content"],
    embeddings=[[0.1, 0.2, ...]],
    ids=["msg1"]
)

# Retrieve
results = collection.query(
    query_embeddings=[[0.1, 0.2, ...]],
    n_results=5
)

機能：

• 完全なオープンソース
• 組み込み型またはクライアント・サーバー型
• ローカル開発が高速

料金： 無料（セルフホスト）

LangChain

最適な用途： 迅速なプロトタイピングと実験**

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=2000)

機能：

• 複数のメモリタイプ
• フレームワークとの統合
• 充実したドキュメント

料金： 無料（LLM API の利用料金は別途かかります）

圧縮ライブラリ

LLMLingua

最適な用途： 品質の低下を最小限に抑えた極端な圧縮**

from llmlingua import PromptCompressor

compressor = PromptCompressor()

compressed = compressor.compress_prompt(
    context="Long conversation history...",
    instruction="Current user query",
    target_token=500
)

# Achieves 20x compression with 1.5% accuracy loss

機能：

• 20倍の圧縮率
• 品質の低下は2％未満
• 高速な推論（<500ms）

料金： 無料（オープンソース）

---

ユースケースとパターン

チャットボット（カスタマーサポート）

要件：

• 複数ターンの会話（50～100ターン）
• 顧客のコンテキストを保持する
• 高速な応答時間
• コスト効率が高い

推奨されるアプローチ：

• ConversationSummaryBufferMemory（LangChain）
• 70％のしきい値：セマンティック重複排除
• 85％のしきい値：ローリング要約
• 頻出パターンに対するプロンプトキャッシュ

実装：

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,
    return_messages=True
)

# Add customer conversation
for turn in customer_conversation:
    memory.save_context(
        {"input": turn['customer_message']},
        {"output": turn['agent_response']}
    )

# Retrieve compressed context
context = memory.load_memory_variables({})

コードアシスタント

要件：

• 長時間の開発セッション（100ターン以上）
• 技術的な詳細を保持する
• 大きなコードブロックを処理する
• 段階的な変更を追跡する

推奨されるアプローチ：

• 全体的なコンテキストには階層的要約を使用する
• 特定のコード参照にはRAG検索を使用する
• 反復的な編集には差分圧縮を使用する
• システムプロンプトにはプロンプトキャッシュを使用する

実装：

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

class CodeAssistantMemory:
    def __init__(self):
        self.hierarchy = HierarchicalMemory(client, chunk_size=15)
        self.rag = RAGMemory(anthropic_client=client, openai_client=openai_client)
        self.deltas = DeltaCompressor()

    def add_interaction(self, code_change: dict):
        # Store in hierarchy
        self.hierarchy.add_message({
            'role': 'user',
            'content': code_change['description']
        })

        # Store in RAG for retrieval
        self.rag.add_message(code_change)

        # Store as delta if incremental
        if code_change.get('is_incremental'):
            self.deltas.add_message(code_change)

    def get_context(self, current_query: str):
        # Combine hierarchical summary + RAG retrieval
        summary_context = self.hierarchy.get_context(max_tokens=2000)
        rag_context = self.rag.retrieve_context(current_query, max_tokens=2000)

        return summary_context + rag_context

教育チューター

要件：

• 複数セッションにわたる追跡
• 生徒の進捗の永続化
• パーソナライズされたコンテキスト検索
• 長期的な知識保持

推奨されるアプローチ：

• 複数セッションにはVectorStoreRetrieverMemoryを使用する
• 生徒の知識を事実として抽出する
• セッションをまたいで段階的に圧縮する
• 再開可能な会話

実装：

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

class TutorMemory:
    def __init__(self, student_id: str):
        self.student_id = student_id

        # Vector store for all sessions
        embeddings = OpenAIEmbeddings()
        vectorstore = Chroma(
            collection_name=f"student_{student_id}",
            embedding_function=embeddings
        )

        self.memory = VectorStoreRetrieverMemory(
            retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
        )

    def add_lesson_content(self, lesson: dict):
        """Add lesson interaction to student memory."""
        self.memory.save_context(
            {"input": lesson['topic']},
            {"output": lesson['explanation']}
        )

    def get_student_context(self, current_topic: str):
        """Retrieve relevant past lessons for current topic."""
        return self.memory.load_memory_variables({
            "prompt": current_topic
        })

---

ベストプラクティス

1. ユースケースに合った手法を選ぶ

• 短い会話（20ターン未満）： 圧縮は不要です
• 中程度の会話（20～50ターン）： ConversationSummaryBufferMemory
• 長い会話（50～100ターン）： 階層的要約またはローリング要約
• 非常に長い会話（100ターン以上）： ハイブリッド（RAG＋要約＋キャッシュ）
• 複数セッション： VectorStoreRetrieverMemoryまたはMem0

2. 段階的な圧縮を実装する

最初から積極的に圧縮しないでください。しきい値を使用します。

• 0～70％：そのまま保存
• 70～85％：軽度の圧縮（重複排除）
• 85～95％：中程度の圧縮（要約）
• 95～100％：積極的な圧縮（階層化）

3. 手法を組み合わせる

単一の手法だけを使うアプローチは最適ではありません。本番環境で最適なシステムでは、次の手法を使用します。

• ローリング要約（短期）
• RAG検索（長期）
• プロンプトキャッシュ（コスト最適化）
• セマンティック重複排除（冗長性の除去）

4. 品質指標を監視する

圧縮の影響を追跡します。

• 応答の関連性スコア
• 情報保持率
• ユーザー満足度指標
• トークン使用量の削減

5. プロンプトキャッシュを戦略的に使う

安定したコンテンツをキャッシュします。

• 会話の要約
• システムプロンプト
• ナレッジベースのコンテキスト
• ユーザープロファイル

頻繁に変化するコンテンツはキャッシュしないでください。

• 現在のユーザークエリ
• リアルタイムデータ
• セッション固有の状態

6. チェックポイントを実装する

次の目的のために圧縮状態を保存します。

• 障害からの復旧
• 複数セッションにわたる継続性
• 分析とデバッグ
• 異なる戦略のA/Bテスト

7. 圧縮パラメーターを調整する

次の項目をテストして最適化します。

• 要約のトークン上限
• 圧縮しきい値
• 検索結果の件数
• キャッシュのTTL
• 階層化のチャンクサイズ

8. エッジケースに対処する

次のケースに備えます。

• 非常に長いメッセージ（個別に分割または圧縮する）
• コードブロック（書式を保持する）
• 多言語コンテンツ
• 急速に変化するコンテキスト

---

トラブルシューティング

問題：要約で重要な情報が失われる

解決策：

• 圧縮率を下げる（圧縮を緩やかにする）
• 重要度スコアリングを実装して、重要なメッセージを保持する
• 重要なセクションには抽出型要約を使用する
• 要約のトークン予算を増やす

問題：検索で無関係なコンテキストが返される

解決策：

• 埋め込みモデルの品質を高める
• メタデータフィルタリング（タイムスタンプ、トピック）を追加する
• 類似度のしきい値を調整する
• ハイブリッド検索（セマンティック検索＋キーワード検索）を使用する

問題：圧縮によるレイテンシーが高い

解決策：

• 非同期で圧縮する（バックグラウンドタスク）
• 要約にはより高速なモデルを使用する（SonnetではなくHaiku）
• 要約をより積極的にキャッシュする
• 圧縮の頻度を下げる

問題：会話がコンテキストウィンドウを超過する

解決策：

• 階層型圧縮を実装します
• ベクトルデータベースへのアーカイブをより積極的に行います
• より高い圧縮率を使用します
• より大きなコンテキストウィンドウを持つモデルへの切り替えを検討します

問題：圧縮してもコストが高い

解決策：

• プロンプトキャッシュを実装します
• 要約にはより安価なモデル（Haiku）を使用します
• 要約処理をバッチ化します
• 要約の頻度を減らします

問題：会話の継続性が失われる

解決策：

• 直近のメッセージウィンドウを広げます
• すべてのリクエストに要約を含めます
• より説明的な要約を使用します
• コンテキスト注入によるセッション再開を実装します

---

高度なトピック

ストリーミング圧縮

会話の進行に合わせてリアルタイムに圧縮します：

async def streaming_compress(messages: list):
    """Compress while streaming responses."""
    compressor = ProgressiveCompressor()

    async for message in conversation_stream:
        compressor.add_message(message)

        # Compression happens asynchronously
        if compressor.should_compress():
            asyncio.create_task(compressor.compress_async())

    return compressor.get_context()

マルチユーザーのセッション管理

共有コンテキストを使用して、並行する会話を処理します：

class MultiUserMemory:
    def __init__(self):
        self.user_sessions = {}

    def get_or_create_session(self, user_id: str):
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = HybridMemorySystem(...)
        return self.user_sessions[user_id]

    def cleanup_inactive_sessions(self, timeout: int = 3600):
        """Remove sessions inactive for > timeout seconds."""
        current_time = time.time()
        inactive = [
            user_id for user_id, session in self.user_sessions.items()
            if current_time - session.last_activity > timeout
        ]

        for user_id in inactive:
            self._archive_session(user_id)
            del self.user_sessions[user_id]

カスタム重要度スコアリング

メッセージの重要度をスコアリングする ML モデルをトレーニングします：

from transformers import pipeline

class MLImportanceScorer:
    def __init__(self):
        # Use pre-trained classifier or fine-tune on your data
        self.classifier = pipeline(
            "text-classification",
            model="your-importance-model"
        )

    def score(self, message: dict) -> float:
        """Score message importance (0-1)."""
        result = self.classifier(message['content'])
        return result[0]['score']

コンテキストウィンドウの利用率最適化

トークン予算内で情報密度を最大化します：

def optimize_context_allocation(
    summary_tokens: int,
    recent_tokens: int,
    retrieval_tokens: int,
    max_tokens: int
):
    """
    Optimal allocation (empirically tested):
    - 20% summary
    - 50% recent messages
    - 30% retrieved context
    """
    return {
        'summary': int(max_tokens * 0.20),
        'recent': int(max_tokens * 0.50),
        'retrieval': int(max_tokens * 0.30)
    }

---

今後の展望

新たな手法（2025年以降）

1. 無限アテンション機構

• 1,000万トークンを超えるコンテキストウィンドウを持つモデル（Gemini 1.5、将来の Claude）
• 圧縮の必要性は減りますが、コストの懸念は解消されません

2. 学習型圧縮モデル

• 会話を最適に圧縮するようトレーニングされたニューラルネットワーク
• トークン数を最小限に抑えながら、意味を維持します
• 例：LLMLingua v2、PromptCompressor

3. マルチモーダルセッション圧縮

• 画像、音声、動画を含む会話を圧縮します
• モダリティをまたぐコンテキストの関係を維持します

4. フェデレーテッドメモリシステム

• 複数のメモリストアにまたがる分散型圧縮
• 機密性の高い会話向けのプライバシー保護圧縮

5. 適応型圧縮戦略

• ユーザーやドメインごとに最適な圧縮を学習する、RL ベースのシステム
• 会話の重要度に応じた動的なしきい値調整

---

参考資料

学術論文

• 「Recursively Summarizing Enables Long-Term Dialogue Memory」（arXiv:2308.15022）
• 「LLMLingua: Compressing Prompts for Accelerated Inference」（arXiv:2310.05736）
• 「Lost in the Middle: How Language Models Use Long Contexts」（arXiv:2307.03172）

ドキュメント

• Anthropic のプロンプトキャッシュ
• LangChain のメモリ
• OpenAI の関数呼び出し

ツール

• Mem0 - マネージドメモリサービス
• Zep - 高速メモリレイヤー
• LLMLingua - プロンプト圧縮
• ChromaDB - ベクトルデータベース

---

最終更新日： 2025-11-30 バージョン： 1.0.0 ライセンス： MIT