AI Agent 记忆工程：从短期上下文到持久化知识体系的完整架构

"""
上下文预算管理器：智能分配工作记忆空间
核心思想：将有限的上下文窗口视为稀缺资源，按优先级分配
"""

from dataclasses import dataclass
from typing import List, Dict, Optional
import tiktoken


@dataclass
class MemoryBudget:
    """上下文窗口的预算分配"""
    system_prompt_max: int = 4000
    tool_defs_max: int = 8000
    conversation_max: int = 32000
    retrieval_max: int = 16000
    reserved_for_output: int = 4000

    @property
    def total(self) -> int:
        return (self.system_prompt_max + self.tool_defs_max +
                self.conversation_max + self.retrieval_max +
                self.reserved_for_output)


class ContextBudgetManager:
    """管理上下文窗口的预算分配与溢出处理"""

    def __init__(self, model: str = "gpt-5.5", budget: Optional[MemoryBudget] = None):
        self.budget = budget or MemoryBudget()
        self.encoder = tiktoken.encoding_for_model(model)

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def build_context(
        self,
        system_prompt: str,
        tool_definitions: List[Dict],
        conversation_history: List[Dict],
        retrieved_knowledge: List[str],
    ) -> List[Dict]:
        """构建上下文，自动处理预算溢出"""

        messages = []

        # 1. System Prompt（最高优先级）
        sp_tokens = self.count_tokens(system_prompt)
        if sp_tokens > self.budget.system_prompt_max:
            system_prompt = self._truncate(system_prompt, self.budget.system_prompt_max)
        messages.append({"role": "system", "content": system_prompt})

        # 2. 工具定义（固定预算）
        tool_tokens = sum(self.count_tokens(str(t)) for t in tool_definitions)
        if tool_tokens > self.budget.tool_defs_max:
            tool_definitions = self._trim_tools(
                tool_definitions, self.budget.tool_defs_max
            )

        # 3. 对话历史（从新到旧，按重要性保留）
        conv_tokens = 0
        kept_messages = []
        for msg in reversed(conversation_history):
            msg_tokens = self.count_tokens(str(msg))
            if conv_tokens + msg_tokens > self.budget.conversation_max:
                break
            kept_messages.insert(0, msg)
            conv_tokens += msg_tokens

        # 如果对话历史被截断，添加摘要
        if len(kept_messages) < len(conversation_history):
            summary = self._summarize_old_messages(
                conversation_history[:len(conversation_history) - len(kept_messages)]
            )
            kept_messages.insert(0, {
                "role": "system",
                "content": f"[历史对话摘要] {summary}"
            })

        messages.extend(kept_messages)

        # 4. 检索知识（最低优先级，按相关性排序填充剩余空间）
        used_tokens = sum(self.count_tokens(str(m)) for m in messages)
        remaining = self.budget.retrieval_max
        for knowledge in retrieved_knowledge:
            k_tokens = self.count_tokens(knowledge)
            if k_tokens > remaining:
                break
            messages.append({
                "role": "system",
                "content": f"[检索知识] {knowledge}"
            })
            remaining -= k_tokens

        return messages

    def _truncate(self, text: str, max_tokens: int) -> str:
        tokens = self.encoder.encode(text)[:max_tokens]
        return self.encoder.decode(tokens)

    def _trim_tools(self, tools: List[Dict], max_tokens: int) -> List[Dict]:
        """按使用频率裁剪低频工具"""
        # 简化实现：按顺序保留直到超出预算
        result = []
        used = 0
        for tool in tools:
            t = self.count_tokens(str(tool))
            if used + t > max_tokens:
                break
            result.append(tool)
            used += t
        return result

    def _summarize_old_messages(self, messages: List[Dict]) -> str:
        """用 LLM 压缩旧对话为摘要"""
        # 实际实现中调用小模型（如 GPT-4o-mini）
        content = " ".join(m.get("content", "") for m in messages[-10:])
        return f"[{len(messages)} 条历史消息的摘要] 讨论了：{content[:500]}..."

/**
 * MemGPT 风格的虚拟上下文管理器
 * 核心概念：将上下文窗口视为「主存」，向量数据库视为「辅存」
 * 通过页面置换算法在两者之间搬运信息
 */

interface MemoryPage {
  id: string;
  content: string;
  tokens: number;
  lastAccessed: number;   // LRU 淘汰依据
  accessCount: number;    // 频率统计
  importance: number;     // 0-1 重要性评分
  category: "conversation" | "knowledge" | "skill" | "metadata";
}

class VirtualMemoryManager {
  private mainMemory: MemoryPage[] = [];   // 上下文窗口（主存）
  private archivalMemory: MemoryPage[] = []; // 向量数据库（辅存）
  private maxMainTokens: number;

