E-memで社内ヘルプデスクボットの長期記憶を実装しトークンコストを70%削減する

E-memで社内ヘルプデスクボットの長期記憶を実装しトークンコストを70%削減する

この記事でわかること

• E-mem（Episodic Context Reconstruction）のマスター・アシスタント階層アーキテクチャの仕組みと設計思想
• 社内ヘルプデスクボットにエピソディックメモリを組み込み、過去の対話文脈を保持する実装パターン
• 3段階パイプライン（ルーティング→アシスタント推論→マスター集約）をPythonで構築する方法
• Mem0やLangMemとの使い分けと、検索想起精度を高めるマルチシグナル検索の実装
• LoCoMoベンチマークでのE-memの評価結果と、本番運用時のホット/コールドパス設計

対象読者

• 想定読者: 中級者〜上級者のLLMアプリケーション開発者
• 必要な前提知識:
  • Python 3.11以上の基本的な非同期プログラミング（asyncio）
  • LangChainまたはLlamaIndexでのRAG構築経験
  • ベクトルデータベース（Qdrant、Pinecone等）の基本操作
  • OpenAI APIまたはAnthropic APIの利用経験

結論・成果

E-memのエピソード文脈再構成アプローチを社内ヘルプデスクボットに適用することで、論文のベンチマークではLoCoMo F1スコア54.17%（従来手法GAM比+8.86%）を達成しながら、トークンコストを70%以上削減できたと報告されています。特にMulti-Hop質問（+10%改善）とTemporal質問（+8.87%改善）で大きな効果が確認されており、「先月の○○の件、その後どうなりましたか？」のような時系列を跨ぐヘルプデスク問い合わせに有効です。

ただし、E-memはマルチエージェント構成のため初期構築の複雑さがあり、月間問い合わせ1,000件未満ではMem0単体のほうがROIが高い場合もあります。

エージェントメモリの課題とE-memの位置づけを理解する

社内ヘルプデスクボットが抱えるメモリの壁

社内ヘルプデスクボットではコンテキストウィンドウの制約が深刻なボトルネックです。

ユーザー: 先週VPN接続できなくて問い合わせた者ですが、また同じ症状です
ボット: 申し訳ございません。もう少し詳しく症状を教えていただけますか？

これは典型的な失敗例です。過去の対話履歴が保持されていないため、ユーザーは毎回同じ説明を繰り返すことになります。

従来のメモリ実装では、以下の3つのアプローチが取られてきました。

E-memの論文では、E-memのトークンコストは3,621であり、Full Context（169,100）やGAM（12,540）と比較して大幅な削減を実現しています。

E-memが解決する「破壊的脱文脈化」問題

既存のメモリ前処理パイプライン（静的な埋め込みやナレッジグラフへの変換）は、破壊的脱文脈化（destructive de-contextualization） という根本的な問題を抱えています。会話の時系列的な依存関係や因果チェーンが、構造化の過程で失われてしまうのです。

E-memはこの問題に対し、生物学的なエングラムに着想を得たエピソード文脈再構成で対応します。メモリを圧縮せず分散保持し、必要時に動的に再構成する設計です。

2026年のエージェントメモリフレームワーク比較

E-memの位置づけを理解するために、主要フレームワークとの比較を確認しましょう。

注意: Mem0のスコアは独自ベンチマーク設定での評価値であり、E-memの論文評価とは条件が異なります。直接比較する際はベンチマーク条件の違いに留意してください。

E-memの3段階パイプラインを実装する

アーキテクチャの全体像

E-memは異種階層型マルチエージェントアーキテクチャを採用しています。中央のマスターエージェントが全体の計画を統括し、複数のアシスタントエージェントがそれぞれの記憶セグメントを非圧縮で保持します。

E-memの論文では、マスターエージェントにGPT-4o-miniまたはQwen 2.5-14B、アシスタントエージェントにQwen 3-4Bを使用しています。大規模モデルと小規模モデルを使い分けることで、精度とコストのバランスを取る設計です。

Stage 1: マルチパスウェイ・ルーティングを構築する

ルーティング機構は3つの直交するシグナルを並列に走らせ、いずれか1つでもヒットしたメモリユニットを活性化します（Union方式）。

from dataclasses import dataclass
from typing import Protocol
import numpy as np
from sentence_transformers import SentenceTransformer


