Python为何成为TVA的神经与感官系统（3）

重磅预告：本专栏将独家连载系列丛书《AI智能体视觉技术与应用》部分精华内容，该书是世界首套系统阐述“因式智能体”视觉理论与实践的专著，特邀美国 TypeOne 公司首席科学家、斯坦福大学博士 Bohan 担任技术顾问。Bohan先生师从美国三院院士、“AI教母”李飞飞教授，学术引用量在近四年内突破万次，是全球AI与机器人视觉领域的标杆性人物（www.type-one.com）。全书严格遵循“基础—原理—实操—进阶—赋能—未来”的六步进阶逻辑，致力于引入“类人智眼”新范式，系统破解从数字世界到物理世界“最后一公里”的世界级难题。该书精彩内容将优先在本专栏陆续发布，其纸质专著亦将正式出版。敬请关注！

前沿技术背景介绍：AI智能体视觉（TVA，Transformer-based Vision Agent）是依托Transformer架构与“因式智能体”理论所构建的颠覆性工业视觉技术，属于“物理AI” 领域的一种全新技术形态，实现了从“虚拟世界”到“真实世界”的历史性跨越。它区别于传统计算机视觉和常规AI视觉技术，代表了工业智能化转型与视觉检测模式的根本性重构（www.tianyance.cn)。 在实质内涵上，TVA是一种复合概念，是集深度强化学习（DRL）、卷积神经网络（CNN）、因式分解算法（FRA）于一体的系统工程框架，构建了能够“感知-推理-决策-行动-反馈”的迭代运作闭环，完成从“看见”到“看懂”的范式突破，不仅被业界誉为“AI视觉检测专家”，而且也被理解为“具身视觉智能体“，是智能机器人视觉与灵巧运动控制的关键技术支撑。

版权声明：本文系作者原创首发于 CSDN 的技术类文章，受《中华人民共和国著作权法》保护，转载或商用敬请注明出处。

——TVA的长期记忆与学习机制Python实现

感知编码是认知过程的第一步，将感官输入转化为工作记忆可以处理的形式。在TVA中，感知编码涉及特征提取、对象识别和类别划分。我们将用Python实现一个感知编码系统，模拟从视觉输入到工作记忆存储的过程。

（接上篇）

3.1 长期记忆系统架构

长期记忆是认知系统的知识库，存储从工作记忆中巩固的知识。在TVA中，长期记忆包括语义记忆、情景记忆和程序记忆。以下是Python实现的完整长期记忆系统。

import hashlib
import json
import pickle
import numpy as np
from datetime import datetime, timedelta
from collections import defaultdict, deque
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, List, Any, Optional, Tuple
import heapq
import networkx as nx
from dataclasses import dataclass, field
from enum import Enum
import zlib
import time

class MemoryType(Enum):
"""记忆类型枚举"""
SEMANTIC = "semantic" # 语义记忆
EPISODIC = "episodic" # 情景记忆
PROCEDURAL = "procedural" # 程序记忆
DECLARATIVE = "declarative" # 陈述性记忆
NON_DECLARATIVE = "non_declarative" # 非陈述性记忆

class ConsolidationType(Enum):
"""巩固类型"""
SYNAPTIC = "synaptic" # 突触巩固
SYSTEMS = "systems" # 系统巩固
REACTIVATION = "reactivation" # 重新激活

@dataclass
class MemoryTrace:
"""记忆痕迹"""
id: str
content: Any
memory_type: MemoryType
strength: float = 0.1
access_count: int = 0
last_accessed: datetime = field(default_factory=datetime.now)
created_at: datetime = field(default_factory=datetime.now)
associations: List[str] = field(default_factory=list)
context: Dict[str, Any] = field(default_factory=dict)
emotional_valence: float = 0.0
importance: float = 0.5
consolidation_level: float = 0.0
forgetting_rate: float = 0.01
retrieval_threshold: float = 0.3

