解析 ‘Temporal Context Weighting’：在图中如何为三年前的记忆与三秒前的记忆分配不同的逻辑权重？

各位同仁，下午好！

今天，我们齐聚一堂，探讨一个在人工智能和软件工程领域都至关重要的话题：时间上下文加权（Temporal Context Weighting）。在我们的日常生活中，时间的流逝赋予了事件不同的意义和重要性。三秒前发生的事情，比如你刚刚说的一句话，往往对当前的对话至关重要；而三年前的记忆，比如你三年前早餐吃了什么，通常则无关紧要，除非它被某种特定情境再次激活。

在构建智能系统时，我们如何让机器也具备这种对时间敏感的“常识”？如何为三年前的记忆与三秒前的记忆分配不同的逻辑权重，以确保系统决策的及时性、相关性和准确性？这正是时间上下文加权的核心挑战与魅力所在。

作为编程专家，我们不仅要理解其背后的理论，更要能够将其转化为可执行的代码，构建出能够高效处理时间序列信息，并从中提取出有意义上下文的智能系统。本次讲座，我将从理论到实践，深入剖析这一主题，并辅以代码示例，希望能够为大家带来启发。

一、 时间上下文加权的必要性：为什么时间很重要？

在许多AI应用中，数据并非孤立存在，而是以序列的形式出现，带有明确的时间戳。例如：

• 对话系统（Chatbots）：用户最近的几句话决定了当前对话的焦点。而几小时前的对话内容，可能需要被淡化。
• 推荐系统（Recommendation Systems）：用户最近的购买行为或浏览历史，往往比一年前的数据更能预测其当前的兴趣。
• 自动驾驶（Autonomous Driving）：传感器在毫秒级更新的数据流，决定了车辆的即时决策。几秒前的障碍物信息可能已经过时。
• 个性化学习（Personalized Learning）：学生最近的学习进度和遇到的难题，比其早期学习记录更能反映当前的知识掌握状态。
• 金融欺诈检测（Financial Fraud Detection）：异常交易模式的识别往往依赖于实时或近期的交易数据。

在这些场景中，简单地将所有历史数据一视同仁，会带来两个主要问题：

1. 信息过载与噪声：无关紧要的旧信息会干扰系统对当前情况的判断。
2. 计算效率低下：处理大量可能已不相关的数据，会浪费宝贵的计算资源。

因此，我们需要一种机制来动态调整不同时间点信息的“重要性”或“权重”，使其能够反映其对当前任务的实际贡献。这就是时间上下文加权的核心目标。

二、 记忆与上下文的系统化定义

在深入探讨加权机制之前，我们首先需要明确在系统语境下，“记忆”和“上下文”的含义。

1. 记忆（Memory）

在我们的讨论中，“记忆”可以被定义为系统捕获到的，带有时间戳的离散或连续事件。它是一个数据单元，包含了：

• 时间戳（Timestamp）：事件发生的确切时间，通常以UNIX时间戳、ISO 8601格式或其他高精度时间表示。这是我们进行时间加权的基础。
• 内容（Content）：事件的实际信息，可以是文本、数值、向量、对象ID等。例如：
  • 用户在聊天中说的一句话。
  • 传感器读取的一个温度值。
  • 用户购买了一个商品。
  • 系统做出的一个决策。
• 类型（Type，可选）：记忆的类别，这有助于我们应用不同的加权策略。例如，用户偏好记忆可能比用户短暂兴趣记忆具有更慢的衰减率。

我们可以将记忆存储在一个结构中，例如一个字典或一个类实例。

import datetime
from typing import Any, Dict, Optional

class SystemMemory:
    """
    系统中的一个记忆单元。
    """
    def __init__(self, content: Any, timestamp: Optional[datetime.datetime] = None, memory_type: str = "general"):
        self.content = content
        self.timestamp = timestamp if timestamp is not None else datetime.datetime.now(datetime.timezone.utc)
        self.memory_type = memory_type
        # 初始权重可以在创建时设定，但通常在检索时动态计算
        self.initial_weight = 1.0

    def __repr__(self):
        return (f"Memory(content='{self.content}', timestamp={self.timestamp.isoformat()}, "
                f"type='{self.memory_type}')")

# 示例：
mem1 = SystemMemory("用户说 '你好'", timestamp=datetime.datetime.now(datetime.timezone.utc))
mem2 = SystemMemory("用户购买了 '智能音箱'", timestamp=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=365*3), memory_type="preference")
mem3 = SystemMemory("用户点击了 'T恤'", timestamp=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=5))

print(mem1)
print(mem2)
print(mem3)

2. 上下文（Context）

“上下文”则是指在某一特定时刻，系统为了完成当前任务而需要考虑的相关记忆的集合。这个集合中的每个记忆都将根据其时间戳以及其他因素被赋予一个逻辑权重。上下文可以是：

