エージェントメモリシステム設計：AIエージェントの記憶のすべて

# これが感覚記憶です:
messages = [
    {"role": "user", "content": "私の名前は田中です"},
    {"role": "assistant", "content": "こんにちは、田中さん！"},
    {"role": "user", "content": "私の名前は何ですか？"},
    # LLMは上の会話内容をそのまま見られるので答えられます
]

from langchain.memory import ConversationBufferWindowMemory

# 最近10回の会話のやり取りのみ保持
memory = ConversationBufferWindowMemory(k=10, return_messages=True)

# 会話追加
memory.save_context(
    {"input": "Pythonのリスト内包表記って何？"},
    {"output": "リスト内包表記は、既存のリストをもとに新しいリストを作る簡潔な方法です。"}
)

# メモリ読み込み（LLMに注入）
history = memory.load_memory_variables({})

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

# 長期メモリストア作成
embedding = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(["placeholder"], embedding)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

memory = VectorStoreRetrieverMemory(retriever=retriever)

# 記憶の保存
memory.save_context(
    {"input": "私の好きなプログラミング言語はPythonです"},
    {"output": "わかりました！Pythonがお好きなんですね。"}
)

memory.save_context(
    {"input": "私は東京に住んでいてスタートアップで働いています"},
    {"output": "東京のスタートアップにお勤めなんですね！"}
)

# 後で：関連記憶の検索
relevant = memory.load_memory_variables({"prompt": "私に合う技術を教えて"})
# Python、東京スタートアップ関連の記憶を返す

import json
from datetime import datetime

class EpisodicMemory:
    def __init__(self, db_connection):
        self.db = db_connection

    def record_episode(self, user_id: str, episode: dict):
        """特定のイベント/インタラクションを記録"""
        self.db.insert("episodes", {
            "user_id": user_id,
            "timestamp": datetime.now().isoformat(),
            "event_type": episode["type"],  # "purchase", "complaint", "success"
            "summary": episode["summary"],
            "metadata": json.dumps(episode.get("metadata", {}))
        })

    def retrieve_episodes(
        self,
        user_id: str,
        event_type: str = None,
        limit: int = 10
    ):
        """ユーザーの過去のエピソードを検索"""
        if event_type:
            query = """
                SELECT * FROM episodes
                WHERE user_id = ? AND event_type = ?
                ORDER BY timestamp DESC LIMIT ?
            """
            return self.db.query(query, [user_id, event_type, limit])
        else:
            query = """
                SELECT * FROM episodes
                WHERE user_id = ?
                ORDER BY timestamp DESC LIMIT ?
            """
            return self.db.query(query, [user_id, limit])

# 使用例
memory = EpisodicMemory(db)
memory.record_episode("user_123", {
    "type": "purchase",
    "summary": "ユーザーがPython上級コースを購入",
    "metadata": {"course_id": "py-advanced", "price": 8900}
})

ユーザーメッセージ
     |
     v
+--------------------------------------------------+
| メモリルーター                                    |
| (関連メモリ収集・優先度決定)                      |
+------------+----------+-----------+--------------+
             |          |           |
             v          v           v           v
          感覚      短期記憶      長期記憶    エピソード
        (context)  (buffer)     (vector)    (event log)
             |          |           |           |
             +----------+-----------+-----------+
                                |
                                v
                 統合されたコンテキスト
                 (最も関連性の高い記憶を選択)
                                |
                                v
                         LLM生成
                                |
                                v
                         メモリ更新
                     (新情報を適切なメモリに保存)

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.memory import (
    ConversationBufferWindowMemory,
    VectorStoreRetrieverMemory,
    CombinedMemory
)
from langchain.prompts import PromptTemplate
from langchain.chains import ConversationChain

# メモリ構成
short_term = ConversationBufferWindowMemory(
    k=5,
    memory_key="chat_history",
    return_messages=True
)

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(["placeholder"], embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
long_term = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="relevant_history"
)

# 2つのメモリを結合
combined_memory = CombinedMemory(memories=[short_term, long_term])

# プロンプト設定
template = """
あなたはユーザーをよく覚えている個人AIアシスタントです。

過去の関連記憶：
{relevant_history}

最近の会話：
{chat_history}

ユーザー：{input}
アシスタント："""

prompt = PromptTemplate(
    input_variables=["relevant_history", "chat_history", "input"],
    template=template
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = ConversationChain(
    llm=llm,
    memory=combined_memory,
    prompt=prompt,
    verbose=True
)

# 会話
chain.predict(input="こんにちは！私はバックエンド開発者です。")
chain.predict(input="最近Rustを学んでいます。")

# 新しいセッションでも長期記憶を活用
chain.predict(input="私が学んでいる言語でWebサーバーを作る方法を教えて。")
# 「Rustで...」と答えることができる

from mem0 import Memory

m = Memory()

# メモリ追加（LLMが重要な情報を自動的に抽出して保存）
result = m.add(
    messages=[
        {"role": "user", "content": "私はベジタリアンで、日本人で東京に住んでいます。Pythonが好きです。"},
        {"role": "assistant", "content": "わかりました！覚えておきます。"}
    ],
    user_id="user_123"
)

# 保存されたメモリの確認
print(result)
# [
#   {"memory": "ユーザーはベジタリアン", "id": "..."},
#   {"memory": "ユーザーは日本人で東京在住", "id": "..."},
#   {"memory": "ユーザーはPythonを好む", "id": "..."}
# ]

# メモリ検索（意味的類似性に基づく）
memories = m.search("食べ物を推薦して", user_id="user_123")
# 返る：ベジタリアン、東京在住関連の記憶

# エージェントで活用
context = "\n".join([item["memory"] for item in memories["results"]])
response = llm.invoke(
    f"ユーザー情報：\n{context}\n\nリクエスト：食べ物を推薦して"
)

# 3年前: "私は東京に住んでいます"
# 今日: "去年大阪に引っ越しました"

# どう処理する？
# オプションA: 最新情報で上書き
# オプションB: タイムスタンプ付きで両方保持
# オプションC: ユーザーに確認を求める

# GDPR準拠のためのメモリ削除
def delete_user_memory(user_id: str):
    m.delete_all(user_id=user_id)
    print(f"ユーザー {user_id} のすべてのメモリを削除完了")

import math
from datetime import datetime

def calculate_memory_importance(memory: dict) -> float:
    """
    記憶の現在の重要度を計算します。
    時間が経つほど重要度が下がりますが、
    頻繁に参照される記憶は重要度が維持されます。
    """
    days_since_last_access = (
        datetime.now() - memory["last_accessed"]
    ).days
    access_count = memory["access_count"]

    # エビングハウス忘却曲線に基づく計算
    # 基本的な忘却：時間が経つほど低下
    base_retention = math.exp(-days_since_last_access / 30)

    # アクセス頻度による強化（頻繁に参照されるほど忘れにくい）
    reinforcement = math.log(1 + access_count) * 0.3

    return min(1.0, base_retention + reinforcement)

# 重要度が低いメモリを定期的に整理
def prune_memories(user_id: str, threshold: float = 0.1):
    memories = m.get_all(user_id=user_id)
    for memory in memories:
        if calculate_memory_importance(memory) < threshold:
            m.delete(memory["id"])