@dataclass(frozen=True)
class MemoryUnit:
    unit_id: str
    raw_text: str
    summary: str
    embedding: np.ndarray
    entities: frozenset[str]
    timestamp: float


class ActivationPathway(Protocol):
    def activate(
        self, query: str, units: list[MemoryUnit]
    ) -> set[str]: ...


class GlobalAlignmentPathway:
    """要約ベースの意味的類似度で大局的な関連性を判定する"""

    def __init__(self, model_name: str = "intfloat/multilingual-e5-large"):
        self._encoder = SentenceTransformer(model_name)
        self._threshold = 0.45

    def activate(
        self, query: str, units: list[MemoryUnit]
    ) -> set[str]:
        q_emb = self._encoder.encode(query, normalize_embeddings=True)
        activated = set()
        for unit in units:
            summary_emb = self._encoder.encode(
                unit.summary, normalize_embeddings=True
            )
            score = float(np.dot(q_emb, summary_emb))
            if score >= self._threshold:
                activated.add(unit.unit_id)
        return activated


class SemanticAssociationPathway:
    """生チャンク埋め込みとの高次元ベクトル類似度で検出する"""

    def __init__(self, threshold: float = 0.5):
        self._threshold = threshold

    def activate(
        self, query: str, units: list[MemoryUnit]
    ) -> set[str]:
        encoder = SentenceTransformer("intfloat/multilingual-e5-large")
        q_emb = encoder.encode(query, normalize_embeddings=True)
        activated = set()
        for unit in units:
            score = float(np.dot(q_emb, unit.embedding))
            if score >= self._threshold:
                activated.add(unit.unit_id)
        return activated


class SymbolicTriggerPathway:
    """BM25ベースのキーワードマッチで正確なエンティティ参照を検出する"""

    def activate(
        self, query: str, units: list[MemoryUnit]
    ) -> set[str]:
        query_tokens = set(query.lower().split())
        activated = set()
        for unit in units:
            overlap = query_tokens & unit.entities
            if overlap:
                activated.add(unit.unit_id)
        return activated


class MultiPathwayRouter:
    """3経路のUnion方式でメモリユニットを活性化する"""

    def __init__(self, pathways: list[ActivationPathway]):
        self._pathways = pathways

    def route(
        self, query: str, units: list[MemoryUnit]
    ) -> list[MemoryUnit]:
        activated_ids: set[str] = set()
        for pathway in self._pathways:
            activated_ids |= pathway.activate(query, units)
        return [u for u in units if u.unit_id in activated_ids]

なぜUnion方式なのか：
E-memの論文のアブレーション実験では、Global Alignmentを除去するとF1が-10.46ポイント低下し、Semantic Associationの除去でも47.90%まで低下することが確認されています。3つの経路は相互補完的であり、いずれか1つを外すと特定タイプの質問で大幅に精度が落ちます。

注意点:

Union方式はrecall（再現率）を優先する設計であるため、活性化されるメモリユニットが多くなりがちです。E-memの論文では、k=8〜20チャンクで「情報飽和」が起こると報告されており、それ以上のユニットを活性化してもF1は改善しません。ヘルプデスクボットでは、直近30日以内のチケットに絞るなどの時間窓フィルタを併用することを推奨します。

Stage 2: アシスタントエージェントによるローカル推論を実装する

活性化されたメモリユニットごとに、小規模LLM（SLM）がローカル推論を実行します。各アシスタントは自身が保持するエピソード文脈内で推論し、タイムスタンプ付きの証拠タプルを生成します。

from datetime import datetime
from openai import AsyncOpenAI


@dataclass
class EvidenceTuple:
    finding: str
    timestamp: datetime
    source_unit_id: str
    confidence: float


class AssistantAgent:
    """小規模LLMで局所的なエピソード推論を行うアシスタント"""

    def __init__(
        self,
        client: AsyncOpenAI,
        model: str = "qwen/qwen3-4b",
    ):
        self._client = client
        self._model = model

    async def extract_evidence(
        self, query: str, unit: MemoryUnit
    ) -> list[EvidenceTuple]:
        prompt = f"""以下のエピソード文脈から、クエリに関連する証拠を抽出してください。

クエリ: {query}

エピソード文脈:
{unit.raw_text}

以下のJSON形式で回答してください:
[{{"finding": "発見内容", "confidence": 0.0-1.0}}]
"""
        response = await self._client.chat.completions.create(
            model=self._model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1,
            response_format={"type": "json_object"},
        )
        import json

        results = json.loads(response.choices[0].message.content)
        return [
            EvidenceTuple(
                finding=r["finding"],
                timestamp=datetime.fromtimestamp(unit.timestamp),
                source_unit_id=unit.unit_id,
                confidence=r["confidence"],
            )
            for r in results.get("evidences", results)
            if isinstance(r, dict) and "finding" in r
        ]