def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
'id': self.id,
'memory_type': self.memory_type.value,
'strength': self.strength,
'access_count': self.access_count,
'last_accessed': self.last_accessed.isoformat(),
'created_at': self.created_at.isoformat(),
'associations': self.associations,
'context': self.context,
'emotional_valence': self.emotional_valence,
'importance': self.importance,
'consolidation_level': self.consolidation_level,
'forgetting_rate': self.forgetting_rate,
'retrieval_threshold': self.retrieval_threshold
}

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'MemoryTrace':
"""从字典创建"""
memory = cls(
id=data['id'],
content=None, # 内容需要单独加载
memory_type=MemoryType(data['memory_type']),
strength=data['strength'],
access_count=data['access_count'],
last_accessed=datetime.fromisoformat(data['last_accessed']),
created_at=datetime.fromisoformat(data['created_at']),
associations=data['associations'],
context=data['context'],
emotional_valence=data['emotional_valence'],
importance=data['importance'],
consolidation_level=data['consolidation_level'],
forgetting_rate=data['forgetting_rate'],
retrieval_threshold=data['retrieval_threshold']
)
return memory

def update_strength(self, delta: float, consolidation: bool = False):
"""更新记忆强度"""
old_strength = self.strength
self.strength = max(0.0, min(1.0, self.strength + delta))

# 如果巩固，降低遗忘率
if consolidation:
self.forgetting_rate *= 0.9
self.consolidation_level = min(1.0, self.consolidation_level + 0.1)

return self.strength – old_strength

def decay(self, time_elapsed_hours: float):
"""记忆衰减"""
decay_amount = self.forgetting_rate * time_elapsed_hours * (1 – self.consolidation_level)
self.strength = max(0.0, self.strength – decay_amount)
return decay_amount

def can_be_retrieved(self) -> bool:
"""检查记忆是否可检索"""
return self.strength >= self.retrieval_threshold

def get_retrieval_probability(self) -> float:
"""获取检索概率"""
if not self.can_be_retrieved():
return 0.0

base_prob = self.strength
# 考虑最近访问的影响
recency = np.exp(-(datetime.now() – self.last_accessed).total_seconds() / 3600 / 24) # 天为单位
frequency = min(1.0, self.access_count / 100) # 频率因子

retrieval_prob = (
base_prob * 0.6 +
recency * 0.2 +
frequency * 0.2
)

return min(1.0, retrieval_prob)

class LongTermMemory:
"""长期记忆系统"""

def __init__(self, config: Dict[str, Any]):
# 记忆存储
self.memories: Dict[str, MemoryTrace] = {}
self.content_store: Dict[str, Any] = {} # 分开存储内容

# 索引结构
self.semantic_network = nx.Graph() # 语义网络
self.episodic_timeline = [] # 情景时间线
self.context_index: Dict[str, List[str]] = defaultdict(list) # 上下文索引
self.content_index: Dict[str, List[str]] = defaultdict(list) # 内容索引

# 巩固系统
self.consolidation_system = ConsolidationSystem(config.get('consolidation', {}))

# 检索系统
self.retrieval_system = RetrievalSystem(config.get('retrieval', {}))

# 遗忘机制
self.forgetting_system = ForgettingSystem(config.get('forgetting', {}))

# 配置参数
self.capacity = config.get('capacity', float('inf')) # 记忆容量
self.auto_consolidate = config.get('auto_consolidate', True)
self.auto_forget = config.get('auto_forget', True)

# 统计信息
self.stats = {
'total_memories': 0,
'semantic_count': 0,
'episodic_count': 0,
'procedural_count': 0,
'consolidation_count': 0,
'retrieval_count': 0,
'forgetting_count': 0
}

# 启动维护线程
self._start_maintenance_thread()

def store(self, content: Any, memory_type: MemoryType,
context: Dict[str, Any] = None, importance: float = 0.5) -> str:
"""
存储记忆

参数:
content: 记忆内容
memory_type: 记忆类型
context: 上下文信息
importance: 重要性

返回:
记忆ID
"""
# 生成记忆ID
content_hash = hashlib.md5(pickle.dumps(content)).hexdigest()
memory_id = f"{memory_type.value}_{content_hash}_{int(time.time())}"

# 创建记忆痕迹
memory_trace = MemoryTrace(
id=memory_id,
content=content, # 先存储引用
memory_type=memory_type,
strength=0.1, # 初始强度
context=context or {},
importance=importance,
emotional_valence=self._compute_emotional_valence(content, context),
forgetting_rate=self._compute_forgetting_rate(memory_type, importance)
)

# 存储记忆
self.memories[memory_id] = memory_trace
self.content_store[memory_id] = content

# 更新索引
self._update_indexes(memory_id, memory_trace)

