AI セッション圧縮技術

要約

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

セッション圧縮により、要約、埋め込みベースの検索、インテリジェントなコンテキスト管理を通じて、トークン使用量を70～95%削減しながらマルチターン会話を効率的に管理できます。最小限のパフォーマンス低下で3～20倍の圧縮率を達成します。

主な利点:

• コスト削減: 階層的メモリにより80～90%のトークンコスト削減
• パフォーマンス: 圧縮されたコンテキストで2倍高速なレスポンス
• スケーラビリティ: 1Mトークンを超える会話に対応
• 品質: 2%未満の精度低下で重要情報を保持

使用すべき場合

以下の場合にセッション圧縮を使用してください:

• マルチターン会話がコンテキストウィンドウ上限に近づいている（容量の50%以上）
• 長時間実行されるチャットセッション（カスタマーサポート、チュートリアル、コード支援）
• トークンコストが大きい場合（高容量アプリケーション）
• 大規模なコンテキストによりレスポンスレイテンシが増加
• 複数のセッションにまたがる会話履歴の管理

使用しないケース:

• コンテキストに容易に収まる短い会話（10ターン未満）
• すべての詳細を逐語的に保持する必要がある場合（法務、コンプライアンス）
• シングルターンまたはステートレスな相互作用
• コンテキストウィンドウ使用率が30%未満

理想的なシナリオ:

• 50ターン以上の会話を持つチャットボット
• 長い開発セッションを追跡するAIコードアシスタント
• マルチセッションのチケット履歴を持つカスタマーサポート
• 学生の進捗を追跡する教育用チューター
• 複数日にまたがるコラボレーティブなAIワークフロー

クイックスタート

LangChainによる基本的な設定

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

# Claude クライアントを初期化
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    api_key="your-api-key"
)

# 自動要約機能付きメモリをセットアップ
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # これを超えると要約
    return_messages=True
)

# 会話ターンを追加
memory.save_context(
    {"input": "セッション圧縮とは何ですか?"},
    {"output": "セッション圧縮は会話のトークン使用量を削減します..."}
)

# 圧縮されたコンテキストを取得
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 コンテキストウィンドウ

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

        # 圧縮が必要かどうかをチェック
        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):
        """圧縮レベルに基づいて圧縮を適用。"""
        if level == 1:  # 70% しきい値: 軽い圧縮
            self._remove_redundant_messages()
        elif level == 2:  # 85% しきい値: 中程度の圧縮
            self._summarize_old_messages(keep_recent=10)
        else:  # 95% しきい値: 積極的な圧縮
            self._summarize_old_messages(keep_recent=5)

    def _remove_redundant_messages(self):
        """重複または低価値のメッセージを削除。"""
        # 実装: セマンティック重複排除を使用
        pass

    def _summarize_old_messages(self, keep_recent: int):
        """古いメッセージを要約し、最近のものはそのまま保持。"""
        if len(self.messages) <= keep_recent:
            return

        # 要約するメッセージ
        to_summarize = self.messages[:-keep_recent]
        recent = self.messages[-keep_recent:]

        # 要約を生成
        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"この会話を要約してください:\n\n{conversation_text}"
            }]
        )

        # 古いメッセージを要約で置き換え
        summary = {
            "role": "system",
            "content": f"[要約]\n{response.content[0].text}"
        }
        self.messages = [summary] + recent

# 使用方法
compressor = ProgressiveCompressor()

for i in range(100):
    compressor.add_message("user", f"メッセージ {i}")
    compressor.add_message("assistant", f"レスポンス {i}")

Anthropic プロンプトキャッシング（90%コスト削減）

from anthropic import Anthropic

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

# キャッシュ制御付きでコンテキストを構築
messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "長い会話コンテキスト...",
                "cache_control": {"type": "ephemeral"}  # これをキャッシュ
            }
        ]
    },
    {
        "role": "assistant",
        "content": "以前のレスポンス..."
    },
    {
        "role": "user",
        "content": "新しい質問"  # キャッシュされない、頻繁に変更
    }
]

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

# キャッシュヒットでキャッシュされたコンテンツのコスト90%削減