• 一个加权后的记忆向量：通过对多个记忆的数值表示进行加权平均或加权求和得到。
• 一个加权的记忆列表：每个记忆都带有一个明确的权重值，供后续模块使用。
• 一个动态的注意力分布：在深度学习模型中，注意力机制会学习如何为不同的输入（包括时间序列输入）分配权重。

上下文的构建是动态的，它会随着时间的推移和新事件的发生而不断演变。

三、 时间加权的基本原则与数学模型

时间加权的核心思想是：距离当前时刻越近的记忆，其权重通常越高；距离越远的记忆，其权重通常越低。 这种权重衰减并非总是线性的，它可能呈现指数级、多项式或其他形式的衰减。

1. 衰减函数的选择

选择合适的衰减函数至关重要，它决定了权重随时间变化的曲线形态。

• 线性衰减（Linear Decay）：

  • 最简单直观的衰减方式。权重随时间均匀减少。
  • 公式：w(Δt) = max(0, 1 - α * Δt)
    • Δt 是时间差（当前时间 – 记忆时间）。
    • α 是衰减率，一个正数，决定了权重下降的速度。当 1 - α * Δt 小于0时，权重设为0。
  • 优点：简单易理解和实现。
  • 缺点：在现实世界中，信息的价值往往不会如此均匀地下降，可能在初期保持较高，随后迅速下降，或者反之。

• 指数衰减（Exponential Decay）：

  • 最常用且在许多自然现象（如放射性衰变、遗忘曲线）中观察到的衰减模式。权重随时间呈指数级下降。
  • 公式：w(Δt) = e^(-λ * Δt) 或 w(Δt) = A * e^(-λ * Δt)
    • Δt 是时间差。
    • λ (lambda) 是衰减常数，一个正数，决定了衰减的速度。λ 越大，衰减越快。
    • A 是初始权重（通常设为1）。
  • 优点：符合直觉，能够很好地模拟“遗忘”效应，对近期事件赋予高权重，对遥远事件赋予极低但非零的权重。
  • 缺点：对 λ 的选择敏感，不同的 λ 会导致非常不同的衰减曲线。

• 多项式衰减（Polynomial Decay）：

  • 一种更灵活的衰减方式，介于线性和指数之间。
  • 公式：w(Δt) = 1 / (1 + Δt^p)
    • Δt 是时间差。
    • p 是衰减幂次，一个正数。p 越大，衰减越快。
  • 优点：可以通过调整 p 来控制衰减的陡峭程度。
  • 缺点：当 Δt 较小时，衰减可能不明显；当 Δt 较大时，衰减速度可能过快或过慢，需要仔细调参。

• Sigmoid 衰减（Sigmoid-based Decay）：

  • Sigmoid 函数本身是一个S形曲线，通常用于将值映射到0到1之间。我们可以使用其变体或倒置来模拟衰减。
  • 例如，一个倒置的Sigmoid函数：w(Δt) = 1 - (1 / (1 + e^(-k * (Δt - t_0))))
    • k 控制曲线的陡峭程度。
    • t_0 控制衰减的中心点。
  • 优点：可以模拟在某个时间点之前保持较高权重，之后迅速衰减的模式。
  • 缺点：参数更多，调整更复杂。

2. 时间单位的统一

在计算 Δt 时，务必确保时间单位的一致性。例如，如果 λ 是每秒的衰减率，那么 Δt 也必须是秒。

import datetime
import math

def calculate_weight(memory_timestamp: datetime.datetime, current_time: datetime.datetime,
                       decay_function: str = "exponential", **kwargs) -> float:
    """
    根据时间差计算记忆的逻辑权重。
    Args:
        memory_timestamp: 记忆发生的时间戳。
        current_time: 当前时间戳。
        decay_function: 衰减函数类型 ('linear', 'exponential', 'polynomial', 'sigmoid')。
        **kwargs: 衰减函数所需的参数。
    Returns:
        float: 计算出的权重，范围通常在 [0, 1] 之间。
    """
    time_delta = (current_time - memory_timestamp).total_seconds()

    # 如果记忆发生在未来（不应该发生），或时间差为负，返回0
    if time_delta < 0:
        return 0.0

    if decay_function == "linear":
        alpha = kwargs.get("alpha", 0.001) # 默认每秒衰减0.001
        weight = max(0.0, 1.0 - alpha * time_delta)
    elif decay_function == "exponential":
        lambda_val = kwargs.get("lambda_val", 0.01) # 默认衰减常数0.01
        weight = math.exp(-lambda_val * time_delta)
    elif decay_function == "polynomial":
        p = kwargs.get("p", 2.0) # 默认幂次为2
        weight = 1.0 / (1.0 + time_delta**p)
    elif decay_function == "sigmoid":
        k = kwargs.get("k", 0.1)
        t0 = kwargs.get("t0", 3600.0) # 默认中心衰减点为1小时 (3600秒)
        # 倒置sigmoid，使时间越长权重越小
        weight = 1.0 - (1.0 / (1.0 + math.exp(-k * (time_delta - t0))))
        # 确保权重不会低于0
        weight = max(0.0, weight)
    else:
        raise ValueError(f"Unknown decay function: {decay_function}")

    return weight