# 更新统计
self._update_stats(memory_type, 'store')

# 自动巩固
if self.auto_consolidate and importance > 0.7:
self.consolidate(memory_id, priority='high')

return memory_id

def _update_indexes(self, memory_id: str, memory: MemoryTrace):
"""更新索引"""
# 更新语义网络
if memory.memory_type == MemoryType.SEMANTIC:
self.semantic_network.add_node(memory_id,
content_type=type(memory.content).__name__,
strength=memory.strength)

# 建立关联
for other_id, other_memory in self.memories.items():
if (other_id != memory_id and
other_memory.memory_type == MemoryType.SEMANTIC and
self._compute_semantic_similarity(memory.content, other_memory.content) > 0.7):

similarity = self._compute_semantic_similarity(memory.content, other_memory.content)
self.semantic_network.add_edge(memory_id, other_id,
weight=similarity)
memory.associations.append(other_id)
other_memory.associations.append(memory_id)

# 更新情景时间线
elif memory.memory_type == MemoryType.EPISODIC:
self.episodic_timeline.append(memory_id)
# 按时间排序
self.episodic_timeline.sort(
key=lambda x: self.memories[x].created_at
)

# 更新上下文索引
for key, value in memory.context.items():
context_key = f"{key}:{value}"
self.context_index[context_key].append(memory_id)

# 更新内容索引
if isinstance(memory.content, str):
for word in memory.content.split():
self.content_index[word.lower()].append(memory_id)

def retrieve(self, query: Any, context: Dict[str, Any] = None,
memory_type: Optional[MemoryType] = None,
max_results: int = 10) -> List[Dict[str, Any]]:
"""
检索记忆

参数:
query: 查询内容
context: 查询上下文
memory_type: 记忆类型过滤
max_results: 最大结果数

返回:
检索结果列表
"""
retrieval_start = time.time()

# 1. 候选记忆生成
candidates = self._generate_candidates(query, context, memory_type)

# 2. 计算相关性得分
scored_candidates = []
for memory_id in candidates:
memory = self.memories[memory_id]

# 基本相关性
relevance = self._compute_relevance(memory, query, context)

# 记忆强度
strength_factor = memory.strength

# 近因效应
recency = np.exp(-(datetime.now() – memory.last_accessed).total_seconds() / 3600 / 24)

# 频率效应
frequency = min(1.0, memory.access_count / 100)

# 总得分
score = (
relevance * 0.4 +
strength_factor * 0.3 +
recency * 0.2 +
frequency * 0.1
)

# 重要性加成
score *= (1 + memory.importance * 0.5)

scored_candidates.append((score, memory_id))

# 3. 排序和过滤
scored_candidates.sort(reverse=True)
top_candidates = scored_candidates[:max_results]

# 4. 检索结果
results = []
for score, memory_id in top_candidates:
memory = self.memories[memory_id]
content = self.content_store[memory_id]

# 更新访问统计
memory.access_count += 1
memory.last_accessed = datetime.now()

# 重新激活巩固
if self.auto_consolidate:
self.consolidation_system.reactivate(memory_id)

results.append({
'memory_id': memory_id,
'content': content,
'score': score,
'strength': memory.strength,
'access_count': memory.access_count,
'recency': (datetime.now() – memory.last_accessed).total_seconds() / 3600,
'context': memory.context
})

retrieval_time = time.time() – retrieval_start

# 更新统计
self.stats['retrieval_count'] += 1

return {
'results': results,
'total_candidates': len(candidates),
'retrieval_time': retrieval_time
}

def _generate_candidates(self, query: Any, context: Dict[str, Any],
memory_type: Optional[MemoryType]) -> List[str]:
"""生成候选记忆"""
candidates = set()

# 基于内容匹配
if isinstance(query, str):
for word in query.lower().split():
if word in self.content_index:
candidates.update(self.content_index[word][:100]) # 限制数量

# 基于上下文匹配
if context:
for key, value in context.items():
context_key = f"{key}:{value}"
if context_key in self.context_index:
candidates.update(self.context_index[context_key])

# 基于记忆类型过滤
if memory_type is not None:
candidates = {mid for mid in candidates
if self.memories[mid].memory_type == memory_type}

# 确保记忆可检索
candidates = {mid for mid in candidates
if self.memories[mid].can_be_retrieved()}