---

コアコンセプト

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

コンテキストウィンドウ: LLMが単一リクエストで処理できる最大トークン数（入力+出力）。

現在の上限（2025年）:

• Claude 3.5 Sonnet: 200Kトークン（約150K語、約600ページ）
• GPT-4 Turbo: 128Kトークン（約96K語、約384ページ）
• Gemini 1.5 Pro: 2Mトークン（約1.5M語、約6000ページ）

トークン推定:

• 英語: ~4文字あたり1トークン
• コード: ~3文字あたり1トークン
• 経験則: 1トークン ≈ 0.75語

圧縮が重要な理由:

• コスト: Claude Sonnet は 1Mトークンあたり $3/$15（入力/出力）
• レイテンシ: より大きなコンテキストは処理時間を増加
• 品質: 過度なコンテキストは関連情報への注意を希釈化

圧縮率

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

業界ベンチマーク:

• 抽出的要約: 2-3倍
• 抽象的要約: 5-10倍
• 階層的要約: 20倍以上
• LLMLingua（プロンプト圧縮）: 1.5%精度低下で20倍
• KVzip（KVキャッシュ圧縮）: 2倍速度改善で3-4倍

ユースケース別の目標比率:

• カスタマーサポート: 5-7倍（詳細を保持）
• 一般チャット: 8-12倍（品質と効率のバランス）
• コード支援: 3-5倍（技術的正確性を保持）
• 長いドキュメント: 15-20倍（重要な洞察を抽出）

プログレッシブ圧縮しきい値

業界標準パターン:

コンテキスト使用量  アクション                    技術
──────────────────────────────────────────────────
0-70%              圧縮なし                       逐語的に保存
70-85%             軽い圧縮                      冗長性を削除
85-95%             中程度の圧縮                  古いメッセージを要約
95-100%            積極的な圧縮                  階層的 + 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):
    """TF-IDF スコアリングを使用して最も重要なメッセージを抽出。"""
    texts = [msg['content'] for msg in messages]

    # TF-IDF スコアを計算
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform(texts)
    scores = np.array(tfidf_matrix.sum(axis=1)).flatten()

    # トップメッセージを選択
    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):
    """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"""この会話を要約してください。以下を保持してください:
1. 下した主要な決定
2. 重要なコンテキストと事実
3. 未解決の質問
4. アクションアイテム

会話:
{conversation_text}

