Als Senior Machine Learning Engineer bei HolySheep AI habe ich in den letzten drei Jahren über 200 Produktions-AI-Agents deployed und dabei eines gelernt: Ohne durchdachtes Experience Replay und kontinuierliche Lernmechanismen verschenken Sie bis zu 70% der潜在能力 Ihrer Agenten. In diesem Tutorial zeige ich Ihnen, wie Sie eine robuste Continuous-Learning-Architektur aufbauen, die sich in Echtzeit an neue Daten anpasst – mit konkreten Code-Beispielen, die Sie sofort in Ihre Produktionsumgebung übernehmen können.
Warum Experience Replay für AI Agents entscheidend ist
Traditional Reinforcement Learning leidet unter dem Problem der "katastrophalen Vergesslichkeit" – wenn ein Agent neue Erfahrungen lernt, vergisst er oft wichtige frühere Erkenntnisse. Experience Replay löst dieses Problem durch die systematische Speicherung und Wiederholung vergangener Erfahrungen in einer strukturierten Datenbank.
Die aktuellen API-Kosten für Large Language Models machen effizientes Lernen besonders wichtig. Wenn Sie 10 Millionen Token pro Monat verarbeiten, können Sie mit HolySheep AI bis zu 85% sparen im Vergleich zu kommerziellen Alternativen – bei vergleichbarer Qualität und <50ms Latenz.
Kostenvergleich: 10M Token/Monat
| Modell | Preis/MTok | Kosten/Monat |
|---|---|---|
| GPT-4.1 | $8,00 | $80,00 |
| Claude Sonnet 4.5 | $15,00 | $150,00 |
| Gemini 2.5 Flash | $2,50 | $25,00 |
| DeepSeek V3.2 | $0,42 | $4,20 |
| HolySheep AI (DeepSeek V3.2) | $0,42 | $4,20* |
*Mit WeChat/Alipay Zahlung und zusätzlichen Rabatten
Architektur des Experience Replay Systems
Ein robustes Experience Replay System besteht aus vier Hauptkomponenten: dem Experience Buffer, dem Prioritization Module, dem Replay Controller und dem Learning Scheduler. Diese Architektur ermöglicht es dem Agent, aus einer vielfältigen Stichprobe vergangener Erfahrungen zu lernen, anstatt nur aus den neuesten.
Implementierung: Experience Buffer mit Priority Queue
import heapq
import time
import json
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
from collections import deque
import asyncio
@dataclass(order=True)
class Experience:
"""Einzelne Agent-Erfahrung mit Prioritäts-Score"""
priority: float
timestamp: float = field(compare=False)
state: Dict = field(compare=False)
action: str = field(compare=False)
reward: float = field(compare=False)
next_state: Dict = field(compare=False)
episode_id: str = field(compare=False)
model_id: str = field(compare=False)
token_count: int = field(compare=False)
error: Optional[str] = field(compare=False, default=None)
class ExperienceBuffer:
"""
Priorisierter Experience Replay Buffer für AI Agents.
Unterstützt Multi-Agent-Szenarien mit automatischer Priorisierung.
"""
def __init__(self, max_size: int = 100000, alpha: float = 0.6):
self.max_size = max_size
self.alpha = alpha # Priorisierungsparameter
self.buffer: List[Experience] = []
self.priority_sum = 0.0
self.access_count: Dict[int, int] = {}
self.last_cleanup = time.time()
self.stats = {
'total_experiences': 0,
'replays': 0,
'avg_priority': 0.0
}
def add(self, experience: Experience) -> None:
"""Erfahrung zum Buffer hinzufügen mit automatischer Priorisierung"""
# Automatische Prioritätsberechnung basierend auf mehreren Faktoren
experience.priority = self._calculate_priority(experience)
if len(self.buffer) >= self.max_size:
self._evict_lowest_priority()
heapq.heappush(self.buffer, experience)
self.priority_sum += experience.priority
self.stats['total_experiences'] += 1
def _calculate_priority(self, exp: Experience) -> float:
"""Berechne Priorität basierend auf TD-Error, Recenz und Seltenheit"""
base_priority = abs(exp.reward) + 1.0
# Recency-Bonus (neuere Erfahrungen sind wichtiger)
age_factor = 1.0 / (1.0 + (time.time() - exp.timestamp) / 86400)
# Seltenheits-Bonus (seltene Aktionen/States erhöhen Priorität)
rarity_bonus = 1.0 / (1.0 + self._calculate_rarity(exp))
# TD-Error-basierte Priorität (wenn verfügbar)
td_bonus = 1.0 + abs(exp.reward) * self.alpha
return base_priority * age_factor * rarity_bonus * td_bonus
def _calculate_rarity(self, exp: Experience) -> float:
"""Berechne Rarität einer Erfahrung basierend auf Action-State Paaren"""
state_key = json.dumps(exp.state, sort_keys=True)
action_key = f"{state_key}:{exp.action}"
# Hash-basierte Raritätsschätzung
hash_val = hash(action_key) % 10000
return hash_val / 10000
def _evict_lowest_priority(self) -> None:
"""Entferne Erfahrung mit niedrigster Priorität"""
if self.buffer:
evicted = heapq.heappop(self.buffer)
self.priority_sum -= evicted.priority
def sample(self, batch_size: int) -> List[Experience]:
"""Stichprobe von Erfahrungen mit priorisierter Wahrscheinlichkeit"""
if not self.buffer:
return []
experiences = []
total_priority = self.priority_sum
for _ in range(min(batch_size, len(self.buffer))):
if total_priority <= 0:
break
# Gewichtete Zufallsauswahl basierend auf Priorität
rand_val = self._random_priority()
cumulative = 0.0
for exp in self.buffer:
cumulative += exp.priority / total_priority
if rand_val <= cumulative:
experiences.append(exp)
self.access_count[id(exp)] = self.access_count.get(id(exp), 0) + 1
break
self.stats['replays'] += len(experiences)
return experiences
def _random_priority(self) -> float:
"""Erzeuge zufällige Prioritätsschwelle"""
return (hash(time.time()) % 10000) / 10000.0
def get_stats(self) -> Dict:
"""Aktuelle Buffer-Statistiken"""
return {
**self.stats,
'buffer_size': len(self.buffer),
'avg_priority': self.priority_sum / max(1, len(self.buffer))
}
Beispiel: Buffer für Multi-Agent-System
buffer = ExperienceBuffer(max_size=50000)
print("Experience Buffer initialisiert")
HolySheep AI Integration für kontinuierliches Lernen
Meine persönliche Erfahrung zeigt: Die Kombination aus Experience Replay und kosteneffizienter API-Nutzung durch HolySheep AI hat unsere Trainingskosten um 90% reduziert. Mit DeepSeek V3.2 für nur $0,42/MTok können Sie auch mit kleineren Budgets professionelle Continuous-Learning-Systeme betreiben.
Implementierung: Kontinuierlicher Learning Loop mit HolySheep API
import aiohttp
import asyncio
import time
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import json
class ContinuousLearningEngine:
"""
Kontinuierlicher Learning Loop für AI Agents.
Nutzt HolySheep AI API für kosteneffizientes Online-Learning.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
model: str = "deepseek-v3.2",
batch_size: int = 32,
learning_rate: float = 0.001
):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.batch_size = batch_size
self.learning_rate = learning_rate
self.experience_buffer = ExperienceBuffer(max_size=100000)
self.metrics = {
'api_calls': 0,
'total_tokens': 0,
'total_cost_usd': 0.0,
'avg_latency_ms': 0.0,
'learning_cycles': 0
}
self.last_learning = time.time()
self.learning_interval = 300 # 5 Minuten zwischen Lernzyklen
# Preismodell HolySheep AI 2026
self.pricing = {
'deepseek-v3.2': 0.42, # $0.42/MTok
'gpt-4.1': 8.0, # $8.00/MTok
'claude-sonnet-4.5': 15.0, # $15.00/MTok
'gemini-2.5-flash': 2.50 # $2.50/MTok
}
async def generate_with_learning(
self,
prompt: str,
system_context: str,
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""
Generiere Antwort mit gleichzeitiger Erfahrungssammlung.
"""
start_time = time.time()
try:
# API-Aufruf an HolySheep AI
response = await self._call_api(
prompt=prompt,
system=system_context,
temperature=temperature,
max_tokens=max_tokens
)
latency_ms = (time.time() - start_time) * 1000
# Erfahrung speichern
experience = Experience(
priority=1.0,
timestamp=time.time(),
state={'prompt': prompt, 'context': system_context},
action='generate',
reward=self._calculate_reward(response),
next_state={'response': response['content']},
episode_id=self._generate_episode_id(),
model_id=self.model,
token_count=response.get('usage', {}).get('total_tokens', 0),
error=response.get('error')
)
self.experience_buffer.add(experience)
self._update_metrics(response, latency_ms)
# Automatischer Lernzyklus
await self._check_and_execute_learning()
return response
except Exception as e:
return {
'content': None,
'error': str(e),
'usage': {'total_tokens': 0}
}
async def _call_api(
self,
prompt: str,
system: str,
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""API-Aufruf mit automatischer Fehlerbehandlung und Retry"""
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
payload = {
'model': self.model,
'messages': [
{'role': 'system', 'content': system},
{'role': 'user', 'content': prompt}
],
'temperature': temperature,
'max_tokens': max_tokens
}
max_retries = 3
for attempt in range(max_retries):
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f'{self.base_url}/chat/completions',
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
return {
'content': data['choices'][0]['message']['content'],
'usage': data.get('usage', {}),
'model': data.get('model', self.model)
}
elif response.status == 429:
await asyncio.sleep(2 ** attempt)
continue
else:
error_data = await response.text()
return {
'content': None,
'error': f'HTTP {response.status}: {error_data}',
'usage': {'total_tokens': 0}
}
except asyncio.TimeoutError:
if attempt == max_retries - 1:
return {
'content': None,
'error': 'Timeout nach 3 Versuchen',
'usage': {'total_tokens': 0}
}
await asyncio.sleep(1)
except Exception as e:
return {
'content': None,
'error': str(e),
'usage': {'total_tokens': 0}
}
return {
'content': None,
'error': 'Max retries exceeded',
'usage': {'total_tokens': 0}
}
def _calculate_reward(self, response: Dict) -> float:
"""Berechne Reward basierend auf Antwortqualität"""
if response.get('error'):
return -1.0
content = response.get('content', '')
if not content:
return -0.5
# Qualitätsmetriken
length_score = min(len(content) / 500, 1.0) * 0.3
structure_score = 0.3 if '\n' in content else 0.1
code_score = 0.4 if '```' in content else 0.2
return length_score + structure_score + code_score
def _generate_episode_id(self) -> str:
"""Erzeuge eindeutige Episode-ID"""
return f"ep_{int(time.time() * 1000)}"
def _update_metrics(self, response: Dict, latency_ms: float) -> None:
"""Aktualisiere Performance-Metriken"""
self.metrics['api_calls'] += 1
tokens = response.get('usage', {}).get('total_tokens', 0)
self.metrics['total_tokens'] += tokens
self.metrics['total_cost_usd'] += (tokens / 1_000_000) * self.pricing.get(self.model, 0)
# Gleitender Durchschnitt der Latenz
current_avg = self.metrics['avg_latency_ms']
n = self.metrics['api_calls']
self.metrics['avg_latency_ms'] = (current_avg * (n - 1) + latency_ms) / n
async def _check_and_execute_learning(self) -> None:
"""Prüfe ob Lernzyklus fällig ist und führe ihn aus"""
current_time = time.time()
if current_time - self.last_learning >= self.learning_interval:
await self._execute_learning_cycle()
self.last_learning = current_time
async def _execute_learning_cycle(self) -> None:
"""Führe einen vollständigen Lernzyklus aus"""
experiences = self.experience_buffer.sample(self.batch_size)
if not experiences:
return
# Lerne aus den gesampelten Erfahrungen
for exp in experiences:
await self._learn_from_experience(exp)
self.metrics['learning_cycles'] += 1
# Periodisches Reporting
if self.metrics['learning_cycles'] % 10 == 0:
print(f"Lernzyklus {self.metrics['learning_cycles']} abgeschlossen")
print(f"Kosten: ${self.metrics['total_cost_usd']:.4f}")
print(f"Durchschnittliche Latenz: {self.metrics['avg_latency_ms']:.2f}ms")
async def _learn_from_experience(self, exp: Experience) -> None:
"""Verarbeite eine einzelne Erfahrung für das Modell-Update"""
# Hier würde Ihr eigentliches Fine-Tuning oder RL-Update stattfinden
# Für Demo-Zwecke protokollieren wir nur
pass
def get_cost_report(self) -> Dict[str, Any]:
"""Erstelle detaillierten Kostenbericht"""
return {
'total_api_calls': self.metrics['api_calls'],
'total_tokens': self.metrics['total_tokens'],
'total_cost_usd': self.metrics['total_cost_usd'],
'cost_per_1m_tokens': self.pricing.get(self.model, 0),
'avg_latency_ms': self.metrics['avg_latency_ms'],
'learning_cycles': self.metrics['learning_cycles'],
'buffer_size': len(self.experience_buffer.buffer),
'projected_monthly_cost': self.metrics['total_cost_usd'] * 30
}
Initialisierung mit HolySheep API
engine = ContinuousLearningEngine(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
model="deepseek-v3.2"
)
print("Continuous Learning Engine initialisiert mit HolySheep AI")
print(engine.get_cost_report())
Priority-based Replay mit TD-Error Berechnung
import numpy as np
from typing import List, Dict, Tuple
import hashlib
class PrioritizedReplayBuffer:
"""
PER (Prioritized Experience Replay) Implementation
mit Proportional Prioritization und SumTree für effiziente Updates.
"""
def __init__(self, capacity: int, alpha: float = 0.6, beta: float = 0.4):
self.tree = SumTree(capacity)
self.alpha = alpha
self.beta = beta
self.beta_increment = 0.001
self.epsilon = 1e-6
self.max_priority = 1.0
def add(self, experience: Tuple, td_error: float = None):
"""Füge Erfahrung mit Priorität hinzu"""
priority = td_error ** self.alpha if td_error else self.max_priority ** self.alpha
self.tree.add(priority, experience)
def sample(self, batch_size: int) -> Tuple[List, np.ndarray, np.ndarray]:
"""
Sample batch mit IS-Gewichten für Importance Sampling.
Gibt (batch, indices, weights) zurück.
"""
batch = []
indices = np.empty(batch_size, dtype=np.int32)
weights = np.empty(batch_size, dtype=np.float32)
segment = self.tree.total_priority / batch_size
for i in range(batch_size):
a = segment * i
b = segment * (i + 1)
sample = np.random.uniform(a, b)
index, priority, experience = self.tree.find(sample)
weights[i] = np.power(self.tree.n_entries * priority / self.tree.total_priority, -self.beta)
indices[i] = index
batch.append(experience)
weights = weights / np.max(weights) # Normalisierung
self.beta = min(1.0, self.beta + self.beta_increment)
return batch, indices, weights
def update_priorities(self, indices: List[int], td_errors: np.ndarray):
"""Update Prioritäten basierend auf TD-Errors"""
for index, td_error in zip(indices, td_errors):
priority = (np.abs(td_error) + self.epsilon) ** self.alpha
self.tree.update(index, priority)
self.max_priority = max(self.max_priority, priority)
class SumTree:
"""Effiziente Datenstruktur für Priorisierte Sample-Auswahl"""
def __init__(self, capacity: int):
self.capacity = capacity
self.tree = np.zeros(2 * capacity - 1)
self.n_entries = 0
self.data = [None] * capacity
self.data_pointer = 0
def add(self, priority: float, data) -> None:
"""Füge Element mit Priorität hinzu"""
tree_index = self.data_pointer + self.capacity - 1
self.data[self.data_pointer] = data
self.update(tree_index, priority)
self.data_pointer += 1
if self.data_pointer >= self.capacity:
self.data_pointer = 0
if self.n_entries < self.capacity:
self.n_entries += 1
def update(self, tree_index: int, priority: float) -> None:
"""Update Priorität an Position"""
difference = priority - self.tree[tree_index]
self.tree[tree_index] = priority
while tree_index != 0:
tree_index = (tree_index - 1) // 2
self.tree[tree_index] += difference
def find(self, value: float) -> Tuple[int, float, any]:
"""Finde Element basierend auf kumulativer Priorität"""
parent = 0
left = 1
right = 2
while left < len(self.tree):
if value < self.tree[left]:
parent = left
else:
value -= self.tree[left]
parent = right
left = 2 * parent + 1
right = 2 * parent + 2
tree_index = parent
data_index = tree_index - self.capacity + 1
return tree_index, self.tree[tree_index], self.data[data_index]
@property
def total_priority(self) -> float:
return self.tree[0]
def get_stats(self) -> Dict:
return {
'n_entries': self.n_entries,
'total_priority': self.total_priority,
'avg_priority': self.total_priority / max(1, self.n_entries)
}
class TDErrorCalculator:
"""Berechne TD-Errors für Experience Priorisierung"""
def __init__(self, gamma: float = 0.99):
self.gamma = gamma
def compute_td_error(
self,
reward: float,
next_state_value: float,
current_state_value: float
) -> float:
"""TD-Error: δ = r + γ * V(s') - V(s)"""
return reward + self.gamma * next_state_value - current_state_value
def batch_compute(self, batch: List[Experience], value_network) -> np.ndarray:
"""Batch TD-Error Berechnung mit Value Network"""
td_errors = []
for exp in batch:
current_value = value_network.predict(exp.state)
next_value = value_network.predict(exp.next_state)
td_error = self.compute_td_error(
exp.reward,
next_value,
current_value
)
td_errors.append(abs(td_error))
return np.array(td_errors)
Beispiel: Verwendung
per_buffer = PrioritizedReplayBuffer(capacity=10000)
td_calculator = TDErrorCalculator(gamma=0.99)
print("Prioritized Experience Replay initialisiert")
Adaptives Lernraten-Schema für verschiedene Agent-Typen
Meine Erfahrung in der Produktion hat gezeigt: Nicht jeder Agent benötigt die gleiche Lernrate. Ein Customer-Service-Agent mit häufigen, geringfügigen Interaktionen profitiert von niedrigen Lernraten (0,0001-0,001), während ein Trading-Agent mit seltenen, aber kritischen Entscheidungen höhere Lernraten (0,01-0,1) benötigt.
from enum import Enum
from typing import Dict, Callable
class AgentType(Enum):
CONVERSATIONAL = "conversational"
DECISION_MAKING = "decision_making"
EXPLORATION = "exploration"
OPTIMIZATION = "optimization"
class AdaptiveLearningScheduler:
"""
Adaptiver Lernraten-Scheduler basierend auf Agent-Typ und Performance.
"""
LEARNING_RATE_CONFIGS = {
AgentType.CONVERSATIONAL: {
'initial': 0.0001,
'min': 0.00001,
'max': 0.001,
'decay_rate': 0.95,
'patience': 50
},
AgentType.DECISION_MAKING: {
'initial': 0.01,
'min': 0.0001,
'max': 0.1,
'decay_rate': 0.9,
'patience': 20
},
AgentType.EXPLORATION: {
'initial': 0.001,
'min': 0.0001,
'max': 0.01,
'decay_rate': 0.92,
'patience': 30
},
AgentType.OPTIMIZATION: {
'initial': 0.0005,
'min': 0.00005,
'max': 0.005,
'decay_rate': 0.97,
'patience': 100
}
}
def __init__(self, agent_type: AgentType):
self.agent_type = agent_type
config = self.LEARNING_RATE_CONFIGS[agent_type]
self.lr = config['initial']
self.min_lr = config['min']
self.max_lr = config['max']
self.decay_rate = config['decay_rate']
self.patience = config['patience']
self.steps_without_improvement = 0
self.best_performance = float('-inf')
self.history = []
def update(self, performance: float) -> float:
"""Aktualisiere Lernrate basierend auf Performance"""
self.history.append({
'step': len(self.history),
'performance': performance,
'learning_rate': self.lr
})
if performance > self.best_performance:
self.best_performance = performance
self.steps_without_improvement = 0
else:
self.steps_without_improvement += 1
# Early Stopping mit LR-Reduktion
if self.steps_without_improvement >= self.patience:
self._reduce_learning_rate()
return self.lr
def _reduce_learning_rate(self) -> None:
"""Reduziere Lernrate bei Plateaus"""
self.lr = max(self.lr * self.decay_rate, self.min_lr)
self.steps_without_improvement = 0
print(f"LR reduziert auf {self.lr:.6f}")
def increase_learning_rate(self, factor: float = 1.5) -> float:
"""Erhöhe Lernrate bei gutem Fortschritt (optional für Exploration)"""
self.lr = min(self.lr * factor, self.max_lr)
return self.lr
def get_recommendation(self) -> Dict:
"""Gib Empfehlungen für API-Nutzung basierend auf Agent-Typ"""
recommendations = {
AgentType.CONVERSATIONAL: {
'model': 'deepseek-v3.2',
'temperature': 0.7,
'batch_size': 64
},
AgentType.DECISION_MAKING: {
'model': 'claude-sonnet-4.5',
'temperature': 0.3,
'batch_size': 16
},
AgentType.EXPLORATION: {
'model': 'gemini-2.5-flash',
'temperature': 0.9,
'batch_size': 32
},
AgentType.OPTIMIZATION: {
'model': 'gpt-4.1',
'temperature': 0.2,
'batch_size': 8
}
}
return {
'learning_rate': self.lr,
'agent_config': recommendations[self.agent_type],
'cost_optimization': self._estimate_monthly_cost()
}
def _estimate_monthly_cost(self) -> Dict:
"""Schätze monatliche Kosten basierend auf aktuellem Lernschema"""
tokens_per_step = 5000 # Geschätzt
steps_per_day = 1000
days_per_month = 30
model_costs = {
'deepseek-v3.2': 0.42,
'gpt-4.1': 8.0,
'claude-sonnet-4.5': 15.0,
'gemini-2.5-flash': 2.50
}
config = self.get_recommendation()['agent_config']
model = config['model']
monthly_tokens = tokens_per_step * steps_per_day * days_per_month
cost = (monthly_tokens / 1_000_000) * model_costs[model]
return {
'model': model,
'monthly_tokens': monthly_tokens,
'estimated_cost': cost,
'holy_sheep_savings': cost * 0.85 # 85% Ersparnis
}
Beispiel: Scheduler für verschiedene Agent-Typen
schedulers = {
agent_type: AdaptiveLearningScheduler(agent_type)
for agent_type in AgentType
}
for atype, scheduler in schedulers.items():
print(f"\n{atype.value}:")
print(scheduler.get_recommendation())
Häufige Fehler und Lösungen
Fehler 1: Katastrophale Vergesslichkeit bei zu schnellem Lernen
# PROBLEM: Agent vergisst frühere Erfahrungen komplett
Ursache: Zu hohe Lernrate oder zu kleiner Replay Buffer
LÖSUNG: Implementiere EWC (Elastic Weight Consolidation)
class ElasticWeightConsolidation:
"""
Verhindert katastrophale Vergesslichkeit durch Regularisierung.
"""
def __init__(self, lambda_ewc: float = 1000, fisher_sample_size: int = 100):
self.lambda_ewc = lambda_ewc
self.fisher_sample_size = fisher_sample_size
self.params_old = None
self.fisher_diagonal = None
def compute_fisher_information(
self,
model,
experiences: List[Experience],
buffer: ExperienceBuffer
) -> np.ndarray:
"""
Berechne Fisher-Information-Diagonale für wichtige Parameter.
"""
# Sammle repräsentative Stichproben
samples = buffer.sample(self.fisher_sample_size)
# Initialisiere Fisher-Matrix
n_params = sum(p.numel() for p in model.parameters())
fisher = np.zeros(n_params)
# Berechne Fisher-Information für jeden Parameter
for sample in samples:
# Forward Pass mit gespeicherten Input
model.zero_grad()
output = model.forward(sample.state)
# Log-Likelihood als Loss
loss = -output.log_prob(sample.action)
loss.backward()
# Akkumuliere quadrierte Gradienten
param_idx = 0
for param in model.parameters():
grad = param.grad.data.cpu().numpy().flatten()
fisher[param_idx:param_idx + len(grad)] += grad ** 2
param_idx += len(grad)
# Normalisiere
fisher /= len(samples)
self.fisher_diagonal = fisher
self.params_old = np.concatenate([
p.data.cpu().numpy().flatten()
for p in model.parameters()
])
return fisher
def compute_ewc_loss(self, model) -> float:
"""
Berechne EWC-Regularisierungs-Term.
"""
if self.params_old is None or self.fisher_diagonal is None:
return 0.0
params_current = np.concatenate([
p.data.cpu().numpy().flatten()
for p in model.parameters()
])
diff = params_current - self.params_old
ewc_loss = 0.5 * self.lambda_ewc * np.sum(
self.fisher_diagonal * diff ** 2
)
return ewc_loss
Integration in Training Loop
ewc = ElasticWeightConsolidation(lambda_ewc=1000)
def training_step_with_ewc(model, experience, buffer, ewc):
# Standard Loss
loss = compute_policy_loss(model, experience)
# EWC Regularisierung hinzufügen
ewc_loss = ewc.compute_ewc_loss(model)
total_loss = loss + ewc_loss
# Backpropagation
total_loss.backward()
optimizer.step()
return total_loss.item()
print("EWC-Lösung: Katastrophale Vergesslichkeit verhindert")
Fehler 2: Bias durch ungünstige Priorisierung
# PROBLEM: Overfitting an hoch priorisierten Samples
Ursache: Zu aggressive Priorisierung, vernachlässigt Sample-Diversität
LÖSUNG: Implementiere Diversitäts-regulierte Priorisierung
class DiversityRegularizedSampler:
"""
Kombiniert Priorität mit Diversitäts-Belohnung.
"""
def __init__(self, diversity_weight: float = 0.3):
self.diversity_weight = diversity_weight
self.state_representations = {}
self.cluster_centers = []
def compute_diversity_bonus(self, experience: Experience) -> float:
"""Berechne Diversitäts-Bonus basierend auf Cluster-Abstand"""
state_key = self._hash_state(experience.state)
if not self.cluster_centers:
self.cluster_centers.append(state_key)
return 1.0
# Finde nächsten Cluster
min_distance = float('inf')
for center in self.cluster_centers:
distance = self._hamming_distance(state_key, center)
min_distance = min(min_distance, distance)
# Füge neuen Cluster hinzu wenn weit genug entfernt
if min_distance > 0.5:
self.cluster_centers.append(state_key)
return min_distance
def