return list(candidates)

def consolidate(self, memory_id: str, priority: str = 'normal',
consolidation_type: ConsolidationType = ConsolidationType.SYNAPTIC) -> Dict[str, Any]:
"""
巩固记忆

参数:
memory_id: 记忆ID
priority: 优先级
consolidation_type: 巩固类型

返回:
巩固结果
"""
if memory_id not in self.memories:
return {'success': False, 'error': 'Memory not found'}

memory = self.memories[memory_id]

# 执行巩固
consolidation_result = self.consolidation_system.consolidate(
memory, consolidation_type, priority
)

if consolidation_result['success']:
# 更新记忆
memory.strength += consolidation_result['strength_increase']
memory.consolidation_level = consolidation_result['new_consolidation_level']
memory.forgetting_rate *= consolidation_result['forgetting_reduction']

# 更新关联记忆
for assoc_id in memory.associations:
if assoc_id in self.memories:
assoc_memory = self.memories[assoc_id]
# 关联记忆也得到一定巩固
assoc_memory.strength += consolidation_result['strength_increase'] * 0.1

# 更新统计
self.stats['consolidation_count'] += 1

return consolidation_result

def forget(self, memory_id: str, forgetting_type: str = 'natural') -> Dict[str, Any]:
"""
遗忘记忆

参数:
memory_id: 记忆ID
forgetting_type: 遗忘类型

返回:
遗忘结果
"""
if memory_id not in self.memories:
return {'success': False, 'error': 'Memory not found'}

memory = self.memories[memory_id]

# 检查是否可以遗忘
if memory.importance > 0.8 and memory.strength > 0.5:
return {'success': False, 'error': 'Important memory cannot be forgotten'}

# 执行遗忘
forgetting_result = self.forgetting_system.forget(memory, forgetting_type)

if forgetting_result['success']:
# 从记忆中移除
del self.memories[memory_id]
if memory_id in self.content_store:
del self.content_store[memory_id]

# 更新索引
self._remove_from_indexes(memory_id)

# 更新统计
self.stats['forgetting_count'] += 1

return forgetting_result

def _remove_from_indexes(self, memory_id: str):
"""从索引中移除记忆"""
# 从语义网络移除
if memory_id in self.semantic_network:
self.semantic_network.remove_node(memory_id)

# 从情景时间线移除
if memory_id in self.episodic_timeline:
self.episodic_timeline.remove(memory_id)

# 从上下文索引移除
for key in list(self.context_index.keys()):
if memory_id in self.context_index[key]:
self.context_index[key].remove(memory_id)
if not self.context_index[key]: # 如果为空，删除键
del self.context_index[key]

# 从内容索引移除
for word in list(self.content_index.keys()):
if memory_id in self.content_index[word]:
self.content_index[word].remove(memory_id)
if not self.content_index[word]:
del self.content_index[word]

def replay_episodes(self, recent_hours: float = 24.0,
num_episodes: int = 5) -> List[Dict[str, Any]]:
"""
重放情景记忆（睡眠中的记忆重放）

参数:
recent_hours: 最近多少小时内的记忆
num_episodes: 重放的情景数量

返回:
重放结果
"""
# 获取最近的情景记忆
recent_cutoff = datetime.now() – timedelta(hours=recent_hours)
recent_episodes = []

for memory_id in self.episodic_timeline:
memory = self.memories[memory_id]
if memory.created_at >= recent_cutoff:
recent_episodes.append(memory_id)

# 选择要重放的情景
episodes_to_replay = recent_episodes[-num_episodes:] # 最近的情景

replay_results = []
for episode_id in episodes_to_replay:
replay_result = self._replay_episode(episode_id)
replay_results.append(replay_result)

# 巩固重放的记忆
self.consolidate(episode_id, priority='high',
consolidation_type=ConsolidationType.REACTIVATION)

return replay_results

def _replay_episode(self, memory_id: str) -> Dict[str, Any]:
"""重放单个情景"""
memory = self.memories[memory_id]
content = self.content_store[memory_id]

# 模拟记忆重放
replay_intensity = np.random.uniform(0.5, 1.0)
replay_duration = np.random.exponential(0.5) # 平均0.5秒

# 记忆增强
strength_increase = replay_intensity * 0.1
memory.strength = min(1.0, memory.strength + strength_increase)

