Upload agentframe_quad.py with huggingface_hub

agentframe_quad.py

@@ -13,6 +13,7 @@ AgentFrame 四层融合实现
   L4 KVPager      — 物理层: 热/温/冷三级换页 + 预测性驱逐
 """
 import numpy as np
+import math
 from dataclasses import dataclass, field
 from typing import Optional, Literal
 
@@ -29,6 +30,8 @@ class CompressedKV:
     heat: float = 0.5           # 热度 (物理层)
     last_access: float = 0.0    # 最后访问时间
     size_bytes: int = 0         # 实际字节数
+    importance: float = 0.5     # 重要性 (遗忘曲线用, 0-1)
+    access_count: int = 0       # 访问次数 (遗忘曲线用)
 
 
 class AbsorbedMLA:
@@ -179,10 +182,51 @@ class LandmarkRouter:
 # ============================================================
 # L4 物理层: KV 换页 + 预测性驱逐
 # ============================================================
+# ============================================================
+# L4 物理层: KV 换页 + 预测性驱逐 (含遗忘曲线)
+# ============================================================
+class ForgettingCurve:
+    """
+    遗忘曲线 (源自 Project-Aura): 决定 KV 块的热度衰减
+
+    公式: S(t) = I · 2^(-t / τ)
+    其中:
+      S(t) = t 时刻的记忆强度 (热度)
+      I = 初始重要性
+      τ = 半衰期 (高频访问的块半衰期延长)
+    访问增强: boost = log2(access_count+1) × 0.1
+    """
+    def __init__(self, default_half_life: float = 100.0):
+        # KV 场景时间尺度小 (轮次而非天), 默认 100 轮
+        self.default_half_life = default_half_life
+
+    def compute_half_life(self, kv: CompressedKV) -> float:
+        """根据访问频率动态调整半衰期"""
+        base = self.default_half_life
+        if kv.access_count > 10:
+            base *= 7                           # 高频访问延至 7×
+        elif kv.access_count > 5:
+            base *= 3                           # 中频延至 3×
+        base *= (0.5 + kv.importance)           # 高重要性更持久
+        return base
+
+    def strength(self, kv: CompressedKV, now: float) -> float:
+        """记忆强度: 指数衰减 + 访问增强 (替代简单线性热度)"""
+        elapsed = max(now - kv.last_access, 0.0)
+        half_life = self.compute_half_life(kv)
+        decay = math.pow(2, -elapsed / half_life)
+        boost = math.log2(kv.access_count + 1) * 0.1
+        return min(1.0, kv.importance * decay + boost)
+
+    def should_forget(self, kv: CompressedKV, now: float, threshold: float = 0.15) -> bool:
+        """强度低于阈值 → 该换出"""
+        return self.strength(kv, now) < threshold
+
+
 class KVPager:
     """
     三级存储: 显存(热) / 内存(温) / 磁盘(冷, mmap)
-    预测性驱逐: 注意力衰减 + 访问频率 + 语义相关
+    预测性驱逐: 遗忘曲线强度 + 注意力衰减
     """
     def __init__(self, vram_limit_mb=10240, ram_limit_mb=32768):
         self.vram_limit = vram_limit_mb * 1024 * 1024
@@ -193,16 +237,18 @@ class KVPager:
         self.vram_used = 0
         self.ram_used = 0
         self.access_log = []  # (chunk_id, time, attention_score)
+        self.curve = ForgettingCurve()  # 遗忘曲线 (Aura 移植)
 
     def place(self, kv: CompressedKV, now: float = 0.0):
-        """放置: 按热度分层"""
-        if kv.heat >= 0.8:
+        """放置: 按遗忘曲线强度分层"""
+        strength = self.curve.strength(kv, now)
+        kv.heat = strength
+        if strength >= 0.6:
             if self.vram_used + kv.size_bytes <= self.vram_limit:
                 self.vram[kv.chunk_id] = kv
                 self.vram_used += kv.size_bytes
                 return
-            kv.heat = 0.6
-        if kv.heat >= 0.4:
+        if strength >= 0.3:
             if self.ram_used + kv.size_bytes <= self.ram_limit:
                 self.ram[kv.chunk_id] = kv
                 self.ram_used += kv.size_bytes
@@ -210,31 +256,31 @@ class KVPager:
         self.disk[kv.chunk_id] = kv
 
     def access(self, chunk_id: int, attention_score: float, now: float):
-        """访问: 更新热度 + 日志"""
+        """访问: 更新访问计数 + 重要性 + 日志"""
         self.access_log.append((chunk_id, now, attention_score))
         kv = self.vram.get(chunk_id) or self.ram.get(chunk_id) or self.disk.get(chunk_id)
         if kv:
-            kv.heat = min(1.0, kv.heat + 0.15)
+            kv.access_count += 1
+            kv.importance = min(1.0, kv.importance + abs(attention_score) * 0.05)
             kv.last_access = now
+            # 重新计算强度并分层
+            strength = self.curve.strength(kv, now)
+            kv.heat = strength
             # 磁盘 → 提升
-            if chunk_id in self.disk and kv.heat >= 0.5:
+            if chunk_id in self.disk and strength >= 0.5:
                 self.disk.pop(chunk_id)
-                kv.heat = 0.6
                 self.place(kv, now)
 
     def eviction_score(self, kv: CompressedKV, now: float) -> float:
-        """预测性驱逐分数 (越高越该驱逐)"""
+        """预测性驱逐分数 (越高越该驱逐): 遗忘曲线 + 注意��衰减"""
+        # 遗忘曲线: 强度越低越该走 (主因子)
+        forget_factor = 1.0 - self.curve.strength(kv, now)
         # 注意力衰减: 最近分数越低越该走
         logs = [a for cid, t, a in self.access_log if cid == kv.chunk_id]
-        if not logs:
-            return 0.5
-        attn_decay = 1.0 - np.mean(logs[-5:])  # 最近注意力均值
-        # 访问频率: 越频繁越留
-        freq = len(logs) / max(now, 1.0)
-        access_factor = min(freq * 5, 1.0)
+        attn_decay = 1.0 - np.mean(logs[-5:]) if logs else 0.5
         # 时间衰减: 越久没访问越该走
-        time_decay = min((now - kv.last_access) / 100.0, 1.0)
-        return 0.5 * attn_decay + 0.3 * (1 - access_factor) + 0.2 * time_decay
+        time_decay = min((now - kv.last_access) / self.curve.compute_half_life(kv), 1.0)
+        return 0.5 * forget_factor + 0.3 * attn_decay + 0.2 * time_decay
 
     def evict(self, now: float, target_layer: str = "vram"):
         """驱逐: 选分数最高的换出"""
@@ -245,11 +291,9 @@ class KVPager:
         victim = pool.pop(victim_id)
         if target_layer == "vram":
             self.vram_used -= victim.size_bytes
-            victim.heat = 0.5
-            self.place(victim, now)  # 降级到 RAM/DISK
+            self.place(victim, now)  # 降级到 RAM/DISK (按遗忘曲线自动分)
         else:
             self.ram_used -= victim.size_bytes
-            victim.heat = 0.2
             self.disk[victim_id] = victim
         return victim_id