E-memではアシスタントに小規模モデル（Qwen 3-4B等）を割り当て、大規模モデルのトークンを消費せず並列処理します。論文のコスト分析ではSLMが約2,271トークン、LLMが約135トークン/クエリでコスト比は約1:10です。

Stage 3: マスターエージェントによる統合集約を実装する

マスターエージェントは、各アシスタントから集まった証拠タプルを時系列論理に基づいて統合し、状態の矛盾を解消します。

class MasterAgent:
    """分散証拠を統合し、一貫した応答を生成するマスター"""

    def __init__(
        self,
        client: AsyncOpenAI,
        model: str = "gpt-4o-mini",
    ):
        self._client = client
        self._model = model

    async def synthesize(
        self,
        query: str,
        evidences: list[EvidenceTuple],
    ) -> str:
        sorted_evidences = sorted(
            evidences, key=lambda e: e.timestamp
        )

        evidence_text = "\n".join(
            f"[{e.timestamp.strftime('%Y-%m-%d %H:%M')}] "
            f"(確信度: {e.confidence:.2f}) {e.finding}"
            for e in sorted_evidences
            if e.confidence >= 0.3
        )

        prompt = f"""あなたは社内ヘルプデスクのアシスタントです。
以下の時系列に整理された証拠を基に、ユーザーのクエリに正確に回答してください。

クエリ: {query}

時系列証拠:
{evidence_text}

回答のルール:
- 最新の情報を優先する（同じ事象について矛盾がある場合）
- 証拠がない場合は推測せず「確認中」と回答する
- 過去の対応履歴がある場合は参照する
"""
        response = await self._client.chat.completions.create(
            model=self._model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2,
        )
        return response.choices[0].message.content

なぜこの実装を選んだか：

• マスターエージェントは「生のメモリ保持の負担から切り離された中央オーケストレータ」（論文より）として機能します
• 時系列ソートにより、ヘルプデスクで頻出する「以前○○したのですが」パターンに対応できます
• 確信度0.3未満のノイズを除去することで、マスターエージェントへの入力トークンをさらに削減します

E-memパイプラインを統合する

3つのステージを統合したパイプラインクラスを構築します。

import asyncio


class EMemPipeline:
    """E-memの3段階パイプラインを統合した推論エンジン"""

    def __init__(
        self,
        router: MultiPathwayRouter,
        assistant: AssistantAgent,
        master: MasterAgent,
        max_activated_units: int = 20,
    ):
        self._router = router
        self._assistant = assistant
        self._master = master
        self._max_units = max_activated_units

    async def answer(
        self,
        query: str,
        memory_store: list[MemoryUnit],
    ) -> str:
        activated = self._router.route(query, memory_store)
        activated = activated[: self._max_units]

        evidence_tasks = [
            self._assistant.extract_evidence(query, unit)
            for unit in activated
        ]
        evidence_lists = await asyncio.gather(*evidence_tasks)
        all_evidences = [
            e for sublist in evidence_lists for e in sublist
        ]

        return await self._master.synthesize(query, all_evidences)

社内ヘルプデスクボットにメモリ層を組み込む

メモリストアの設計パターン

社内ヘルプデスクでは、問い合わせ内容に応じてメモリの保持期間と検索戦略を変える必要があります。ここではホットパス（即時コンテキスト）とコールドパス（長期ストア） の2層アーキテクチャを採用します。

ホットパスとコールドパスの分離を実装する

from datetime import timedelta