# 假设当前时间
now = datetime.datetime.now(datetime.timezone.utc)

# 记忆时间点
three_seconds_ago = now - datetime.timedelta(seconds=3)
one_minute_ago = now - datetime.timedelta(minutes=1)
one_hour_ago = now - datetime.timedelta(hours=1)
one_day_ago = now - datetime.timedelta(days=1)
three_years_ago = now - datetime.timedelta(days=365 * 3)

# 打印不同衰减函数下的权重
memories_to_test = [
    ("三秒前", three_seconds_ago),
    ("一分钟前", one_minute_ago),
    ("一小时前", one_hour_ago),
    ("一天前", one_day_ago),
    ("三年前", three_years_ago)
]

print("--- 线性衰减 (alpha=0.000001) ---") # 调整alpha使其在较长时间内不完全衰减
for label, ts in memories_to_test:
    weight = calculate_weight(ts, now, decay_function="linear", alpha=0.000001)
    print(f"{label}: 权重 = {weight:.6f}")

print("n--- 指数衰减 (lambda=0.00001) ---") # 调整lambda使其在较长时间内不完全衰减
for label, ts in memories_to_test:
    weight = calculate_weight(ts, now, decay_function="exponential", lambda_val=0.00001)
    print(f"{label}: 权重 = {weight:.6f}")

print("n--- 多项式衰减 (p=0.5) ---") # 调整p使其在较长时间内不完全衰减
for label, ts in memories_to_test:
    weight = calculate_weight(ts, now, decay_function="polynomial", p=0.5)
    print(f"{label}: 权重 = {weight:.6f}")

print("n--- Sigmoid 衰减 (k=0.000005, t0=86400*30) ---") # k和t0参数需要根据时间尺度调整，这里设置为30天为中心
for label, ts in memories_to_test:
    weight = calculate_weight(ts, now, decay_function="sigmoid", k=0.000005, t0=86400*30)
    print(f"{label}: 权重 = {weight:.6f}")

输出示例 (实际数值会因当前时间而略有不同):

--- 线性衰减 (alpha=0.000001) ---
三秒前: 权重 = 0.999997
一分钟前: 权重 = 0.999940
一小时前: 权重 = 0.996400
一天前: 权重 = 0.913600
三年前: 权重 = 0.000000

--- 指数衰减 (lambda=0.00001) ---
三秒前: 权重 = 0.999970
一分钟前: 权重 = 0.999400
一小时前: 权重 = 0.964595
一天前: 权重 = 0.170643
三年前: 权重 = 0.000000

--- 多项式衰减 (p=0.5) ---
三秒前: 权重 = 0.577350
一分钟前: 权重 = 0.117851
一小时前: 权重 = 0.038318
一天前: 权重 = 0.006093
三年前: 权重 = 0.000000

--- Sigmoid 衰减 (k=0.000005, t0=2592000.0) ---
三秒前: 权重 = 0.047425
一分钟前: 权重 = 0.047425
一小时前: 权重 = 0.047425
一天前: 权重 = 0.047425
三年前: 权重 = 0.000000

从上面的例子可以看出，通过调整参数，我们可以控制各种衰减函数的行为。特别是对于长时间跨度，如“三年前”，线性衰减和指数衰减在合适的参数下都能迅速将权重降至接近0，而多项式衰减则会更快。Sigmoid衰减则更复杂，其t0参数（衰减中心点）对权重影响显著，使得在t0之前权重会保持较高，之后急剧下降。

四、 实践中的时间加权：数据结构与算法

在实际系统中，我们需要一个能够存储记忆并高效检索其加权上下文的机制。

1. 记忆存储（Memory Store）

一个简单的记忆存储可以使用列表或 collections.deque（双端队列），后者在需要固定大小的上下文窗口时特别有用。

from collections import deque
import datetime
from typing import List, Tuple

class MemoryStore:
    """
    一个简单的记忆存储，可以添加和检索带有时间加权的记忆。
    """
    def __init__(self, max_size: Optional[int] = None):
        self.memories: deque[SystemMemory] = deque(maxlen=max_size)
        self.max_size = max_size

    def add_memory(self, content: Any, memory_type: str = "general"):
        """添加一个新记忆，自动打上当前时间戳。"""
        new_memory = SystemMemory(content, memory_type=memory_type)
        self.memories.append(new_memory)
        print(f"Added memory: {new_memory}")

    def get_weighted_context(self, current_time: Optional[datetime.datetime] = None,
                             decay_function: str = "exponential", **decay_params) -> List[Tuple[SystemMemory, float]]:
        """
        检索所有记忆，并根据时间差计算它们的权重。
        Args:
            current_time: 用于计算时间差的当前时间。如果为None，则使用datetime.now()。
            decay_function: 衰减函数类型。
            **decay_params: 衰减函数所需的参数。
        Returns:
            List[Tuple[SystemMemory, float]]: 记忆及其对应权重的列表，按权重降序排列。
        """
        if current_time is None:
            current_time = datetime.datetime.now(datetime.timezone.utc)

        weighted_memories = []
        for mem in self.memories:
            weight = calculate_weight(mem.timestamp, current_time, decay_function, **decay_params)
            # 过滤掉权重过低的记忆，以提高效率
            if weight > 0.0001: # 设定一个阈值
                weighted_memories.append((mem, weight))

        # 通常按权重降序排列，确保最相关的记忆在前
        weighted_memories.sort(key=lambda x: x[1], reverse=True)
        return weighted_memories

    def clear(self):
        """清空所有记忆。"""
        self.memories.clear()
        print("Memory store cleared.")