  constructor(maxContextTokens: number = 128000) {
    // 预留 30% 给输出和系统 prompt
    this.maxMainTokens = Math.floor(maxContextTokens * 0.7);
  }

  /** 访问页面：如果在主存中则刷新 LRU，否则从辅存加载 */
  async access(pageId: string): Promise<MemoryPage | null> {
    const inMain = this.mainMemory.find(p => p.id === pageId);
    if (inMain) {
      inMain.lastAccessed = Date.now();
      inMain.accessCount++;
      return inMain;
    }

    // 页面不在主存 → 从辅存加载
    const inArchival = this.archivalMemory.find(p => p.id === pageId);
    if (!inArchival) return null;

    // 检查主存是否有空间
    const usedTokens = this.mainMemory.reduce((s, p) => s + p.tokens, 0);
    if (usedTokens + inArchival.tokens > this.maxMainTokens) {
      // 需要换出一个页面
      await this.evictPage();
    }

    // 加载到主存
    inArchival.lastAccessed = Date.now();
    inArchival.accessCount++;
    this.mainMemory.push(inArchival);
    this.archivalMemory = this.archivalMemory.filter(p => p.id !== pageId);
    return inArchival;
  }

  /** 页面置换：使用加权 LRU 算法（综合考虑时间、频率、重要性） */
  private async evictPage(): Promise<void> {
    if (this.mainMemory.length === 0) return;

    // 计算每个页面的综合得分（越低越优先换出）
    const now = Date.now();
    let minScore = Infinity;
    let victimIndex = 0;

    this.mainMemory.forEach((page, i) => {
      const ageHours = (now - page.lastAccessed) / 3600000;
      const recencyScore = 1 / (1 + ageHours);        // 越近越高
      const frequencyScore = Math.log(1 + page.accessCount); // 对数压缩
      const importanceScore = page.importance;

      // 综合得分 = 重要性 × (时间衰减 + 频率)
      const score = importanceScore * (recencyScore + frequencyScore * 0.3);

      if (score < minScore) {
        minScore = score;
        victimIndex = i;
      }
    });

    // 换出到辅存
    const victim = this.mainMemory.splice(victimIndex, 1)[0];
    this.archivalMemory.push(victim);
  }

  /** 语义检索：从辅存中搜索相关记忆 */
  async semanticSearch(query: string, topK: number = 5): Promise<MemoryPage[]> {
    // 实际实现中使用向量数据库（如 Qdrant/Pinecone）
    // 这里简化为按关键词匹配
    const scored = this.archivalMemory.map(page => ({
      page,
      score: this.computeRelevance(query, page.content),
    }));
    scored.sort((a, b) => b.score - a.score);
    return scored.slice(0, topK).map(s => s.page);
  }

  private computeRelevance(query: string, content: string): number {
    // 简化实现：实际使用 embedding 余弦相似度
    const queryWords = new Set(query.toLowerCase().split(/\s+/));
    const contentWords = content.toLowerCase().split(/\s+/);
    let matches = 0;
    for (const w of contentWords) {
      if (queryWords.has(w)) matches++;
    }
    return matches / Math.max(queryWords.size, 1);
  }
}

"""
情景记忆存储系统
支持：存储、检索、去重、冲突检测、自动压缩
"""

import uuid
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum


class MemoryTier(Enum):
    """记忆存储层级"""
    HOT = "hot"          # 最近 7 天，完整内容
    WARM = "warm"        # 7-30 天，压缩摘要
    COLD = "cold"        # 30+ 天，仅关键事实
    ARCHIVED = "archived" # 90+ 天，归档