class HelpdeskMemoryStore:
    """ヘルプデスク向け2層メモリストア"""

    def __init__(
        self,
        hot_window: timedelta = timedelta(hours=1),
        cold_retention: timedelta = timedelta(days=90),
    ):
        self._hot_window = hot_window
        self._cold_retention = cold_retention
        self._hot_store: dict[str, list[MemoryUnit]] = {}
        self._cold_store: list[MemoryUnit] = []
        self._encoder = SentenceTransformer(
            "intfloat/multilingual-e5-large"
        )

    def ingest(
        self,
        user_id: str,
        text: str,
        summary: str,
        entities: set[str],
    ) -> MemoryUnit:
        """新しい対話をメモリに取り込む"""
        import time
        import uuid

        embedding = self._encoder.encode(
            text, normalize_embeddings=True
        )
        unit = MemoryUnit(
            unit_id=str(uuid.uuid4()),
            raw_text=text,
            summary=summary,
            embedding=embedding,
            entities=frozenset(entities),
            timestamp=time.time(),
        )

        if user_id not in self._hot_store:
            self._hot_store[user_id] = []
        self._hot_store[user_id].append(unit)

        return unit

    def get_relevant_units(
        self, user_id: str
    ) -> list[MemoryUnit]:
        """ホットパス＋コールドパスの統合ビューを返す"""
        import time

        now = time.time()
        hot_cutoff = now - self._hot_window.total_seconds()
        cold_cutoff = now - self._cold_retention.total_seconds()

        hot_units = [
            u
            for u in self._hot_store.get(user_id, [])
            if u.timestamp >= hot_cutoff
        ]
        cold_units = [
            u
            for u in self._cold_store
            if u.timestamp >= cold_cutoff
        ]
        return hot_units + cold_units

    async def flush_to_cold(self) -> None:
        """ホットパスの期限切れユニットをコールドパスに移動する"""
        import time

        now = time.time()
        cutoff = now - self._hot_window.total_seconds()

        for user_id, units in self._hot_store.items():
            expired = [u for u in units if u.timestamp < cutoff]
            self._cold_store.extend(expired)
            self._hot_store[user_id] = [
                u for u in units if u.timestamp >= cutoff
            ]

非同期メモリ書き込みを導入する

2026年のエージェントメモリ設計では、非同期メモリ書き込みがデファクトスタンダードです。ユーザーへの応答レイテンシに影響を与えずにメモリを更新できます。

import asyncio
from collections.abc import Callable, Awaitable


class AsyncMemoryWriter:
    """応答レイテンシに影響を与えない非同期メモリ書き込み"""

    def __init__(
        self,
        store: HelpdeskMemoryStore,
        summarizer: Callable[[str], Awaitable[str]],
        entity_extractor: Callable[
            [str], Awaitable[set[str]]
        ],
    ):
        self._store = store
        self._summarize = summarizer
        self._extract_entities = entity_extractor
        self._queue: asyncio.Queue[
            tuple[str, str]
        ] = asyncio.Queue()

    async def enqueue(
        self, user_id: str, text: str
    ) -> None:
        """書き込みをキューに追加（即座に返る）"""
        await self._queue.put((user_id, text))

    async def process_loop(self) -> None:
        """バックグラウンドで書き込みを処理する"""
        while True:
            user_id, text = await self._queue.get()
            try:
                summary, entities = await asyncio.gather(
                    self._summarize(text),
                    self._extract_entities(text),
                )
                self._store.ingest(
                    user_id, text, summary, entities
                )
            except Exception:
                pass
            finally:
                self._queue.task_done()

ハマりポイント：

非同期書き込みでは、書き込み完了前にユーザーが次の質問をすると、直前の対話がまだメモリに反映されていない場合があります。ヘルプデスクでは「さっき言ったのに覚えてない」というクレームに直結するため、ホットパスの直近5ターンはコンテキストウィンドウに直接含めるバッファ戦略が必須です。

検索想起精度を高めるマルチシグナル検索を実装する

E-memのルーティングとMem0のマルチシグナル検索を組み合わせる

E-memの3経路ルーティングは記憶の「活性化」に焦点を当てていますが、本番のヘルプデスクではさらにリランキングを加えることで検索想起精度を高められます。Mem0が採用しているマルチシグナル検索スタック（セマンティック類似度＋キーワードマッチ＋エンティティマッチの並列スコアリング）の考え方を組み合わせます。

@dataclass
class ScoredUnit:
    unit: MemoryUnit
    semantic_score: float
    keyword_score: float
    entity_score: float
    combined_score: float