# 示例使用
memory_store = MemoryStore(max_size=100)

# 添加一些记忆
memory_store.add_memory("用户登录系统")
memory_store.add_memory("用户浏览了商品A", memory_type="interaction")
memory_store.add_memory("用户点击了购物车按钮", memory_type="interaction")
# 模拟一些更早的记忆
older_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(minutes=5)
memory_store.memories.append(SystemMemory("用户搜索了 '智能手表'", timestamp=older_time))
even_older_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=2)
memory_store.memories.append(SystemMemory("用户购买了 '耳机'", timestamp=even_older_time, memory_type="purchase"))
three_years_ago_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=365*3)
memory_store.memories.append(SystemMemory("用户注册了账号", timestamp=three_years_ago_time, memory_type="system_event"))

print("n--- 获取加权上下文 (指数衰减 λ=0.001, 每秒衰减) ---")
# λ=0.001 意味着 1000秒后（约16.6分钟）权重降至 e^-1 ≈ 0.368
# 10秒后权重 e^(-0.001*10) = 0.99
# 1分钟后权重 e^(-0.001*60) = 0.94
# 1小时后权重 e^(-0.001*3600) = 0.03
# 1天后权重 e^(-0.001*86400) = ~0
weighted_ctx = memory_store.get_weighted_context(decay_function="exponential", lambda_val=0.001)

for mem, weight in weighted_ctx:
    print(f"[{weight:.4f}] {mem}")

print("n--- 获取加权上下文 (线性衰减 α=0.000005, 每秒衰减) ---")
# α=0.000005 意味着 1/0.000005 = 200000 秒后（约2.3天）权重降至0
weighted_ctx_linear = memory_store.get_weighted_context(decay_function="linear", alpha=0.000005)
for mem, weight in weighted_ctx_linear:
    print(f"[{weight:.4f}] {mem}")

输出示例 (实际数值会因当前时间而略有不同):

Added memory: Memory(content='用户登录系统', timestamp=2023-10-27T10:00:00.123456+00:00, type='general')
Added memory: Memory(content='用户浏览了商品A', timestamp=2023-10-27T10:00:00.234567+00:00, type='interaction')
Added memory: Memory(content='用户点击了购物车按钮', timestamp=2023-10-27T10:00:00.345678+00:00, type='interaction')

--- 获取加权上下文 (指数衰减 λ=0.001, 每秒衰减) ---
[1.0000] Memory(content='用户点击了购物车按钮', timestamp=2023-10-27T10:00:00.345678+00:00, type='interaction')
[1.0000] Memory(content='用户浏览了商品A', timestamp=2023-10-27T10:00:00.234567+00:00, type='interaction')
[1.0000] Memory(content='用户登录系统', timestamp=2023-10-27T10:00:00.123456+00:00, type='general')
[0.9690] Memory(content='用户搜索了 '智能手表'', timestamp=2023-10-27T09:55:00.000000+00:00, type='general')
[0.3292] Memory(content='用户购买了 '耳机'', timestamp=2023-10-27T08:00:00.000000+00:00, type='purchase')

--- 获取加权上下文 (线性衰减 α=0.000005, 每秒衰减) ---
[1.0000] Memory(content='用户点击了购物车按钮', timestamp=2023-10-27T10:00:00.345678+00:00, type='interaction')
[1.1000] Memory(content='用户浏览了商品A', timestamp=2023-10-27T10:00:00.234567+00:00, type='interaction') # 注意：这里权重可能因time_delta过小而导致显示为1.0000
[1.0000] Memory(content='用户登录系统', timestamp=2023-10-27T10:00:00.123456+00:00, type='general')
[0.9985] Memory(content='用户搜索了 '智能手表'', timestamp=2023-10-27T09:55:00.000000+00:00, type='general')
[0.9640] Memory(content='用户购买了 '耳机'', timestamp=2023-10-27T08:00:00.000000+00:00, type='purchase')

（注：上述输出中，由于当前时间与记忆时间差极小，部分线性衰减权重可能显示为1.0000。对于三年前的记忆，在上述参数下，其权重会远低于0.0001的阈值，因此不会被包含在结果中。）

2. 上下文聚合（Context Aggregation）

在获取到加权记忆列表后，下一步通常是将这些记忆聚合成一个可用于下游任务的单一上下文表示。

