- 비전 및 철학: 존재형 AI 에이전트 개념 - 윤리 원칙과 안전 기준 - 사용자 시나리오 및 유즈케이스 - 게임화 요소 (레벨업, 스탯, 스킬) - 기술 아키텍처 (기억 시스템, 감정 모델, DB 설계) - 멀티 에이전트 협업 구조 - DID 기반 신원 체계 - 장기 로드맵 (1년, 3년 비전)
425 lines
13 KiB
Markdown
425 lines
13 KiB
Markdown
# 기억 시스템의 수학적 모델
|
||
|
||
## 엔트로피 기반 기억 관리
|
||
|
||
### 정보 엔트로피 계산
|
||
|
||
기억의 중요도를 정보 이론의 엔트로피로 측정합니다.
|
||
|
||
```python
|
||
import numpy as np
|
||
from scipy.stats import entropy
|
||
|
||
class MemoryEntropy:
|
||
def calculate_information_entropy(self, memory_item):
|
||
"""
|
||
Shannon 엔트로피를 이용한 정보량 계산
|
||
H(X) = -Σ p(xi) * log2(p(xi))
|
||
"""
|
||
# 메모리 항목의 특징 분포
|
||
features = self.extract_features(memory_item)
|
||
probabilities = self.normalize_distribution(features)
|
||
|
||
# 엔트로피 계산
|
||
H = -np.sum(probabilities * np.log2(probabilities + 1e-10))
|
||
|
||
return H
|
||
|
||
def calculate_relative_entropy(self, memory_new, memory_context):
|
||
"""
|
||
KL-Divergence를 이용한 상대적 중요도
|
||
D_KL(P||Q) = Σ P(i) * log(P(i)/Q(i))
|
||
"""
|
||
p = self.get_distribution(memory_new)
|
||
q = self.get_distribution(memory_context)
|
||
|
||
return entropy(p, q)
|
||
|
||
def information_gain(self, memory_before, memory_after):
|
||
"""
|
||
새로운 정보가 추가한 가치 측정
|
||
IG = H(before) - H(after|new_info)
|
||
"""
|
||
h_before = self.calculate_information_entropy(memory_before)
|
||
h_after = self.calculate_information_entropy(memory_after)
|
||
|
||
return h_before - h_after
|
||
```
|
||
|
||
### 시간 감쇠 함수
|
||
|
||
에빙하우스 망각 곡선을 기반으로 한 기억 감쇠 모델:
|
||
|
||
```python
|
||
class ForgettingCurve:
|
||
def __init__(self):
|
||
self.initial_strength = 1.0
|
||
self.decay_rate = 0.5
|
||
|
||
def memory_strength(self, time_elapsed, repetitions=0):
|
||
"""
|
||
S(t) = S0 * exp(-λt) * (1 + α * log(1 + repetitions))
|
||
|
||
S(t): 시간 t에서의 기억 강도
|
||
S0: 초기 기억 강도
|
||
λ: 감쇠율
|
||
α: 반복 학습 효과 계수
|
||
"""
|
||
base_decay = self.initial_strength * np.exp(-self.decay_rate * time_elapsed)
|
||
repetition_bonus = 1 + 0.2 * np.log(1 + repetitions)
|
||
|
||
return base_decay * repetition_bonus
|
||
|
||
def optimal_review_time(self, target_retention=0.9):
|
||
"""
|
||
목표 기억률을 유지하기 위한 최적 복습 시점
|
||
t_optimal = -ln(target_retention) / λ
|
||
"""
|
||
return -np.log(target_retention) / self.decay_rate
|
||
```
|
||
|
||
## 베이지안 기억 우선순위
|
||
|
||
### 베이지안 추론 모델
|
||
|
||
기억의 미래 유용성을 베이지안 추론으로 예측:
|
||
|
||
```python
|
||
class BayesianMemoryPriority:
|
||
def __init__(self):
|
||
self.prior_probabilities = {}
|
||
self.likelihood_matrix = {}
|
||
|
||
def calculate_posterior(self, memory_item, context):
|
||
"""
|
||
베이즈 정리: P(A|B) = P(B|A) * P(A) / P(B)
|
||
|
||
P(유용|맥락) = P(맥락|유용) * P(유용) / P(맥락)
|
||
"""
|
||
prior = self.get_prior(memory_item)
|
||
likelihood = self.get_likelihood(context, memory_item)
|
||
evidence = self.get_evidence(context)
|
||
|
||
posterior = (likelihood * prior) / (evidence + 1e-10)
|
||
return posterior
|
||
|
||
def update_beliefs(self, observation):
|
||
"""
|
||
온라인 베이지안 업데이트
|
||
P_new(θ) ∝ P(data|θ) * P_old(θ)
|
||
"""
|
||
for memory_id, memory in self.memories.items():
|
||
old_prior = self.prior_probabilities.get(memory_id, 0.5)
|
||
likelihood = self.compute_likelihood(observation, memory)
|
||
|
||
# 베이지안 업데이트
|
||
new_prior = likelihood * old_prior
|
||
|
||
# 정규화
|
||
self.prior_probabilities[memory_id] = new_prior / sum(self.prior_probabilities.values())
|
||
|
||
def predict_utility(self, memory_item, future_contexts):
|
||
"""
|
||
미래 맥락에서의 기억 유용성 예측
|
||
U(m) = Σ P(c) * V(m, c)
|
||
"""
|
||
utility = 0
|
||
for context, prob in future_contexts.items():
|
||
value = self.calculate_value(memory_item, context)
|
||
utility += prob * value
|
||
|
||
return utility
|
||
```
|
||
|
||
## 그래프 기반 연상 기억
|
||
|
||
### 의미 네트워크 구조
|
||
|
||
```python
|
||
import networkx as nx
|
||
|
||
class SemanticMemoryGraph:
|
||
def __init__(self):
|
||
self.graph = nx.DiGraph()
|
||
self.activation_threshold = 0.3
|
||
|
||
def add_memory(self, memory_id, content, embeddings):
|
||
"""노드로 기억 추가"""
|
||
self.graph.add_node(
|
||
memory_id,
|
||
content=content,
|
||
embedding=embeddings,
|
||
activation=0.0,
|
||
last_accessed=time.time()
|
||
)
|
||
|
||
def create_association(self, memory1, memory2, strength):
|
||
"""연관성 엣지 생성"""
|
||
self.graph.add_edge(
|
||
memory1, memory2,
|
||
weight=strength,
|
||
co_activation_count=0
|
||
)
|
||
|
||
def spreading_activation(self, source_memory, decay=0.8):
|
||
"""
|
||
활성화 확산 알고리즘
|
||
A(n) = Σ w(m,n) * A(m) * decay^d(m,n)
|
||
"""
|
||
activation = {source_memory: 1.0}
|
||
visited = set()
|
||
queue = [(source_memory, 1.0, 0)]
|
||
|
||
while queue:
|
||
current, current_activation, depth = queue.pop(0)
|
||
|
||
if current in visited:
|
||
continue
|
||
visited.add(current)
|
||
|
||
# 이웃 노드로 활성화 전파
|
||
for neighbor in self.graph.neighbors(current):
|
||
edge_weight = self.graph[current][neighbor]['weight']
|
||
new_activation = current_activation * edge_weight * (decay ** depth)
|
||
|
||
if new_activation > self.activation_threshold:
|
||
activation[neighbor] = max(
|
||
activation.get(neighbor, 0),
|
||
new_activation
|
||
)
|
||
queue.append((neighbor, new_activation, depth + 1))
|
||
|
||
return activation
|
||
|
||
def retrieve_associated_memories(self, query_memory, top_k=5):
|
||
"""연상 기억 검색"""
|
||
activations = self.spreading_activation(query_memory)
|
||
|
||
# 활성화 강도순 정렬
|
||
sorted_memories = sorted(
|
||
activations.items(),
|
||
key=lambda x: x[1],
|
||
reverse=True
|
||
)
|
||
|
||
return sorted_memories[:top_k]
|
||
```
|
||
|
||
## 계층적 시간 기억 (HTM)
|
||
|
||
### Hierarchical Temporal Memory 구현
|
||
|
||
```python
|
||
class HierarchicalTemporalMemory:
|
||
def __init__(self, input_size, column_count, cells_per_column):
|
||
self.input_size = input_size
|
||
self.column_count = column_count
|
||
self.cells_per_column = cells_per_column
|
||
|
||
# 시냅스 연결 강도
|
||
self.proximal_synapses = np.random.rand(column_count, input_size)
|
||
self.distal_synapses = np.zeros((column_count, cells_per_column, column_count, cells_per_column))
|
||
|
||
def spatial_pooling(self, input_vector):
|
||
"""
|
||
공간 풀링: 입력을 희소 분산 표현으로 변환
|
||
"""
|
||
# 각 컬럼의 overlap 계산
|
||
overlaps = np.dot(self.proximal_synapses, input_vector)
|
||
|
||
# 상위 k% 컬럼만 활성화 (희소성)
|
||
k = int(0.02 * self.column_count) # 2% 활성화
|
||
active_columns = np.argpartition(overlaps, -k)[-k:]
|
||
|
||
# 부스팅으로 사용 빈도 균등화
|
||
boost_factors = self.calculate_boost(active_columns)
|
||
overlaps *= boost_factors
|
||
|
||
return active_columns
|
||
|
||
def temporal_memory(self, active_columns, previous_state):
|
||
"""
|
||
시간 기억: 시퀀스 학습과 예측
|
||
"""
|
||
predicted_cells = []
|
||
active_cells = []
|
||
|
||
for col in active_columns:
|
||
# 예측된 셀이 있으면 활성화
|
||
if self.has_predicted_cell(col, previous_state):
|
||
active_cells.append(self.get_predicted_cell(col))
|
||
else:
|
||
# 버스트: 모든 셀 활성화
|
||
for cell in range(self.cells_per_column):
|
||
active_cells.append((col, cell))
|
||
|
||
# 다음 상태 예측
|
||
predicted_cells.extend(self.predict_next_cells(col))
|
||
|
||
return active_cells, predicted_cells
|
||
|
||
def learn_sequence(self, sequence):
|
||
"""시퀀스 패턴 학습"""
|
||
previous_state = None
|
||
|
||
for item in sequence:
|
||
active_columns = self.spatial_pooling(item)
|
||
active_cells, predicted = self.temporal_memory(active_columns, previous_state)
|
||
|
||
# 시냅스 강화
|
||
self.reinforce_synapses(active_cells, predicted)
|
||
previous_state = active_cells
|
||
```
|
||
|
||
## 압축 기억 모델
|
||
|
||
### 정보 압축과 복원
|
||
|
||
```python
|
||
class CompressiveMemory:
|
||
def __init__(self, compression_ratio=0.1):
|
||
self.compression_ratio = compression_ratio
|
||
self.dictionary = {} # 압축 사전
|
||
|
||
def compress_memory(self, memory_stream):
|
||
"""
|
||
PCA 기반 차원 축소로 기억 압축
|
||
"""
|
||
from sklearn.decomposition import PCA
|
||
|
||
# 원본 차원
|
||
original_dim = memory_stream.shape[1]
|
||
compressed_dim = int(original_dim * self.compression_ratio)
|
||
|
||
# PCA 압축
|
||
pca = PCA(n_components=compressed_dim)
|
||
compressed = pca.fit_transform(memory_stream)
|
||
|
||
# 복원을 위한 정보 저장
|
||
self.dictionary['pca'] = pca
|
||
self.dictionary['original_dim'] = original_dim
|
||
|
||
# 압축률과 정보 손실 계산
|
||
reconstruction_error = self.calculate_reconstruction_error(
|
||
memory_stream,
|
||
pca.inverse_transform(compressed)
|
||
)
|
||
|
||
return compressed, reconstruction_error
|
||
|
||
def adaptive_compression(self, memory, importance_score):
|
||
"""
|
||
중요도에 따른 적응적 압축
|
||
높은 중요도 = 낮은 압축률
|
||
"""
|
||
adaptive_ratio = self.compression_ratio * (2 - importance_score)
|
||
adaptive_ratio = np.clip(adaptive_ratio, 0.05, 0.95)
|
||
|
||
return self.compress_with_ratio(memory, adaptive_ratio)
|
||
|
||
def hierarchical_compression(self, memories):
|
||
"""
|
||
계층적 압축: 자주 접근하는 기억은 덜 압축
|
||
"""
|
||
layers = {
|
||
'hot': 0.9, # 거의 압축 안함
|
||
'warm': 0.5, # 중간 압축
|
||
'cold': 0.1 # 높은 압축
|
||
}
|
||
|
||
compressed_memories = {}
|
||
for memory_id, memory_data in memories.items():
|
||
access_frequency = self.get_access_frequency(memory_id)
|
||
|
||
if access_frequency > 10:
|
||
layer = 'hot'
|
||
elif access_frequency > 3:
|
||
layer = 'warm'
|
||
else:
|
||
layer = 'cold'
|
||
|
||
compressed_memories[memory_id] = self.compress_with_ratio(
|
||
memory_data,
|
||
layers[layer]
|
||
)
|
||
|
||
return compressed_memories
|
||
```
|
||
|
||
## 양자 기억 모델 (실험적)
|
||
|
||
### 양자 중첩 상태 기억
|
||
|
||
```python
|
||
class QuantumMemory:
|
||
"""
|
||
양자 컴퓨팅 원리를 적용한 기억 모델
|
||
여러 상태를 동시에 유지
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.superposition_states = {}
|
||
self.entangled_memories = {}
|
||
|
||
def create_superposition(self, memory_states):
|
||
"""
|
||
여러 가능한 기억 상태의 중첩
|
||
|ψ⟩ = α|0⟩ + β|1⟩ + γ|2⟩ + ...
|
||
"""
|
||
amplitudes = self.normalize_amplitudes(memory_states)
|
||
|
||
superposition = {
|
||
'states': memory_states,
|
||
'amplitudes': amplitudes,
|
||
'coherence': 1.0 # 결맞음 정도
|
||
}
|
||
|
||
return superposition
|
||
|
||
def quantum_entanglement(self, memory1, memory2):
|
||
"""
|
||
기억 간 양자 얽힘
|
||
한 기억의 관측이 다른 기억에 즉시 영향
|
||
"""
|
||
entangled_state = {
|
||
'memory_pair': (memory1, memory2),
|
||
'correlation_matrix': self.calculate_correlation(memory1, memory2),
|
||
'bell_state': self.create_bell_state(memory1, memory2)
|
||
}
|
||
|
||
self.entangled_memories[(memory1.id, memory2.id)] = entangled_state
|
||
|
||
def collapse_wavefunction(self, superposition, observation):
|
||
"""
|
||
관측 시 파동함수 붕괴
|
||
확률적으로 하나의 상태로 수렴
|
||
"""
|
||
probabilities = np.abs(superposition['amplitudes']) ** 2
|
||
collapsed_state = np.random.choice(
|
||
superposition['states'],
|
||
p=probabilities
|
||
)
|
||
|
||
return collapsed_state
|
||
```
|
||
|
||
## 성능 메트릭
|
||
|
||
### 기억 시스템 평가 지표
|
||
|
||
```python
|
||
def evaluate_memory_system(memory_system):
|
||
metrics = {
|
||
'recall_precision': calculate_precision(memory_system),
|
||
'recall_speed': measure_retrieval_time(memory_system),
|
||
'storage_efficiency': calculate_compression_ratio(memory_system),
|
||
'association_accuracy': test_association_strength(memory_system),
|
||
'temporal_coherence': measure_sequence_prediction(memory_system),
|
||
'information_retention': calculate_retention_rate(memory_system)
|
||
}
|
||
|
||
# 종합 점수
|
||
overall_score = np.mean(list(metrics.values()))
|
||
|
||
return metrics, overall_score
|
||
``` |