damjan_savicandClaude Opus 4.5
e1bbe5455d
Vollständige Next.js 15 Portfolio-Website mit: - Blog-System mit 100+ Artikeln - Supabase-Integration - Responsive Design mit Tailwind CSS - TypeScript-Konfiguration - Testing-Setup mit Vitest und Playwright Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
593 lines
15 KiB
Markdown
593 lines
15 KiB
Markdown
# Event-Driven Architecture für AI-Systeme
|
||
|
||
**Meta-Description:** Design patterns für ereignisgesteuerte AI-Architekturen. Message Queues, Event Sourcing und Reactive Pipelines für skalierbare AI-Anwendungen.
|
||
|
||
**Keywords:** Event-Driven Architecture, AI Systems, Message Queue, Redis, RabbitMQ, Event Sourcing, Reactive AI, BullMQ
|
||
|
||
---
|
||
|
||
## Einführung
|
||
|
||
AI-Systeme sind inhärent asynchron: LLM-Calls dauern Sekunden, nicht Millisekunden. Event-Driven Architecture (EDA) ist die natürliche Lösung für **skalierbare, resiliente AI-Pipelines**.
|
||
|
||
---
|
||
|
||
## Warum EDA für AI?
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────┐
|
||
│ WHY EVENT-DRIVEN FOR AI? │
|
||
├─────────────────────────────────────────────────────────────┤
|
||
│ │
|
||
│ Problem: Request/Response für AI │
|
||
│ ───────────────────────────────────────── │
|
||
│ Client ──Request──→ Server ──LLM Call (5s)──→ Response │
|
||
│ └── Timeout! Connection lost! │
|
||
│ │
|
||
│ Lösung: Event-Driven │
|
||
│ ───────────────────────────────────────── │
|
||
│ Client ──Event──→ Queue ──Worker──→ LLM ──Event──→ Client │
|
||
│ └── Sofort bestätigt, async verarbeitet │
|
||
│ │
|
||
│ Vorteile: │
|
||
│ ✓ Keine Timeouts │
|
||
│ ✓ Retry bei Fehlern │
|
||
│ ✓ Horizontal skalierbar │
|
||
│ ✓ Lastverteilung │
|
||
│ ✓ Entkopplung │
|
||
│ │
|
||
└─────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## Architecture Pattern: AI Processing Pipeline
|
||
|
||
```typescript
|
||
// src/events/types.ts
|
||
interface AIEvent {
|
||
id: string;
|
||
type: string;
|
||
timestamp: Date;
|
||
payload: any;
|
||
metadata: {
|
||
userId?: string;
|
||
correlationId: string;
|
||
source: string;
|
||
};
|
||
}
|
||
|
||
// Event Types
|
||
type AIEventType =
|
||
| 'analysis.requested'
|
||
| 'analysis.started'
|
||
| 'analysis.completed'
|
||
| 'analysis.failed'
|
||
| 'llm.call.started'
|
||
| 'llm.call.completed'
|
||
| 'notification.send';
|
||
```
|
||
|
||
---
|
||
|
||
## BullMQ Implementation
|
||
|
||
### Queue Setup
|
||
|
||
```typescript
|
||
// src/queues/ai-queue.ts
|
||
import { Queue, Worker, Job } from 'bullmq';
|
||
import { Redis } from 'ioredis';
|
||
|
||
const redis = new Redis(process.env.REDIS_URL!, {
|
||
maxRetriesPerRequest: null
|
||
});
|
||
|
||
// AI Processing Queue
|
||
export const aiQueue = new Queue('ai-processing', {
|
||
connection: redis,
|
||
defaultJobOptions: {
|
||
attempts: 3,
|
||
backoff: {
|
||
type: 'exponential',
|
||
delay: 1000
|
||
},
|
||
removeOnComplete: {
|
||
count: 1000,
|
||
age: 24 * 3600 // 24h
|
||
},
|
||
removeOnFail: {
|
||
count: 5000
|
||
}
|
||
}
|
||
});
|
||
|
||
// Job hinzufügen
|
||
export async function queueAITask(
|
||
type: string,
|
||
data: any,
|
||
priority: number = 0
|
||
): Promise<string> {
|
||
const job = await aiQueue.add(type, data, {
|
||
priority,
|
||
jobId: `${type}-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
|
||
});
|
||
|
||
return job.id!;
|
||
}
|
||
```
|
||
|
||
### Worker Implementation
|
||
|
||
```typescript
|
||
// src/workers/ai-worker.ts
|
||
import { Worker, Job } from 'bullmq';
|
||
import Anthropic from '@anthropic-ai/sdk';
|
||
import { eventBus } from './event-bus';
|
||
|
||
const anthropic = new Anthropic();
|
||
|
||
export const aiWorker = new Worker(
|
||
'ai-processing',
|
||
async (job: Job) => {
|
||
const { type, data } = job;
|
||
|
||
// Event: Processing started
|
||
await eventBus.emit('analysis.started', {
|
||
jobId: job.id,
|
||
type
|
||
});
|
||
|
||
try {
|
||
switch (type) {
|
||
case 'analyze-product':
|
||
return await analyzeProduct(job, data);
|
||
|
||
case 'generate-content':
|
||
return await generateContent(job, data);
|
||
|
||
case 'summarize-document':
|
||
return await summarizeDocument(job, data);
|
||
|
||
default:
|
||
throw new Error(`Unknown job type: ${type}`);
|
||
}
|
||
} catch (error) {
|
||
// Event: Processing failed
|
||
await eventBus.emit('analysis.failed', {
|
||
jobId: job.id,
|
||
error: error.message
|
||
});
|
||
throw error;
|
||
}
|
||
},
|
||
{
|
||
connection: redis,
|
||
concurrency: 5, // 5 parallele Jobs
|
||
limiter: {
|
||
max: 10, // Max 10 Jobs
|
||
duration: 1000 // pro Sekunde (Rate Limiting)
|
||
}
|
||
}
|
||
);
|
||
|
||
async function analyzeProduct(job: Job, data: ProductData) {
|
||
// Progress tracking
|
||
await job.updateProgress(10);
|
||
|
||
const response = await anthropic.messages.create({
|
||
model: 'claude-3-haiku-20240307',
|
||
max_tokens: 500,
|
||
messages: [{
|
||
role: 'user',
|
||
content: `Analysiere dieses Produkt: ${JSON.stringify(data)}`
|
||
}]
|
||
});
|
||
|
||
await job.updateProgress(90);
|
||
|
||
const result = {
|
||
analysis: response.content[0].text,
|
||
confidence: 0.85,
|
||
timestamp: new Date()
|
||
};
|
||
|
||
// Event: Processing completed
|
||
await eventBus.emit('analysis.completed', {
|
||
jobId: job.id,
|
||
result
|
||
});
|
||
|
||
await job.updateProgress(100);
|
||
|
||
return result;
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Event Bus Implementation
|
||
|
||
```typescript
|
||
// src/events/event-bus.ts
|
||
import { EventEmitter } from 'events';
|
||
import { Redis } from 'ioredis';
|
||
|
||
class EventBus extends EventEmitter {
|
||
private publisher: Redis;
|
||
private subscriber: Redis;
|
||
|
||
constructor() {
|
||
super();
|
||
this.publisher = new Redis(process.env.REDIS_URL!);
|
||
this.subscriber = new Redis(process.env.REDIS_URL!);
|
||
|
||
this.setupSubscriber();
|
||
}
|
||
|
||
private setupSubscriber() {
|
||
this.subscriber.on('message', (channel, message) => {
|
||
const event = JSON.parse(message);
|
||
this.emit(event.type, event);
|
||
});
|
||
}
|
||
|
||
async subscribe(pattern: string) {
|
||
await this.subscriber.psubscribe(pattern);
|
||
}
|
||
|
||
async emit(type: string, payload: any): Promise<void> {
|
||
const event: AIEvent = {
|
||
id: crypto.randomUUID(),
|
||
type,
|
||
timestamp: new Date(),
|
||
payload,
|
||
metadata: {
|
||
correlationId: payload.correlationId || crypto.randomUUID(),
|
||
source: 'ai-service'
|
||
}
|
||
};
|
||
|
||
// Lokal emittieren
|
||
super.emit(type, event);
|
||
|
||
// An andere Services publishen
|
||
await this.publisher.publish('ai-events', JSON.stringify(event));
|
||
|
||
// Event Log speichern
|
||
await this.logEvent(event);
|
||
}
|
||
|
||
private async logEvent(event: AIEvent) {
|
||
await this.publisher.xadd(
|
||
'events-log',
|
||
'*',
|
||
'event', JSON.stringify(event)
|
||
);
|
||
}
|
||
}
|
||
|
||
export const eventBus = new EventBus();
|
||
```
|
||
|
||
---
|
||
|
||
## Real-Time Updates via WebSocket
|
||
|
||
```typescript
|
||
// src/websocket/event-stream.ts
|
||
import { WebSocketServer } from 'ws';
|
||
import { eventBus } from '../events/event-bus';
|
||
|
||
export function setupEventStream(server: any) {
|
||
const wss = new WebSocketServer({ server, path: '/events' });
|
||
|
||
wss.on('connection', (ws, req) => {
|
||
const userId = extractUserId(req);
|
||
console.log(`Event stream connected: ${userId}`);
|
||
|
||
// Event Listener für diesen User
|
||
const handlers = {
|
||
'analysis.started': (event: AIEvent) => {
|
||
if (event.metadata.userId === userId) {
|
||
ws.send(JSON.stringify(event));
|
||
}
|
||
},
|
||
'analysis.completed': (event: AIEvent) => {
|
||
if (event.metadata.userId === userId) {
|
||
ws.send(JSON.stringify(event));
|
||
}
|
||
},
|
||
'analysis.failed': (event: AIEvent) => {
|
||
if (event.metadata.userId === userId) {
|
||
ws.send(JSON.stringify(event));
|
||
}
|
||
}
|
||
};
|
||
|
||
// Event Listener registrieren
|
||
Object.entries(handlers).forEach(([type, handler]) => {
|
||
eventBus.on(type, handler);
|
||
});
|
||
|
||
ws.on('close', () => {
|
||
// Cleanup
|
||
Object.entries(handlers).forEach(([type, handler]) => {
|
||
eventBus.off(type, handler);
|
||
});
|
||
});
|
||
});
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Client-Side Integration
|
||
|
||
```typescript
|
||
// src/client/event-client.ts
|
||
class AIEventClient {
|
||
private ws: WebSocket | null = null;
|
||
private pendingJobs = new Map<string, {
|
||
resolve: (result: any) => void;
|
||
reject: (error: any) => void;
|
||
}>();
|
||
|
||
connect(url: string) {
|
||
this.ws = new WebSocket(url);
|
||
|
||
this.ws.onmessage = (event) => {
|
||
const aiEvent: AIEvent = JSON.parse(event.data);
|
||
this.handleEvent(aiEvent);
|
||
};
|
||
}
|
||
|
||
private handleEvent(event: AIEvent) {
|
||
const jobId = event.payload.jobId;
|
||
const pending = this.pendingJobs.get(jobId);
|
||
|
||
switch (event.type) {
|
||
case 'analysis.started':
|
||
this.onProgress?.(jobId, 'started');
|
||
break;
|
||
|
||
case 'analysis.completed':
|
||
if (pending) {
|
||
pending.resolve(event.payload.result);
|
||
this.pendingJobs.delete(jobId);
|
||
}
|
||
break;
|
||
|
||
case 'analysis.failed':
|
||
if (pending) {
|
||
pending.reject(new Error(event.payload.error));
|
||
this.pendingJobs.delete(jobId);
|
||
}
|
||
break;
|
||
}
|
||
}
|
||
|
||
// Promise-basierte API über Events
|
||
async analyzeProduct(product: ProductData): Promise<AnalysisResult> {
|
||
// Job einreichen
|
||
const response = await fetch('/api/analyze', {
|
||
method: 'POST',
|
||
body: JSON.stringify(product)
|
||
});
|
||
|
||
const { jobId } = await response.json();
|
||
|
||
// Auf Event warten
|
||
return new Promise((resolve, reject) => {
|
||
this.pendingJobs.set(jobId, { resolve, reject });
|
||
|
||
// Timeout
|
||
setTimeout(() => {
|
||
if (this.pendingJobs.has(jobId)) {
|
||
this.pendingJobs.delete(jobId);
|
||
reject(new Error('Analysis timeout'));
|
||
}
|
||
}, 60000); // 60s Timeout
|
||
});
|
||
}
|
||
|
||
onProgress?: (jobId: string, status: string) => void;
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Event Sourcing für AI Decisions
|
||
|
||
```typescript
|
||
// src/events/event-store.ts
|
||
interface AIDecisionEvent {
|
||
eventId: string;
|
||
aggregateId: string; // z.B. analysisId
|
||
type: string;
|
||
data: any;
|
||
timestamp: Date;
|
||
version: number;
|
||
}
|
||
|
||
class AIEventStore {
|
||
private redis: Redis;
|
||
|
||
constructor() {
|
||
this.redis = new Redis(process.env.REDIS_URL!);
|
||
}
|
||
|
||
async append(aggregateId: string, events: AIDecisionEvent[]) {
|
||
const key = `aggregate:${aggregateId}`;
|
||
|
||
for (const event of events) {
|
||
await this.redis.xadd(
|
||
key,
|
||
'*',
|
||
'data', JSON.stringify(event)
|
||
);
|
||
}
|
||
}
|
||
|
||
async getEvents(aggregateId: string): Promise<AIDecisionEvent[]> {
|
||
const key = `aggregate:${aggregateId}`;
|
||
const entries = await this.redis.xrange(key, '-', '+');
|
||
|
||
return entries.map(([id, fields]) => {
|
||
const data = JSON.parse(fields[1]);
|
||
return { ...data, eventId: id };
|
||
});
|
||
}
|
||
|
||
// Reconstruct State from Events
|
||
async reconstruct<T>(
|
||
aggregateId: string,
|
||
reducer: (state: T, event: AIDecisionEvent) => T,
|
||
initialState: T
|
||
): Promise<T> {
|
||
const events = await this.getEvents(aggregateId);
|
||
return events.reduce(reducer, initialState);
|
||
}
|
||
}
|
||
|
||
// Beispiel: AI-Analyse mit vollständiger History
|
||
const eventStore = new AIEventStore();
|
||
|
||
// Events speichern
|
||
await eventStore.append('analysis-123', [
|
||
{ type: 'analysis.created', data: { input: '...' }, ... },
|
||
{ type: 'llm.called', data: { model: 'claude-3-haiku', tokens: 500 }, ... },
|
||
{ type: 'analysis.completed', data: { result: '...' }, ... }
|
||
]);
|
||
|
||
// State rekonstruieren
|
||
const analysisState = await eventStore.reconstruct(
|
||
'analysis-123',
|
||
(state, event) => {
|
||
switch (event.type) {
|
||
case 'analysis.created':
|
||
return { ...state, input: event.data.input, status: 'pending' };
|
||
case 'analysis.completed':
|
||
return { ...state, result: event.data.result, status: 'completed' };
|
||
default:
|
||
return state;
|
||
}
|
||
},
|
||
{ input: null, result: null, status: 'unknown' }
|
||
);
|
||
```
|
||
|
||
---
|
||
|
||
## Scaling Pattern
|
||
|
||
```yaml
|
||
# docker-compose.yml für skalierbare AI Worker
|
||
version: '3.8'
|
||
services:
|
||
redis:
|
||
image: redis:7-alpine
|
||
ports:
|
||
- "6379:6379"
|
||
|
||
api:
|
||
build: .
|
||
environment:
|
||
- REDIS_URL=redis://redis:6379
|
||
ports:
|
||
- "3000:3000"
|
||
|
||
ai-worker:
|
||
build: .
|
||
command: npm run worker
|
||
environment:
|
||
- REDIS_URL=redis://redis:6379
|
||
- CONCURRENCY=5
|
||
deploy:
|
||
replicas: 3 # 3 Worker-Instanzen
|
||
|
||
scheduler:
|
||
build: .
|
||
command: npm run scheduler
|
||
environment:
|
||
- REDIS_URL=redis://redis:6379
|
||
```
|
||
|
||
---
|
||
|
||
## Monitoring
|
||
|
||
```typescript
|
||
// src/monitoring/queue-metrics.ts
|
||
import { aiQueue } from '../queues/ai-queue';
|
||
|
||
async function getQueueMetrics() {
|
||
const [waiting, active, completed, failed] = await Promise.all([
|
||
aiQueue.getWaitingCount(),
|
||
aiQueue.getActiveCount(),
|
||
aiQueue.getCompletedCount(),
|
||
aiQueue.getFailedCount()
|
||
]);
|
||
|
||
return {
|
||
waiting,
|
||
active,
|
||
completed,
|
||
failed,
|
||
throughput: completed / (Date.now() / 1000 / 60) // per minute
|
||
};
|
||
}
|
||
|
||
// Prometheus Metrics
|
||
import { Counter, Gauge, Histogram } from 'prom-client';
|
||
|
||
const jobsProcessed = new Counter({
|
||
name: 'ai_jobs_processed_total',
|
||
help: 'Total AI jobs processed',
|
||
labelNames: ['type', 'status']
|
||
});
|
||
|
||
const jobDuration = new Histogram({
|
||
name: 'ai_job_duration_seconds',
|
||
help: 'AI job processing duration',
|
||
labelNames: ['type'],
|
||
buckets: [0.5, 1, 2, 5, 10, 30, 60]
|
||
});
|
||
|
||
const queueDepth = new Gauge({
|
||
name: 'ai_queue_depth',
|
||
help: 'Current queue depth',
|
||
labelNames: ['queue']
|
||
});
|
||
```
|
||
|
||
---
|
||
|
||
## Fazit
|
||
|
||
Event-Driven Architecture für AI-Systeme bietet:
|
||
|
||
1. **Resilienz**: Automatische Retries, keine Timeouts
|
||
2. **Skalierbarkeit**: Horizontale Worker-Skalierung
|
||
3. **Transparenz**: Event Sourcing für Audit-Trail
|
||
4. **Real-Time Updates**: WebSocket-Events an Clients
|
||
|
||
Für produktive AI-Anwendungen ist EDA nicht optional – es ist die Grundlage für zuverlässige Systeme.
|
||
|
||
---
|
||
|
||
## Bildprompts
|
||
|
||
1. "Event flow diagram with AI processing nodes, message queue visualization, technical architecture"
|
||
2. "Multiple workers processing AI tasks from central queue, assembly line concept"
|
||
3. "Event stream timeline with AI decision points highlighted, data visualization style"
|
||
|
||
---
|
||
|
||
## Quellen
|
||
|
||
- [BullMQ Documentation](https://docs.bullmq.io/)
|
||
- [Redis Streams](https://redis.io/docs/data-types/streams/)
|
||
- [Event-Driven Architecture Patterns](https://microservices.io/patterns/data/event-driven-architecture.html)
|
||
- [Martin Fowler: Event Sourcing](https://martinfowler.com/eaaDev/EventSourcing.html)
|