class MultiSignalReranker:
    """E-mem活性化後のリランキングで精度を向上させる"""

    def __init__(
        self,
        semantic_weight: float = 0.5,
        keyword_weight: float = 0.2,
        entity_weight: float = 0.3,
    ):
        self._weights = (
            semantic_weight,
            keyword_weight,
            entity_weight,
        )
        self._encoder = SentenceTransformer(
            "intfloat/multilingual-e5-large"
        )

    def rerank(
        self,
        query: str,
        activated_units: list[MemoryUnit],
        top_k: int = 10,
    ) -> list[ScoredUnit]:
        q_emb = self._encoder.encode(
            query, normalize_embeddings=True
        )
        query_tokens = set(query.lower().split())

        scored = []
        for unit in activated_units:
            sem_score = float(np.dot(q_emb, unit.embedding))

            text_tokens = set(unit.raw_text.lower().split())
            if text_tokens:
                kw_score = len(
                    query_tokens & text_tokens
                ) / len(query_tokens | text_tokens)
            else:
                kw_score = 0.0

            if unit.entities:
                ent_overlap = query_tokens & {
                    e.lower() for e in unit.entities
                }
                ent_score = len(ent_overlap) / len(
                    unit.entities
                )
            else:
                ent_score = 0.0

            combined = (
                self._weights[0] * sem_score
                + self._weights[1] * kw_score
                + self._weights[2] * ent_score
            )

            scored.append(
                ScoredUnit(
                    unit=unit,
                    semantic_score=sem_score,
                    keyword_score=kw_score,
                    entity_score=ent_score,
                    combined_score=combined,
                )
            )

        scored.sort(key=lambda s: s.combined_score, reverse=True)
        return scored[:top_k]

なぜこの重み配分なのか：

• セマンティック（0.5）: 「ログインできない」と「認証エラー」のような同義表現の検出に最重要
• エンティティ（0.3）: 「Slack設定」と「Teams設定」のようなツール固有の問い合わせを区別
• キーワード（0.2）: エラーコードや製品バージョンなど完全一致が有効な場面で補完

LoCoMoベンチマークで見るE-memの強み

E-memが特に優れているのは、Multi-Hop質問とTemporal質問です。これはヘルプデスクの典型的な問い合わせパターンに合致します。

Open Domain質問ではGAMにわずかに劣りますが、ヘルプデスクではチケットに紐づく具体的な質問が多く、E-memの特性が有利に働きます。

本番運用の設計パターンとトラブルシューティング

規模別の導入戦略

E-memのマルチエージェント構成は小規模では過剰になる場合があります。規模に応じた選択肢を整理します。

よくある問題と解決方法

本番デプロイ時の留意点

本番環境ではPython 3.11以上、Qdrant（ベクトルDB）、PostgreSQL（構造化データ）の3層構成を推奨します。コールドストアのメモリユニットは90日TTLで定期クリーンアップし、重要フラグ付きのユニットのみ永続保持とするのが運用上のバランス点です。

制約事項: E-memのアシスタントエージェントにセルフホストのSLM（Qwen 3-4B等）を使う場合、GPUサーバーが別途必要になります。API経由で小規模モデルを呼び出す（Together AI、Fireworks等）ほうが初期導入のハードルは低くなります。

まとめと次のステップ

まとめ:

• E-memは非圧縮メモリ保持とマルチエージェント協調により、従来手法比でF1を+8.86%改善しつつトークンコストを70%削減するアプローチです
• 社内ヘルプデスクでは、Multi-Hop（過去チケット参照）とTemporal（時系列推論）の精度向上が特に有効です
• ホットパス（直近セッション）とコールドパス（長期ストア）の2層設計と非同期書き込みが本番運用の基本パターンです
• 月間問い合わせ1,000件未満ならMem0単体、1,000件以上でE-memの導入を検討する段階的アプローチが現実的です
• Open Domain質問では従来手法に劣る点があるため、ナレッジベース検索との併用が必要です

次にやるべきこと:

• E-memのリポジトリをクローンし、LoCoMoベンチマークで動作確認する
• 自社のヘルプデスクチケットデータでメモリユニットのセグメンテーション粒度を調整する
• Mem0のクイックスタートガイドでホットパスの検索精度を検証する

参考

• E-mem: Multi-agent based Episodic Context Reconstruction for LLM Agent Memory (arXiv 2601.21714)
• State of AI Agent Memory 2026 - Mem0 Blog
• A Practical Guide to Memory for Autonomous LLM Agents - Towards Data Science
• Memory for Autonomous LLM Agents: Mechanisms, Evaluation, and Emerging Frontiers (arXiv 2603.07670)
• Agent Memory Techniques - GitHub (NirDiamant)

GitHubで編集を提案