• 加权平均/求和：如果记忆可以表示为数值向量（例如通过词嵌入或特征提取），那么可以直接对这些向量进行加权平均或加权求和。

  import numpy as np
  
  def aggregate_context_vector(weighted_memories: List[Tuple[SystemMemory, float]]) -> Optional[np.ndarray]:
      """
      将加权记忆聚合成一个上下文向量。
      假设每个记忆内容都可以转换为一个数值向量。
      """
      if not weighted_memories:
          return None
  
      # 假设记忆内容是字符串，我们这里用一个简化的方式将其转换为向量
      # 实际应用中会使用Word2Vec, BERT等嵌入模型
      def content_to_vector(content: Any) -> np.ndarray:
          if isinstance(content, str):
              # 简单示例：将字符串转换为一个固定维度的向量，例如基于哈希
              # 实际应用中会是预训练的词向量
              vector_dim = 10
              hash_val = hash(content)
              return np.array([(hash_val >> (i * 8)) & 0xFF for i in range(vector_dim)]) / 255.0
          return np.zeros(10) # 默认向量
  
      context_vector = None
      total_weight = 0.0
  
      for mem, weight in weighted_memories:
          mem_vector = content_to_vector(mem.content)
          if context_vector is None:
              context_vector = mem_vector * weight
          else:
              context_vector += mem_vector * weight
          total_weight += weight
  
      if total_weight > 0:
          return context_vector / total_weight # 加权平均
      return None
  
  # 示例使用
  # 假设 weighted_ctx 是我们之前获取的加权记忆列表
  context_vec = aggregate_context_vector(weighted_ctx)
  if context_vec is not None:
      print(f"nAggregated Context Vector: {context_vec}")
  else:
      print("nNo context to aggregate.")

• 注意力机制（Attention Mechanisms）：在深度学习中，注意力机制可以看作是一种更高级的加权聚合方式。模型会根据当前查询（query）动态地学习如何为序列中的每个元素（key-value对，即我们的记忆）分配注意力权重。时间信息可以通过位置编码（positional encoding）融入到记忆的表示中，从而让模型在学习注意力权重时自然地考虑时间因素。

五、 案例分析：三年前 vs. 三秒前的记忆权重分配

现在，让我们回到最初的问题：如何为三年前的记忆与三秒前的记忆分配不同的逻辑权重？

1. 简单时间衰减模型下的表现

让我们使用指数衰减模型来具体计算：w(Δt) = e^(-λ * Δt)

• 记忆A：三秒前的事件 (Δt = 3秒)
• 记忆B：三年前的事件 (Δt = 3年 = 3 365 24 60 60 = 94,608,000 秒)

假设我们选择一个衰减常数 λ = 0.001 （这意味着大约在1000秒后，权重衰减到初始值的36.8%）。

• 记忆A (3秒前) 的权重：
  w(3) = e^(-0.001 * 3) = e^(-0.003) ≈ 0.997
  这表明三秒前的记忆几乎没有衰减，其相关性非常高。

• 记忆B (三年前) 的权重：
  w(94,608,000) = e^(-0.001 * 94,608,000) = e^(-94,608) ≈ 0 (一个极小的数值，计算机通常会将其表示为0)
  这表明三年前的记忆在没有其他因素影响下，其时间相关性几乎为零。

结论： 简单的指数衰减模型能够非常有效地区分三秒前和三年前记忆的权重。对于近期记忆，权重接近1；对于遥远记忆，权重接近0。这符合我们的直觉：最近的事件对于当前决策通常更重要。

2. 超越简单衰减：语义与情境的重要性

然而，仅凭时间衰减并不总是足够的。考虑以下场景：

• 场景一：遗忘曲线的特例

  • 三秒前的记忆：“用户打了个喷嚏”。
  • 三年前的记忆：“用户在注册时设置了生日是1月1日”。
  • 在当前用户需要填写生日信息的场景下，三年前的生日记忆可能比三秒前的喷嚏记忆更有用。

• 场景二：上下文激活

  • 一个推荐系统，用户三年前购买了一套《指环王》小说。
  • 现在用户正在浏览奇幻电影。
  • 此时，三年前的“购买《指环王》”记忆，其相关性应该被激活或提升，而不是简单地衰减为零。

这引出了一个更深层次的问题：逻辑权重不仅仅是时间的函数，它还是记忆内容与当前任务情境的函数。

引入多维加权因子：

为了处理这种复杂性，我们可以将逻辑权重视为多个因子乘积或函数的组合：

Total_Weight = Temporal_Weight * Semantic_Weight * Contextual_Weight * Type_Weight