# 降低遗忘率
memory.forgetting_rate *= 0.95

return {
'memory_id': memory_id,
'replay_intensity': replay_intensity,
'replay_duration': replay_duration,
'strength_increase': strength_increase,
'new_strength': memory.strength,
'timestamp': datetime.now().isoformat()
}

def semantic_search(self, query: str, max_depth: int = 3,
similarity_threshold: float = 0.6) -> List[Dict[str, Any]]:
"""
语义搜索

参数:
query: 搜索查询
max_depth: 语义网络搜索深度
similarity_threshold: 相似度阈值

返回:
搜索结果
"""
# 找到查询的起点
query_words = set(query.lower().split())
start_nodes = []

for word in query_words:
if word in self.content_index:
for memory_id in self.content_index[word]:
if memory_id in self.semantic_network:
start_nodes.append(memory_id)

if not start_nodes:
return []

# 在语义网络中搜索
results = []
visited = set()

for start_node in start_nodes[:10]: # 限制起点数量
# 广度优先搜索
queue = deque([(start_node, 0)]) # (节点, 深度)

while queue:
current_node, depth = queue.popleft()

if current_node in visited or depth > max_depth:
continue

visited.add(current_node)

# 获取记忆
memory = self.memories.get(current_node)
if not memory:
continue

# 计算相似度
similarity = self._compute_semantic_similarity(query, memory.content)

if similarity >= similarity_threshold:
results.append({
'memory_id': current_node,
'content': self.content_store[current_node],
'similarity': similarity,
'depth': depth,
'strength': memory.strength,
'memory_type': memory.memory_type.value
})

# 添加邻居节点
if current_node in self.semantic_network:
neighbors = list(self.semantic_network.neighbors(current_node))
for neighbor in neighbors:
if neighbor not in visited:
queue.append((neighbor, depth + 1))

# 按相似度排序
results.sort(key=lambda x: x['similarity'], reverse=True)

return results[:20] # 返回前20个结果

def _compute_semantic_similarity(self, item1: Any, item2: Any) -> float:
"""计算语义相似度"""
# 简单实现：基于字符串相似度
if isinstance(item1, str) and isinstance(item2, str):
# 使用Jaccard相似度
words1 = set(item1.lower().split())
words2 = set(item2.lower().split())

if not words1 or not words2:
return 0.0

intersection = len(words1.intersection(words2))
union = len(words1.union(words2))

return intersection / union

elif isinstance(item1, dict) and isinstance(item2, dict):
# 比较字典的键
keys1 = set(item1.keys())
keys2 = set(item2.keys())

if not keys1 or not keys2:
return 0.0

common_keys = keys1.intersection(keys2)
all_keys = keys1.union(keys2)

if not all_keys:
return 0.0

similarity = len(common_keys) / len(all_keys)

# 比较共同键的值
value_similarity_sum = 0
for key in common_keys:
value_sim = self._compute_semantic_similarity(item1[key], item2[key])
value_similarity_sum += value_sim

if common_keys:
value_similarity_avg = value_similarity_sum / len(common_keys)
similarity = (similarity + value_similarity_avg) / 2

return similarity

else:
# 其他类型，尝试转换为字符串
try:
str1 = str(item1)
str2 = str(item2)
return self._compute_semantic_similarity(str1, str2)
except:
return 0.0

def _compute_emotional_valence(self, content: Any, context: Dict[str, Any]) -> float:
"""计算情绪效价"""
# 简化实现：基于关键词
positive_keywords = ['happy', 'good', 'great', 'excellent', 'love', 'success']
negative_keywords = ['bad', 'sad', 'angry', 'hate', 'fail', 'wrong']

content_str = str(content).lower()
context_str = str(context).lower()

combined = f"{content_str} {context_str}"

positive_count = sum(1 for word in positive_keywords if word in combined)
negative_count = sum(1 for word in negative_keywords if word in combined)

total = positive_count + negative_count
if total == 0:
return 0.0

emotional_valence = (positive_count – negative_count) / total
return max(-1.0, min(1.0, emotional_valence))

def _compute_forgetting_rate(self, memory_type: MemoryType, importance: float) -> float:
"""计算遗忘率"""
# 基础遗忘率
base_rates = {
MemoryType.SEMANTIC: 0.001,
MemoryType.EPISODIC: 0.005,
MemoryType.PROCEDURAL: 0.0005,
MemoryType.DECLARATIVE: 0.003,
MemoryType.NON_DECLARATIVE: 0.002
}