要約（元の長さの1/5を目指す）:"""
        }]
    )

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

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

ツリー構造で要約の要約を作成。

利点: 極端な圧縮（20倍以上）、1M以上のトークン会話に対応 欠点: 複雑な実装、複数のLLMコール、情報損失が蓄積 最適: 長期実行会話、マルチセッションアプリケーション

アーキテクチャ:

レベル0（生):    [Msg1][Msg2][Msg3][Msg4][Msg5][Msg6][Msg7][Msg8]
レベル1（チャンク）:  [要約1-2]  [要約3-4]  [要約5-6]  [要約7-8]
レベル2（グループ）:  [要約1-4]              [要約5-8]
レベル3（セッション）: [セッション全体要約]

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]] = [[]]  # レベル0 = 生メッセージ

    def add_message(self, message: Dict):
        """メッセージを追加し、必要に応じて要約をトリガー。"""
        self.levels[0].append(message)

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

    def _summarize_level(self, level: int):
        """レベルを上位レベルに要約。"""
        messages = self.levels[level]

        # 次のレベルが存在することを確認
        while len(self.levels) <= level + 1:
            self.levels.append([])

        # 最初のチャンクを要約
        chunk = messages[:self.chunk_size]
        summary = self._generate_summary(chunk, level)

        # 次のレベルに移動
        self.levels[level + 1].append(summary)
        self.levels[level] = messages[self.chunk_size:]

        # 再帰的に次のレベルが要約が必要かをチェック
        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:
        """チャンクの要約を生成。"""
        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"このレベル {level} 会話チャンクを要約:\n\n{conversation_text}"
            }]
        )

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

    def get_context(self, max_tokens: int = 4000) -> List[Dict]:
        """トークン予算内のコンテキストを取得。"""
        context = []
        token_count = 0

        # 最近の生メッセージを優先
        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

        # 上位レベルから要約を追加
        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):
        """古いメッセージをローリング要約に圧縮。"""
        messages_to_compress = self.recent_messages[:-self.window_size]

        parts = []
        if self.rolling_summary:
            parts.append(f"既存の要約:\n{self.rolling_summary}")

        parts.append("\n新しいメッセージ:\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\n要約を更新:"
            }]
        )

        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"[要約]\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

        # ベクトルストアを初期化
        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):
        """最近のメモリとベクトルストアに追加。"""
        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):
        """ベクトルデータベースにアーカイブ。"""
        # 埋め込みを生成
        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):
        """RAGを使用して関連するコンテキストを取得。"""
        context = []
        token_count = 0

        # 1. 最近のメッセージ（短期メモリ）
        for msg in self.recent_messages:
            context.append(msg)
            token_count += len(msg['content']) // 4

        # 2. 関連する履歴コンテキストを取得
        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"[検索結果] {doc}"
                })
                token_count += len(doc) // 4

        return context

ベクトルデータベースオプション:

• ChromaDB: 組み込み、簡単なローカル開発
• Pinecone: マネージド、50ms p95 レイテンシ
• 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):
        """メッセージをクラスタリングし、代表値を返す。"""
        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)

        # 各重心に最も近いメッセージを選択
        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):
        """意味的に類似したメッセージを削除。"""
        if len(messages) <= 1:
            return messages

        # 埋め込みを生成
        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)

        # ユニークなメッセージをマーク
        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:
        """複合的な重要度スコアを計算。"""
        scores = []

        # 長さスコア（長い = より多くの情報）
        scores.append(min(len(msg['content']) / 500, 1.0))

        # 質問スコア
        if msg['role'] == 'user':
            scores.append(min(msg['content'].count('?') * 0.5, 1.0))

        # エンティティスコア（大文字で始まる単語）
        entities = len(re.findall(r'\b[A-Z][a-z]+', msg['content']))
        scores.append(min(entities / 10, 1.0))

        # 新しさスコア（線形減衰）
        scores.append(index / max(total - 1, 1))

        # ロールスコア
        scores.append(0.6 if msg['role'] == 'user' else 0.4)

        return sum(scores) / len(scores)

    def prioritize(self, messages: list, target_count: int):
        """優先度別に上位N個のメッセージを選択。"""
        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])  # 時系列順に復元

        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

        # 最も類似した前のメッセージを見つける
        last_msg = self.base_messages[-1]

        if last_msg['role'] == message['role']:
            # デルタを計算
            diff = list(difflib.unified_diff(
                last_msg['content'].splitlines(),
                message['content'].splitlines(),
                lineterm=''
            ))

            if len('\n'.join(diff)) < len(message['content']) * 0.7:
                # 圧縮が達成されたらデルタとして保存
                self.deltas.append({
                    'base_index': len(self.base_messages) - 1,
                    'delta': diff,
                    'role': message['role']
                })
                return

        # 新しいベースメッセージとして保存
        self.base_messages.append(message)

    def reconstruct(self):
        """ベース+デルタから完全な会話を再構築。"""
        messages = self.base_messages.copy()

        for delta_info in self.deltas:
            base_content = messages[delta_info['base_index']]['content']
            # diffを再構築に適用（簡略版）
            reconstructed = base_content  # 完全な実装では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)

# 会話を追加
memory.save_context(
    {"input": "Pythonプロジェクトに取り組んでいます"},
    {"output": "素晴らしい! Pythonプロジェクトについてどうお手伝いできますか?"}
)

# 要約を取得
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,  # これを超えると要約
    return_messages=True
)

# 会話を追加
for i in range(50):
    memory.save_context(
        {"input": f"質問 {i}"},
        {"output": f"回答 {i}"}
    )

# 自動的に最近のメッセージ+古い要約を保持
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
)

# トークン上限を超えたときは簡単なFIFO

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

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})
)

# 関連するコンテキストを自動的に検索

利点: 無限の会話長、セマンティック検索 欠点: ベクトルDB が必要、検索オーバーヘッド 最適: 長期実行会話、ナレッジベース

5. Anthropic 固有パターン

5.1 プロンプトキャッシング（90%コスト削減）

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

from anthropic import Anthropic

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

# 長い会話コンテキスト
conversation_history = [
    {"role": "user", "content": "メッセージ 1"},
    {"role": "assistant", "content": "レスポンス 1"},
    # ... さらに多くのメッセージ
]

# キャッシング用にコンテキストをマーク
messages = []
for i, msg in enumerate(conversation_history[:-1]):
    content = msg['content']

    # 最後のコンテキストメッセージにキャッシュ制御を追加
    if i == len(conversation_history) - 2:
        messages.append({
            "role": msg['role'],
            "content": [
                {
                    "type": "text",
                    "text": content,
                    "cache_control": {"type": "ephemeral"}
                }
            ]
        })
    else:
        messages.append(msg)

# 新しいユーザーメッセージを追加（キャッシュなし）
messages.append(conversation_history[-1])

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

# 同じキャッシュされたコンテキストの後続のコールはコスト90%削減

キャッシュ 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"""この会話を分析し、圧縮を推奨:

{conversation_text}

現在のトークン数: {current_tokens}
ターゲット: {target_tokens}
必要な圧縮: {compression_ratio}倍

最適な戦略を推奨."""
    }]
)

# 思考プロセスにアクセス
thinking_content = [
    block for block in response.content
    if block.type == "thinking"
]

# 圧縮推奨を取得
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):
        """現在のメモリ状態を保存。"""
        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):
        """チェックポイントからメモリ状態を読み込む。"""
        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):
        """N個のメッセージごとに自動的に保存。"""
        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):
        """既存セッションを読み込むか新規作成。"""
        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):
        """ユーザーメッセージを追加してレスポンスを取得。"""
        # ユーザーメッセージを追加
        self.memory['messages'].append({
            'role': 'user',
            'content': user_message
        })

        # コンテキスト（圧縮付き）をビルド
        context = self._build_context()

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

        # レスポンスを保存
        assistant_message = response.content[0].text
        self.memory['messages'].append({
            'role': 'assistant',
            'content': assistant_message
        })

        # 必要に応じて圧縮
        if len(self.memory['messages']) > 20:
            self._compress()

        # 状態を保存
        self._save()

        return assistant_message

    def _build_context(self):
        """圧縮付きでコンテキストをビルド。"""
        context = []

        # 要約が存在すれば追加
        if self.memory['summary']:
            context.append({
                'role': 'system',
                'content': f"[以前の会話要約]\n{self.memory['summary']}"
            })

        # 最近のメッセージを追加
        context.extend(self.memory['messages'][-10:])

        return context

    def _compress(self):
        """古いメッセージを圧縮。"""
        if len(self.memory['messages']) < 15:
            return

        # 要約するメッセージ
        to_summarize = self.memory['messages'][:-10]

        # 要約を生成
        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"この会話を要約:\n\n{conversation_text}"
            }]
        )

        # メモリを更新
        self.memory['summary'] = response.content[0].text
        self.memory['messages'] = self.memory['messages'][-10:]

    def _save(self):
        """セッションをディスクに保存。"""
        with open(f'sessions/{self.session_id}.json', 'w') as f:
            json.dump(self.memory, f, indent=2)

# 使用方法
client = Anthropic(api_key="your-api-key")
conversation = ResumableConversation(client, session_id="user123_session1")

# 複数のセッションにまたがって継続
response1 = conversation.add_turn("Pythonとは?")
# ... 後のセッション
response2 = conversation.add_turn("例を示してください")  # コンテキストを記憶

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

最適な結果を得るために複数の技術を組み合わせ。

from anthropic import Anthropic
from openai import OpenAI
import chromadb

class HybridMemorySystem:
    """
    以下を組み合わせ:
    - ローリング要約（短期圧縮）
    - RAG検索（長期メモリ）
    - プロンプトキャッシング（コスト最適化）
    - プログレッシブ圧縮（適応的動作）
    """

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

        # 最近のメッセージ（逐語的）
        self.recent_messages = []
        self.recent_window = 10

        # ローリング要約
        self.rolling_summary = None

        # ベクトルストア（長期）
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection(name="memory")
        self.message_counter = 0

        # 圧縮しきい値
        self.thresholds = {
            'light': 0.70,    # 基本的な圧縮を開始
            'medium': 0.85,   # 積極的な要約
            'heavy': 0.95     # 緊急措置
        }

    def add_message(self, message: dict):
        """インテリジェント圧縮付きでメッセージを追加。"""
        self.recent_messages.append(message)

        # 圧縮ニーズをチェック
        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):
        """冗長性を削除、ベクトルストアにアーカイブ。"""
        if len(self.recent_messages) > self.recent_window * 1.5:
            # 最古のものをベクトルストアにアーカイブ
            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):
        """ローリング要約を生成、積極的にアーカイブ。"""
        if len(self.recent_messages) > self.recent_window:
            # 古いメッセージを要約
            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"既存: {self.rolling_summary}\n\n新規: {summary_text}"

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

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

            # 要約されたすべてのメッセージをアーカイブ
            for msg in to_summarize:
                self._archive_to_vectorstore(msg)

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

    def _emergency_compress(self):
        """限界間近の状況での極端な圧縮。"""
        # 最後の5つのメッセージのみ保持
        to_archive = self.recent_messages[:-5]
        for msg in to_archive:
            self._archive_to_vectorstore(msg)

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

        # 必要に応じてさらに要約を圧縮
        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"超簡潔な要約を作成:\n{self.rolling_summary}"
                }]
            )
            self.rolling_summary = response.content[0].text

    def _archive_to_vectorstore(self, message: dict):
        """ベクトルデータベースに保存して検索に対応。"""
        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):
        """現在のクエリに対して最適なコンテキストをビルド。"""
        context = []
        token_count = 0

        # 1. ローリング要約を追加（存在する場合）
        if self.rolling_summary:
            summary_msg = {
                'role': 'system',
                'content': [
                    {
                        'type': 'text',
                        'text': f"[会話要約]\n{self.rolling_summary}",
                        'cache_control': {'type': 'ephemeral'}  # キャッシュ
                    }
                ]
            }
            context.append(summary_msg)
            token_count += len(self.rolling_summary) // 4

        # 2. 関連する履歴コンテキストを検索（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"[検索結果] {doc}"
                })
                token_count += len(doc) // 4

        # 3. 最近のメッセージを逐語的に追加
        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):
        """現在のコンテキストウィンドウ使用量を推定。"""
        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 コンテキストウィンドウ

# 使用方法
anthropic_client = Anthropic(api_key="your-anthropic-key")
openai_client = OpenAI(api_key="your-openai-key")

memory = HybridMemorySystem(anthropic_client, openai_client)

# 時間をかけてメッセージを追加
for i in range(1000):
    memory.add_message({
        'role': 'user' if i % 2 == 0 else 'assistant',
        'content': f"メッセージ {i} に何かしらのコンテンツを含む..."
    })

# 最適化されたコンテキストを取得
current_query = "価格設定について何を議論しましたか?"
context = memory.get_context(current_query)

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

---

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

圧縮効率

技術	圧縮率	品質低下	レイテンシ	コスト影響
抽出的	2-3倍	<1%	<10ms	なし
抽象的	5-10倍	2-5%	1-2秒	+$0.001/ターン
階層的	20倍+	5-8%	2-5秒	+$0.003/ターン
LLMLingua	20倍	1.5%	500ms	なし
RAG	可変	<1%	100-300ms	+$0.0005/ターン
プロンプトキャッシング	N/A	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の価格（1Mトークンあたり入力 $3、出力 $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")

# 圧縮、要約、RAGを自動的に処理
memory = client.create_memory(
    user_id="user123",
    messages=[
        {"role": "user", "content": "Pythonプロジェクトに取り組んでいます"},
        {"role": "assistant", "content": "素晴らしい! どんなプロジェクトですか?"}
    ]
)

# 関