@dataclass
class EpisodicMemory:
    """单条情景记忆"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=datetime.now)
    session_id: str = ""
    user_id: str = ""
    content: str = ""                    # 原始内容
    summary: str = ""                    # 压缩摘要
    key_facts: List[str] = field(default_factory=list)  # 关键事实
    entities: List[str] = field(default_factory=list)   # 涉及的实体
    topics: List[str] = field(default_factory=list)     # 涉及的主题
    importance: float = 0.5             # 重要性评分 (0-1)
    tier: MemoryTier = MemoryTier.HOT
    embedding: Optional[List[float]] = None
    access_count: int = 0
    last_accessed: Optional[datetime] = None
    supersedes: Optional[str] = None    # 替代了哪条旧记忆


class EpisodicMemoryStore:
    """情景记忆存储管理器"""

    def __init__(self):
        self.memories: Dict[str, EpisodicMemory] = {}
        self.tier_transitions = {
            MemoryTier.HOT: timedelta(days=7),
            MemoryTier.WARM: timedelta(days=23),   # 7+23=30
            MemoryTier.COLD: timedelta(days=60),    # 30+60=90
        }

    def store(self, memory: EpisodicMemory) -> str:
        """存储新记忆，自动检测冲突和重复"""

        # 1. 检查是否与已有记忆冲突
        conflict = self._detect_conflict(memory)
        if conflict:
            # 新记忆替代旧记忆
            memory.supersedes = conflict.id
            self.memories[conflict.id] = conflict  # 保留但标记为已替代

        # 2. 检查是否重复（语义相似度 > 0.95）
        if self._is_duplicate(memory):
            # 更新已有记忆的访问计数
            existing = self._find_most_similar(memory)
            existing.access_count += 1
            existing.last_accessed = datetime.now()
            return existing.id

        # 3. 自动提取关键事实
        memory.key_facts = self._extract_key_facts(memory.content)
        memory.entities = self._extract_entities(memory.content)
        memory.summary = self._generate_summary(memory.content)

        # 4. 存储
        self.memories[memory.id] = memory
        return memory.id

    def retrieve(
        self,
        query: str,
        user_id: str,
        top_k: int = 5,
        time_weight: float = 0.3,
        importance_weight: float = 0.2,
    ) -> List[EpisodicMemory]:
        """混合检索：向量相似度 + 时间衰减 + 重要性加权"""

        candidates = [m for m in self.memories.values()
                      if m.user_id == user_id and not m.supersedes]

        scored = []
        for memory in candidates:
            # 向量相似度（简化为关键词匹配）
            sim_score = self._compute_similarity(query, memory)

            # 时间衰减（指数衰减，半衰期 7 天）
            age_days = (datetime.now() - memory.timestamp).days
            time_score = 0.5 ** (age_days / 7)

            # 重要性（直接取值）
            imp_score = memory.importance

            # 访问频率加成
            freq_score = min(memory.access_count / 10, 1.0) * 0.1

            # 综合得分
            total = (sim_score * 0.5 +
                     time_score * time_weight +
                     imp_score * importance_weight +
                     freq_score)

            scored.append((memory, total))

        scored.sort(key=lambda x: x[1], reverse=True)
        results = [m for m, _ in scored[:top_k]]

        # 更新访问记录
        for m in results:
            m.access_count += 1
            m.last_accessed = datetime.now()

        return results

    def _detect_conflict(self, new: EpisodicMemory) -> Optional[EpisodicMemory]:
        """检测新记忆是否与已有记忆冲突"""
        for m in self.memories.values():
            if m.user_id != new.user_id:
                continue
            # 检查实体重叠 + 内容矛盾
            entity_overlap = set(m.entities) & set(new.entities)
            if entity_overlap and self._has_contradiction(m.content, new.content):
                return m
        return None

    def _is_duplicate(self, memory: EpisodicMemory) -> bool:
        """检查是否为重复记忆"""
        similar = self._find_most_similar(memory)
        if similar:
            sim = self._compute_similarity(memory, similar)
            return sim > 0.95
        return False

    def _find_most_similar(self, memory: EpisodicMemory) -> Optional[EpisodicMemory]:
        best, best_score = None, 0
        for m in self.memories.values():
            if m.user_id == memory.user_id:
                score = self._compute_similarity(memory, m)
                if score > best_score:
                    best, best_score = m, score
        return best

    def _compute_similarity(self, a, b) -> float:
        """简化相似度计算（实际用 embedding 余弦相似度）"""
        if isinstance(a, EpisodicMemory):
            words_a = set(a.content.lower().split())
        else:
            words_a = set(a.lower().split())
        if isinstance(b, EpisodicMemory):
            words_b = set(b.content.lower().split())
        else:
            words_b = set(b.lower().split())
        if not words_a or not words_b:
            return 0.0
        return len(words_a & words_b) / len(words_a | words_b)

    def _extract_key_facts(self, content: str) -> List[str]:
        """用 LLM 从内容中提取关键事实"""
        # 实际实现中调用 LLM
        return [content[:200]]

    def _extract_entities(self, content: str) -> List[str]:
        """NER 实体提取"""
        # 简化实现
        return []

    def _generate_summary(self, content: str) -> str:
        """生成摘要"""
        return content[:300] if len(content) > 300 else content

    def _has_contradiction(self, old_content: str, new_content: str) -> bool:
        """检测两条内容是否矛盾"""
        # 实际实现中用 NLI 模型
        return False

    def run_tier_migration(self):
        """定期执行记忆层级迁移"""
        now = datetime.now()
        for memory in self.memories.values():
            age = now - memory.timestamp
            if memory.tier == MemoryTier.HOT and age > self.tier_transitions[MemoryTier.HOT]:
                memory.tier = MemoryTier.WARM
                memory.content = memory.summary  # 压缩为摘要
            elif memory.tier == MemoryTier.WARM and age > timedelta(days=30):
                memory.tier = MemoryTier.COLD
                memory.content = "; ".join(memory.key_facts)  # 仅保留关键事实
            elif memory.tier == MemoryTier.COLD and age > timedelta(days=90):
                memory.tier = MemoryTier.ARCHIVED