base_rate = base_rates.get(memory_type, 0.005)

# 重要性调整：越重要，遗忘率越低
importance_factor = 1.0 – (importance * 0.8) # 重要性减少遗忘率

return base_rate * importance_factor

def _compute_relevance(self, memory: MemoryTrace, query: Any, context: Dict[str, Any]) -> float:
"""计算相关性"""
relevance = 0.0

# 1. 内容相似度
content_similarity = self._compute_semantic_similarity(query, memory.content)
relevance += content_similarity * 0.4

# 2. 上下文匹配度
if context and memory.context:
context_similarity = self._compute_semantic_similarity(context, memory.context)
relevance += context_similarity * 0.3

# 3. 情绪匹配
if 'emotional_valence' in (context or {}):
query_valence = context.get('emotional_valence', 0.0)
valence_similarity = 1.0 – abs(query_valence – memory.emotional_valence) / 2.0
relevance += valence_similarity * 0.2

# 4. 时间相关性
time_diff_hours = (datetime.now() – memory.created_at).total_seconds() / 3600
time_relevance = np.exp(-time_diff_hours / 24) # 天为单位
relevance += time_relevance * 0.1

return min(1.0, relevance)

def _start_maintenance_thread(self):
"""启动维护线程"""
def maintenance_worker():
while True:
try:
# 执行维护任务
self._perform_memory_maintenance()
time.sleep(3600) # 每小时维护一次
except Exception as e:
print(f"记忆维护错误: {e}")
time.sleep(300) # 错误后等待5分钟

maintenance_thread = threading.Thread(target=maintenance_worker, daemon=True)
maintenance_thread.start()

def _perform_memory_maintenance(self):
"""执行记忆维护"""
# 1. 自然遗忘
if self.auto_forget:
self._apply_natural_forgetting()

# 2. 记忆衰减
self._apply_memory_decay()

# 3. 自动巩固重要记忆
self._consolidate_important_memories()

# 4. 清理无效索引
self._cleanup_indexes()

# 5. 内存优化
self._optimize_memory_usage()

def _apply_natural_forgetting(self):
"""应用自然遗忘"""
forget_candidates = []
current_time = datetime.now()

for memory_id, memory in self.memories.items():
# 检查是否应该遗忘
hours_since_creation = (current_time – memory.created_at).total_seconds() / 3600
hours_since_access = (current_time – memory.last_accessed).total_seconds() / 3600

# 遗忘概率公式
forget_probability = (
memory.forgetting_rate *
(1 – memory.importance) *
(1 – memory.consolidation_level) *
np.log(1 + hours_since_access) *
(1 – np.exp(-hours_since_creation / 24))
)

if np.random.random() < forget_probability:
forget_candidates.append((forget_probability, memory_id))

# 遗忘最可能被遗忘的记忆
forget_candidates.sort(reverse=True)
to_forget = forget_candidates[:min(10, len(forget_candidates))] # 每次最多遗忘10个

for prob, memory_id in to_forget:
self.forget(memory_id, 'natural')

def _apply_memory_decay(self):
"""应用记忆衰减"""
current_time = datetime.now()

for memory_id, memory in self.memories.items():
hours_since_access = (current_time – memory.last_accessed).total_seconds() / 3600
memory.decay(hours_since_access)

def _consolidate_important_memories(self):
"""巩固重要记忆"""
important_memories = []

for memory_id, memory in self.memories.items():
if (memory.importance > 0.7 and
memory.consolidation_level < 0.8 and
memory.strength > 0.3):

# 计算巩固优先级
priority_score = (
memory.importance * 0.4 +
(1 – memory.consolidation_level) * 0.3 +
memory.strength * 0.3
)

important_memories.append((priority_score, memory_id))

# 按优先级排序
important_memories.sort(reverse=True)

# 巩固前5个最重要的记忆
for _, memory_id in important_memories[:5]:
self.consolidate(memory_id, priority='high')

def _cleanup_indexes(self):
"""清理无效索引"""
# 清理不存在的记忆引用
memory_ids = set(self.memories.keys())