• Temporal_Weight (时间权重)：由上述衰减函数计算得到。
• Semantic_Weight (语义权重)：衡量记忆内容与当前任务或查询的语义相似度。这通常通过计算记忆向量和查询向量之间的余弦相似度、点积等方式获得。
• Contextual_Weight (情境权重)：如果记忆被当前情境明确引用或激活，可以人为地提升其权重。例如，在对话系统中，如果用户说“还记得我上次说的那个电影吗？”，那么最近一次提到的电影记忆权重应被提升。
• Type_Weight (类型权重)：不同类型的记忆可以有不同的默认衰减率或重要性因子。例如，“用户偏好”记忆的衰减速度应远慢于“用户临时浏览行为”的记忆。

示例：结合语义相似度的加权

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 假设我们有简单的向量表示
def get_memory_embedding(memory_content: Any) -> np.ndarray:
    """模拟获取记忆内容的嵌入向量"""
    if "喷嚏" in str(memory_content): return np.array([0.1, 0.2, 0.0, 0.0])
    if "生日" in str(memory_content): return np.array([0.0, 0.0, 0.8, 0.9])
    if "指环王" in str(memory_content): return np.array([0.7, 0.6, 0.1, 0.0])
    if "奇幻电影" in str(memory_content): return np.array([0.65, 0.55, 0.05, 0.0])
    return np.array([0.0, 0.0, 0.0, 0.0])

def calculate_semantic_weight(memory_content: Any, query_embedding: np.ndarray) -> float:
    """计算记忆内容与查询的语义相似度（余弦相似度）"""
    mem_embedding = get_memory_embedding(memory_content)
    if np.linalg.norm(mem_embedding) == 0 or np.linalg.norm(query_embedding) == 0:
        return 0.0
    return cosine_similarity(mem_embedding.reshape(1, -1), query_embedding.reshape(1, -1))[0][0]

class AdvancedMemoryStore(MemoryStore):
    def get_context_with_semantic_weight(self, current_time: Optional[datetime.datetime] = None,
                                         decay_function: str = "exponential", query_embedding: Optional[np.ndarray] = None,
                                         semantic_weight_factor: float = 0.5, **decay_params) -> List[Tuple[SystemMemory, float]]:
        """
        检索记忆，并结合时间衰减和语义相似度计算最终权重。
        Args:
            query_embedding: 当前任务或查询的语义嵌入。
            semantic_weight_factor: 语义权重在总权重中的影响力因子 (0到1之间)。
        """
        if current_time is None:
            current_time = datetime.datetime.now(datetime.timezone.utc)

        weighted_memories = []
        for mem in self.memories:
            temporal_weight = calculate_weight(mem.timestamp, current_time, decay_function, **decay_params)

            semantic_weight = 1.0 # 默认语义权重为1
            if query_embedding is not None:
                semantic_weight = calculate_semantic_weight(mem.content, query_embedding)
                # 确保语义权重在0到1之间，并可能进行归一化或缩放
                semantic_weight = max(0.0, min(1.0, semantic_weight))

            # 结合时间权重和语义权重
            # 可以是乘法，也可以是加权平均等
            # Total_Weight = (1-factor)*Temporal_Weight + factor*Semantic_Weight
            # 这里我们使用一个乘法和加权平均的组合，确保在某个方向上权重为0时，总权重也为0
            total_weight = temporal_weight * ((1 - semantic_weight_factor) + semantic_weight_factor * semantic_weight)

            if total_weight > 0.0001:
                weighted_memories.append((mem, total_weight))

        weighted_memories.sort(key=lambda x: x[1], reverse=True)
        return weighted_memories

# 重新初始化存储并添加记忆
adv_memory_store = AdvancedMemoryStore()
now = datetime.datetime.now(datetime.timezone.utc)

# 记忆A: 三秒前的喷嚏
mem_sneeze = SystemMemory("用户打了个喷嚏", timestamp=now - datetime.timedelta(seconds=3))
adv_memory_store.memories.append(mem_sneeze)

# 记忆B: 三年前的生日
mem_birthday = SystemMemory("用户在注册时设置了生日是1月1日", timestamp=now - datetime.timedelta(days=365*3))
adv_memory_store.memories.append(mem_birthday)

# 记忆C: 三年前的指环王购买
mem_lotr_purchase = SystemMemory("用户购买了《指环王》小说", timestamp=now - datetime.timedelta(days=365*3))
adv_memory_store.memories.append(mem_lotr_purchase)

# 记忆D: 10分钟前的无关记忆
mem_irrelevant = SystemMemory("用户点击了新闻链接", timestamp=now - datetime.timedelta(minutes=10))
adv_memory_store.memories.append(mem_irrelevant)

print("n--- 场景1: 查询 '生日' 相关信息 ---")
query_birthday_embedding = get_memory_embedding("用户想知道生日")
weighted_ctx_birthday = adv_memory_store.get_context_with_semantic_weight(
    decay_function="exponential", lambda_val=0.000001, # 设定极慢的衰减率，以凸显语义作用
    query_embedding=query_birthday_embedding, semantic_weight_factor=0.8
)
for mem, weight in weighted_ctx_birthday:
    print(f"[{weight:.4f}] {mem}")

