AIエージェントのメモリ設計：会話を超えた文脈管理の4つのパターン

def compress_history(messages: list[dict], keep_recent: int = 6) -> list[dict]:
    if len(messages) <= keep_recent:
        return messages
    
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]
    
    # 古いメッセージを要約
    summary_prompt = [
        {"role": "user", "content": f"以下の会話を3〜5文で要約してください：\n{format_messages(old_messages)}"}
    ]
    summary = call_llm(summary_prompt)
    
    # 要約を先頭に挿入して返す
    compressed = [
        {"role": "system", "content": f"これまでの会話の要約：{summary}"}
    ] + recent_messages
    
    return compressed

def extract_and_store_facts(user_message: str, agent_response: str, user_id: str):
    extraction_prompt = f"""
以下の会話から、後で参照すべき永続的な事実を抽出してください。
一時的な内容（「今日は疲れた」など）は除外し、
ユーザーについての固定情報や設定のみを抽出してください。
 
ユーザー: {user_message}
アシスタント: {agent_response}
 
抽出すべき事実があれば JSON 形式で返してください。なければ空配列を返してください。
形式: [{"key": "事実の種類", "value": "内容"}]
"""
    facts = call_llm_json(extraction_prompt)
    
    for fact in facts:
        key = f"user:{user_id}:{fact['key']}"
        redis_client.set(key, fact['value'], ex=86400*365)  # 1年

from openai import OpenAI
import chromadb
 
client = OpenAI()
chroma = chromadb.Client()
collection = chroma.get_or_create_collection("agent_memory")
 
def store_exchange(user_msg: str, agent_msg: str, session_id: str):
    # 会話ペアをチャンクとして保存
    text = f"ユーザー: {user_msg}\nアシスタント: {agent_msg}"
    embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    ).data[0].embedding
    
    collection.add(
        documents=[text],
        embeddings=[embedding],
        ids=[f"{session_id}_{int(time.time())}"],
        metadatas=[{"session_id": session_id, "timestamp": time.time()}]
    )
 
def recall_relevant_memory(query: str, n_results: int = 3) -> str:
    query_embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results
    )
    
    if not results["documents"][0]:
        return ""
    
    return "関連する過去の会話:\n" + "\n---\n".join(results["documents"][0])

from dataclasses import dataclass
from datetime import datetime
 
@dataclass
class Episode:
    session_id: str
    timestamp: datetime
    topic: str          # LLMで自動抽出
    participants: list  # ユーザーID等
    summary: str        # LLMで自動生成
    key_decisions: list # 決定事項の抽出
    follow_ups: list    # 次回に持ち越す事項
 
def create_episode(messages: list[dict], session_id: str) -> Episode:
    # セッション終了時に自動生成
    analysis_prompt = f"""
以下の会話を分析し、JSON形式で構造化してください：
- topic: 主なトピック（1文）
- summary: 会話全体の要約（3〜5文）
- key_decisions: 決定したこと（リスト）
- follow_ups: 次回に確認・継続すべきこと（リスト）
 
会話：
{format_messages(messages)}
"""
    structured = call_llm_json(analysis_prompt)
    
    return Episode(
        session_id=session_id,
        timestamp=datetime.now(),
        topic=structured["topic"],
        participants=["user"],
        summary=structured["summary"],
        key_decisions=structured.get("key_decisions", []),
        follow_ups=structured.get("follow_ups", [])
    )