# 清理上下文索引
for context_key in list(self.context_index.keys()):
valid_ids = [mid for mid in self.context_index[context_key] if mid in memory_ids]
if valid_ids:
self.context_index[context_key] = valid_ids
else:
del self.context_index[context_key]

# 清理内容索引
for word in list(self.content_index.keys()):
valid_ids = [mid for mid in self.content_index[word] if mid in memory_ids]
if valid_ids:
self.content_index[word] = valid_ids
else:
del self.content_index[word]

def _optimize_memory_usage(self):
"""优化内存使用"""
# 如果超过容量，删除最不重要的记忆
if len(self.memories) > self.capacity:
# 计算每个记忆的分数
memory_scores = []

for memory_id, memory in self.memories.items():
score = (
memory.importance * 0.4 +
memory.strength * 0.3 +
np.exp(-(datetime.now() – memory.last_accessed).total_seconds() / 3600 / 24) * 0.2 +
(memory.access_count / 100) * 0.1
)

memory_scores.append((score, memory_id))

# 按分数排序
memory_scores.sort()

# 删除分数最低的记忆
to_remove = memory_scores[:len(self.memories) – self.capacity]
for score, memory_id in to_remove:
self.forget(memory_id, 'capacity_limit')

3.2 神经网络记忆模型

使用PyTorch实现神经网络记忆模型，包括记忆编码、存储和检索：

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
import numpy as np
from typing import List, Tuple, Optional

class NeuralMemoryNetwork(nn.Module):
"""神经网络记忆模型"""

def __init__(self, input_dim: int, hidden_dim: int,
memory_size: int, num_heads: int = 8):
super().__init__()

# 编码器
self.encoder = nn.Sequential(
nn.Linear(input_dim, hidden_dim * 2),
nn.ReLU(),
nn.Linear(hidden_dim * 2, hidden_dim),
nn.LayerNorm(hidden_dim)
)

# 记忆矩阵
self.memory_keys = nn.Parameter(torch.randn(memory_size, hidden_dim) * 0.1)
self.memory_values = nn.Parameter(torch.randn(memory_size, hidden_dim) * 0.1)

# 注意机制
self.attention = nn.MultiheadAttention(
embed_dim=hidden_dim,
num_heads=num_heads,
dropout=0.1,
batch_first=True
)

# 解码器
self.decoder = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, input_dim)
)

# 控制门
self.input_gate = nn.Linear(hidden_dim, hidden_dim)
self.forget_gate = nn.Linear(hidden_dim, hidden_dim)
self.output_gate = nn.Linear(hidden_dim, hidden_dim)

# 初始化
self.hidden_dim = hidden_dim
self.memory_size = memory_size
self.reset_memory()

def reset_memory(self):
"""重置记忆"""
self.memory_strengths = torch.zeros(self.memory_size)
self.memory_access_counts = torch.zeros(self.memory_size)
self.memory_timestamps = torch.zeros(self.memory_size)

def forward(self, x: torch.Tensor,
operation: str = 'encode') -> torch.Tensor:
"""
前向传播

参数:
x: 输入张量
operation: 操作类型 ('encode', 'retrieve', 'consolidate')
"""
if operation == 'encode':
return self._encode_memory(x)
elif operation == 'retrieve':
return self._retrieve_memory(x)
elif operation == 'consolidate':
return self._consolidate_memory(x)
else:
raise ValueError(f"未知操作: {operation}")

def _encode_memory(self, x: torch.Tensor) -> torch.Tensor:
"""编码记忆"""
batch_size = x.size(0)

# 编码输入
encoded = self.encoder(x) # [batch_size, hidden_dim]

# 计算与记忆键的相似度
similarities = F.cosine_similarity(
encoded.unsqueeze(1), # [batch_size, 1, hidden_dim]
self.memory_keys.unsqueeze(0), # [1, memory_size, hidden_dim]
dim=2
) # [batch_size, memory_size]

# 找到最相似的记忆槽
top_similarities, top_indices = torch.topk(similarities, k=3, dim=1)

# 更新记忆
for b in range(batch_size):
for i in range(3):
idx = top_indices[b, i]
sim = top_similarities[b, i]

# 更新记忆强度
if sim > 0.7: # 相似度阈值
# 读取现有记忆
existing_value = self.memory_values[idx]

# 计算输入门
input_gate = torch.sigmoid(self.input_gate(encoded[b]))