print("n--- 场景2: 查询 '奇幻电影' 相关信息 ---")
query_fantasy_embedding = get_memory_embedding("推荐奇幻电影")
weighted_ctx_fantasy = adv_memory_store.get_context_with_semantic_weight(
    decay_function="exponential", lambda_val=0.000001, # 设定极慢的衰减率，以凸显语义作用
    query_embedding=query_fantasy_embedding, semantic_weight_factor=0.8
)
for mem, weight in weighted_ctx_fantasy:
    print(f"[{weight:.4f}] {mem}")

输出示例 (实际数值会因当前时间而略有不同):

--- 场景1: 查询 '生日' 相关信息 ---
[0.7200] Memory(content='用户在注册时设置了生日是1月1日', timestamp=2020-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户打了个喷嚏', timestamp=2023-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户购买了《指环王》小说', timestamp=2020-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户点击了新闻链接', timestamp=2023-10-27T09:50:00.000000+00:00, type='general')

--- 场景2: 查询 '奇幻电影' 相关信息 ---
[0.7200] Memory(content='用户购买了《指环王》小说', timestamp=2020-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户打了个喷嚏', timestamp=2023-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户在注册时设置了生日是1月1日', timestamp=2020-10-27T10:00:00.000000+00:00, type='general')
[0.2000] Memory(content='用户点击了新闻链接', timestamp=2023-10-27T09:50:00.000000+00:00, type='general')

（注：由于lambda_val设置得极小，时间衰减效应被极大弱化，使得语义相似度成为主要权重贡献者。在实际应用中，lambda_val会根据业务需求进行精细调整。）

从上述结果可以看出，即使三年前的记忆在时间上非常遥远，但如果其语义与当前查询高度相关，它依然可以获得较高的总权重，甚至超过近期但语义不相关的记忆。这正是我们追求的智能行为。

六、 高级概念与混合方法

在现代AI系统中，尤其是基于深度学习的系统中，时间上下文加权往往以更隐式和复杂的方式实现。

1. 循环神经网络（RNNs, LSTMs, GRUs）

RNN及其变体（LSTM、GRU）被设计用于处理序列数据。它们通过隐藏状态在时间步之间传递信息，从而捕获时间依赖性。LSTM和GRU通过门控机制（输入门、遗忘门、输出门）来选择性地更新和遗忘信息，这使得它们能够有效地处理长期依赖问题。

虽然它们没有显式地计算每个记忆的“权重”，但其内部机制等同于为不同时间步的信息分配了动态的、上下文敏感的逻辑权重。一个通过遗忘门成功保留下来的旧信息，其对当前状态的影响就相当于被赋予了较高的权重。

2. 注意力机制与Transformer

Transformer架构及其核心的自注意力（Self-Attention）机制彻底改变了序列建模。它通过计算查询（Query）、键（Key）和值（Value）之间的相似度来为序列中的每个元素动态分配注意力权重。

在Transformer中，时间信息通常通过位置编码（Positional Encoding）注入到输入嵌入中。这意味着模型在计算注意力权重时，不仅考虑了语义相似度，也考虑了相对或绝对的时间位置。这使得模型能够：

• 关注远距离依赖：不再受限于RNN的序列处理瓶颈。
• 学习复杂的时序模式：通过训练数据学习哪些时间点的哪些信息更重要。

可以想象，三秒前的记忆在位置编码的帮助下，往往会与当前查询产生更高的注意力分数；而三年前的记忆，如果其语义与当前查询高度相关，且模型学到了这种长期关联，也能获得高注意力分数。

# 概念性的Transformer中时间偏置的注意力计算
import torch
import torch.nn as nn
import torch.nn.functional as F

class ConceptualTemporalAttention(nn.Module):
    def __init__(self, d_model):
        super().__init__()
        self.query_proj = nn.Linear(d_model, d_model)
        self.key_proj = nn.Linear(d_model, d_model)
        self.value_proj = nn.Linear(d_model, d_model)

    def forward(self, query_vec, memory_embeddings_with_pos, memory_timestamps, current_timestamp):
        """
        Args:
            query_vec: (1, d_model) 当前查询的嵌入向量
            memory_embeddings_with_pos: (num_memories, d_model) 记忆嵌入（已包含位置编码）
            memory_timestamps: (num_memories,) 记忆时间戳 (秒)
            current_timestamp: 当前时间戳 (秒)
        Returns:
            context_vector: (1, d_model) 加权后的上下文向量
            attention_weights: (num_memories,) 每个记忆的注意力权重
        """
        Q = self.query_proj(query_vec) # (1, d_model)
        K = self.key_proj(memory_embeddings_with_pos) # (num_memories, d_model)
        V = self.value_proj(memory_embeddings_with_pos) # (num_memories, d_model)

        # 1. 计算点积注意力得分 (语义相似度)
        scores = torch.matmul(Q, K.transpose(0, 1)) / math.sqrt(K.size(-1)) # (1, num_memories)

        # 2. 引入时间偏置 (作为一个额外的分数项或调整项)
        # 这里我们使用一个简化的指数衰减作为时间偏置，并将其加到原始得分上
        time_deltas = current_timestamp - memory_timestamps # (num_memories,)
        # 避免负时间差
        time_deltas = torch.max(time_deltas, torch.tensor(0.0, device=time_deltas.device))

        lambda_val = 0.000001 # 极慢的衰减，让语义主导，但时间仍有微弱影响
        # 确保时间偏置在合适范围内，例如0到1
        temporal_bias = torch.exp(-lambda_val * time_deltas) # (num_memories,)

        # 将时间偏置加到原始分数上。注意：这里是简化的，实际Transformer中更多是学习到的相对位置编码
        # 或者在K/Q的计算中直接融入时间信息
        # 另一种方法是： scores = scores + temporal_bias_matrix (如果temporal_bias也是一个矩阵)
        # 这里为了演示，我们直接将标量偏置加到每个记忆的得分上
        # 实际操作中，temporal_bias_matrix可能是一个根据相对时间差预计算或学习得到的矩阵
        scores = scores + temporal_bias.unsqueeze(0) # (1, num_memories) + (1, num_memories)

        attention_weights = F.softmax(scores, dim=-1) # (1, num_memories)

        context_vector = torch.matmul(attention_weights, V) # (1, d_model)

        return context_vector, attention_weights.squeeze(0)

# 模拟数据
d_model = 64
num_memories = 5

query_embedding = torch.randn(1, d_model)

# 记忆嵌入 (假设已包含位置编码)
memory_embeddings = torch.randn(num_memories, d_model)

# 记忆时间戳 (秒)
mem_timestamps = torch.tensor([
    (now - datetime.timedelta(seconds=3)).timestamp(),
    (now - datetime.timedelta(days=365*3)).timestamp(), # 三年前
    (now - datetime.timedelta(minutes=10)).timestamp(),
    (now - datetime.timedelta(hours=2)).timestamp(),
    (now - datetime.timedelta(seconds=1)).timestamp(),
])

current_timestamp = torch.tensor(now.timestamp())

attn_model = ConceptualTemporalAttention(d_model)
context_vec, att_weights = attn_model(query_embedding, memory_embeddings, mem_timestamps, current_timestamp)

print("n--- Conceptual Temporal Attention ---")
print(f"Context Vector shape: {context_vec.shape}")
print(f"Attention Weights: {att_weights}")

这个概念性代码展示了如何将一个简单的时间偏置项集成到注意力计算中。在真实的Transformer中，这种时间偏置通常通过更复杂的相对位置编码或可学习的偏置矩阵来实现，以捕获更丰富的时序关系。

3. 强化学习与经验回放（Prioritized Experience Replay）

在强化学习中，经验回放（Experience Replay）池存储了智能体过去的经验（状态、动作、奖励、新状态）。简单地随机采样会忽视经验的重要性。优先经验回放（Prioritized Experience Replay, PER）则根据经验的“重要性”（例如，TD误差的大小）来非均匀地采样经验。虽然PER主要关注的是学习的重要性而非时间本身，但最近的经验通常有更高的TD误差，从而间接赋予了更高的采样概率，这与时间加权有异曲同工之妙。同时，PER也可以结合时间衰减，例如，让旧的、重要性低的经验逐渐被移除。

七、 设计健壮的时间加权系统

设计一个有效的时间加权系统需要综合考虑多个方面：

1. 明确记忆类型：区分短期交互、长期偏好、系统事件等，对不同类型记忆应用不同的衰减策略。
2. 定义相关性目标：系统希望在什么情况下，旧记忆能够被激活？新记忆又在什么情况下可以被忽略？
3. 参数调优：衰减函数的参数（λ, α, p 等）需要通过实验、AB测试或机器学习方法进行优化。这些参数往往是领域知识和经验的体现。
4. 混合策略：纯粹的时间衰减往往不够。结合语义相似度、情境激活、用户反馈等多种因素，构建一个多维度的加权模型。
5. 实时性与效率：加权计算不应成为系统瓶颈。对于大规模记忆库，需要考虑索引、近似计算和缓存策略。例如，只对最近的N个记忆进行精确加权，对更老的记忆进行稀疏采样或预聚合。
6. 可解释性：在某些应用中，理解为什么某个记忆被赋予高权重很重要。设计时应考虑如何追踪权重的来源。

八、 展望与挑战

时间上下文加权是构建智能、自适应系统不可或缺的一环。它使得系统能够像人类一样，在海量信息中聚焦于当前最相关的部分，从而做出更明智的决策。从简单的指数衰减到复杂的注意力机制，这一领域不断演进，以更好地模拟和利用时间信息。

然而，挑战依然存在。如何动态地调整衰减参数以适应不断变化的用户行为和环境？如何优雅地融合时间、语义和情境信息，尤其是在缺乏大量标注数据的情况下？这些都是未来研究的重要方向。但无论如何，理解并有效利用时间这一基本维度，将始终是智能系统设计中的一个核心能力。

本次讲座就到这里。感谢大家的聆听！