# 更新记忆
new_value = input_gate * encoded[b] + (1 – input_gate) * existing_value
self.memory_values.data[idx] = new_value

# 更新记忆强度
self.memory_strengths[idx] = min(1.0, self.memory_strengths[idx] + 0.1)

# 更新访问统计
self.memory_access_counts[idx] += 1
self.memory_timestamps[idx] = time.time()

return encoded

def _retrieve_memory(self, query: torch.Tensor) -> torch.Tensor:
"""检索记忆"""
# 编码查询
encoded_query = self.encoder(query) # [batch_size, hidden_dim]

# 注意力机制
attn_output, attn_weights = self.attention(
query=encoded_query.unsqueeze(1), # [batch_size, 1, hidden_dim]
key=self.memory_keys.unsqueeze(0), # [1, memory_size, hidden_dim]
value=self.memory_values.unsqueeze(0) # [1, memory_size, hidden_dim]
)

# 解码
retrieved = self.decoder(
torch.cat([encoded_query, attn_output.squeeze(1)], dim=1)
)

return retrieved, attn_weights

def _consolidate_memory(self, importance_weights: torch.Tensor) -> torch.Tensor:
"""巩固记忆"""
# 重要性权重 [memory_size]
if importance_weights.dim() == 1:
importance_weights = importance_weights.unsqueeze(0) # [1, memory_size]

# 注意力巩固
consolidated_keys, _ = self.attention(
query=self.memory_keys.unsqueeze(0),
key=self.memory_keys.unsqueeze(0),
value=self.memory_keys.unsqueeze(0)
)

consolidated_values, _ = self.attention(
query=self.memory_values.unsqueeze(0),
key=self.memory_values.unsqueeze(0),
value=self.memory_values.unsqueeze(0)
)

# 重要性加权
importance_weights = importance_weights.unsqueeze(-1) # [1, memory_size, 1]
consolidated_keys = (1 – importance_weights) * self.memory_keys.unsqueeze(0) + \\
importance_weights * consolidated_keys
consolidated_values = (1 – importance_weights) * self.memory_values.unsqueeze(0) + \\
importance_weights * consolidated_values

# 更新记忆
self.memory_keys.data = consolidated_keys.squeeze(0)
self.memory_values.data = consolidated_values.squeeze(0)

# 增加记忆强度
self.memory_strengths = torch.clamp(self.memory_strengths + importance_weights.squeeze(), 0, 1)

return consolidated_values.squeeze(0)

def get_memory_strengths(self) -> torch.Tensor:
"""获取记忆强度"""
return self.memory_strengths

def get_memory_stats(self) -> Dict[str, Any]:
"""获取记忆统计"""
return {
'mean_strength': float(self.memory_strengths.mean().item()),
'std_strength': float(self.memory_strengths.std().item()),
'total_accesses': int(self.memory_access_counts.sum().item()),
'memory_usage': float((self.memory_strengths > 0.1).sum().item() / self.memory_size)
}

3.3 结语

本文详细探讨了Python在实现TVA长期记忆和学习机制方面的优势。我们构建了完整的长期记忆系统，包括：

Python的强大特性使得实现这些复杂认知机制成为可能：

这个长期记忆系统不仅模拟了人类记忆的基本特性，还展示了Python在构建复杂认知系统方面的独特优势。在第四篇文章中，我们将探讨Python如何实现TVA的决策和问题解决系统。

写在最后——以TVA重构工业视觉的理论内涵与能力边界

本文介绍了基于Python的长期记忆与学习机制实现方案，采用TVA（Theory of Visual Attention）理论框架构建了一个完整的记忆系统。该系统包含三大核心模块：1）多类型记忆存储（语义/情景/程序记忆），通过动态索引和关联网络实现高效存取；2）神经网络记忆模型，使用PyTorch实现基于注意力机制的编码-检索机制；3）记忆生命周期管理，包含自动巩固（突触/系统/重激活）、自适应遗忘和记忆优化策略。关键技术亮点包括：基于NetworkX的语义网络构建、多维度记忆强度计算模型、基于余弦相似度的内容检索，以及模拟人类记忆规律的衰减/强化算法。该实现充分发挥了Python在科学计算（NumPy）、机器学习（PyTorch）和复杂系统建模方面的优势，为构建类脑认知系统提供了可扩展的参考架构。