KI in bestehende Software integrieren: Der praktische Guide
Vergiss den Rewrite — bau eine Brücke
Jede Woche hören wir das Gleiche: „Wir wollen KI, also schreiben wir das System am besten komplett neu." Das ist der teuerste Fehler, den du machen kannst. Dein bestehendes System hat echte Nutzer, echte Daten und echte Verträge. Ein Big-Bang-Rewrite riskiert all das — und dauert dreimal so lange wie geplant.
Was tatsächlich funktioniert: KI als eigenständigen Service Layer daneben stellen. Dein bestehendes System ruft eine neue API auf. Fertig. Keine Migration, kein Rewrite, kein Risiko.
Wir haben das bei SecretStack in über einem Dutzend Projekten genau so umgesetzt — von B2B-SaaS über E-Commerce bis zu internen Enterprise-Tools. Dieser Guide zeigt dir exakt, wie das technisch aussieht.
Architektur: KI als Service Layer
Das zentrale Pattern ist simpel. Dein bestehendes System kennt die KI gar nicht — es kennt nur eine API:
┌──────────────────────────────────────────────────────────────────┐
│ KI ALS SERVICE LAYER │
│ │
│ ┌──────────────┐ ┌───────────────────┐ ┌────────────┐ │
│ │ Bestehendes │────▶│ KI Service Layer │────▶│ LLM API │ │
│ │ System │ │ (FastAPI/Next) │ │ (Claude, │ │
│ │ │◀────│ │◀────│ GPT-4) │ │
│ └──────────────┘ └─────────┬─────────┘ └────────────┘ │
│ │ │ │
│ │ ┌──────┴──────┐ │
│ │ │ │ │
│ │ ┌──────▼──┐ ┌──────▼──────┐ │
│ │ │ Cache │ │ Vector DB │ │
│ │ │ (Redis) │ │ (pgvector) │ │
│ │ └─────────┘ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ PostgreSQL │ ← Bestehende DB bleibt unverändert │
│ │ (Bestand) │ │
│ └──────────────┘ │
└──────────────────────────────────────────────────────────────────┘
Der KI Service Layer ist ein eigener Microservice — bei uns typischerweise FastAPI (Python) für ML-heavy Tasks oder Next.js API Routes (TypeScript) für leichtere Integrationen. Er hat drei Jobs:
- Anfragen annehmen — vom bestehenden System über REST oder Webhooks
- KI orchestrieren — Prompts bauen, Kontext laden, LLM aufrufen, Ergebnis validieren
- Graceful Degradation — wenn die KI nicht liefert, fällt das System auf den manuellen Workflow zurück
RAG-Pipeline: Dein eigenes Wissen nutzbar machen
Der häufigste Use Case: du willst, dass die KI über deine Daten Bescheid weiß. Produktkatalog, Support-Dokumentation, interne Wissensbasis. Dafür brauchst du RAG — Retrieval Augmented Generation.
So sieht der Datenfluss aus:
┌─────────────────────────────────────────────────────────────────┐
│ RAG PIPELINE │
│ │
│ INGESTION (einmalig / periodisch): │
│ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Dokumente │──▶│ Chunking │──▶│ Embedding │──▶│ pgvector │ │
│ │ PDF, MD, │ │ 512 Token │ │ text- │ │ HNSW │ │
│ │ HTML, DB │ │ Overlap │ │ embedding- │ │ Index │ │
│ └──────────┘ │ 50 Token │ │ 3-large │ └───────────┘ │
│ └──────────┘ └───────────┘ │
│ │
│ QUERY (pro Anfrage): │
│ ┌──────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ User │──▶│ Query │──▶│ Semantic │──▶│ LLM + │ │
│ │ Frage │ │ Embedding │ │ Search │ │ Kontext │ │
│ │ │ │ │ │ Top-5 │ │ = Antwort │ │
│ └──────────┘ └───────────┘ └───────────┘ └───────────┘ │
└─────────────────────────────────────────────────────────────────┘
Schritt 1: Vector-Tabelle anlegen
Zuerst brauchst du die Infrastruktur. pgvector ist eine PostgreSQL-Extension — du brauchst keine separate Vektordatenbank. Das läuft in deiner bestehenden Postgres-Instanz:
-- pgvector Extension aktivieren
CREATE EXTENSION IF NOT EXISTS vector;
-- Tabelle für Dokument-Chunks mit Embeddings
CREATE TABLE document_embeddings (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
content TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
source_id TEXT NOT NULL,
source_type TEXT NOT NULL, -- 'support_doc', 'product', 'faq'
embedding vector(3072), -- OpenAI text-embedding-3-large
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now()
);
-- HNSW-Index für schnelle Nearest-Neighbor-Suche
-- m=16 und ef_construction=128 sind gute Defaults bis ~1M Vektoren
CREATE INDEX idx_embeddings_hnsw
ON document_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 128);
-- Partial Index für Source-Type-Filterung
CREATE INDEX idx_embeddings_source_type
ON document_embeddings (source_type);
-- Nützlich für Bulk-Updates: schnell alle Chunks einer Quelle finden
CREATE INDEX idx_embeddings_source_id
ON document_embeddings (source_id);
Schritt 2: Embedding-Erstellung und Suche (TypeScript)
Die komplette RAG-Pipeline in TypeScript — von Dokument-Ingestion bis semantischer Suche:
// lib/rag-pipeline.ts
import OpenAI from "openai";
import { Pool } from "pg";
const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
// --- Embedding-Erstellung ---
interface DocumentChunk {
content: string;
sourceId: string;
sourceType: "support_doc" | "product" | "faq";
metadata?: Record<string, unknown>;
}
async function createEmbedding(text: string): Promise<number[]> {
const response = await openai.embeddings.create({
model: "text-embedding-3-large",
input: text,
dimensions: 3072,
});
return response.data[0].embedding;
}
export async function ingestDocuments(chunks: DocumentChunk[]): Promise<void> {
// Batch-Embedding: max 2048 Inputs pro API-Call
const batchSize = 100;
for (let i = 0; i < chunks.length; i += batchSize) {
const batch = chunks.slice(i, i + batchSize);
const response = await openai.embeddings.create({
model: "text-embedding-3-large",
input: batch.map((c) => c.content),
dimensions: 3072,
});
const query = `
INSERT INTO document_embeddings (content, source_id, source_type, metadata, embedding)
VALUES ($1, $2, $3, $4, $5::vector)
ON CONFLICT (id) DO NOTHING
`;
for (let j = 0; j < batch.length; j++) {
await pool.query(query, [
batch[j].content,
batch[j].sourceId,
batch[j].sourceType,
JSON.stringify(batch[j].metadata ?? {}),
`[${response.data[j].embedding.join(",")}]`,
]);
}
}
}
// --- Semantische Suche ---
interface SearchResult {
content: string;
sourceId: string;
sourceType: string;
similarity: number;
metadata: Record<string, unknown>;
}
export async function semanticSearch(
query: string,
options: {
topK?: number;
sourceType?: string;
similarityThreshold?: number;
} = {}
): Promise<SearchResult[]> {
const { topK = 5, sourceType, similarityThreshold = 0.7 } = options;
const queryEmbedding = await createEmbedding(query);
const sql = `
SELECT
content,
source_id,
source_type,
metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM document_embeddings
WHERE 1 - (embedding <=> $1::vector) > $2
${sourceType ? "AND source_type = $4" : ""}
ORDER BY embedding <=> $1::vector
LIMIT $3
`;
const params: (string | number)[] = [
`[${queryEmbedding.join(",")}]`,
similarityThreshold,
topK,
];
if (sourceType) params.push(sourceType);
const { rows } = await pool.query(sql, params);
return rows.map((row) => ({
content: row.content,
sourceId: row.source_id,
sourceType: row.source_type,
similarity: parseFloat(row.similarity),
metadata: row.metadata,
}));
}
KI-Service Endpoint mit Streaming (Python/FastAPI)
Für ML-heavy Tasks setzen wir auf FastAPI. Hier ein produktionsreifer Endpoint mit Streaming, strukturiertem Output-Parsing und Fallback:
# ai_service/main.py
import json
import hashlib
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from anthropic import Anthropic, APIError
import redis.asyncio as redis
app = FastAPI()
client = Anthropic()
cache = redis.from_url("redis://localhost:6379/0")
TICKET_CLASSIFICATION_PROMPT = """Du bist ein Support-Ticket-Klassifikator.
Analysiere das folgende Support-Ticket und antworte AUSSCHLIESSLICH
mit validem JSON in diesem Format:
{{
"kategorie": "billing" | "technical" | "feature_request" | "complaint" | "other",
"prioritaet": "low" | "medium" | "high" | "critical",
"zusammenfassung": "Max 2 Sätze",
"antwortvorschlag": "Vorgeschlagene Antwort an den Kunden",
"confidence": 0.0 - 1.0
}}
Ticket:
{ticket_text}
Kundenhistorie (letzte 5 Tickets):
{customer_history}
"""
class TicketRequest(BaseModel):
ticket_text: str
customer_id: str
customer_history: list[str] = []
class TicketClassification(BaseModel):
kategorie: str
prioritaet: str
zusammenfassung: str
antwortvorschlag: str
confidence: float
@app.post("/api/classify-ticket")
async def classify_ticket(req: TicketRequest) -> TicketClassification:
# Cache-Key aus Ticket-Text generieren
cache_key = f"ticket:{hashlib.sha256(req.ticket_text.encode()).hexdigest()[:16]}"
cached = await cache.get(cache_key)
if cached:
return TicketClassification(**json.loads(cached))
prompt = TICKET_CLASSIFICATION_PROMPT.format(
ticket_text=req.ticket_text,
customer_history="\n".join(req.customer_history[-5:]) or "Keine Historie"
)
try:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
temperature=0,
messages=[{"role": "user", "content": prompt}],
)
raw = response.content[0].text
parsed = json.loads(raw)
result = TicketClassification(**parsed)
# 1 Stunde cachen
await cache.set(cache_key, json.dumps(parsed), ex=3600)
return result
except (json.JSONDecodeError, KeyError) as e:
# LLM hat kein valides JSON geliefert → Fallback
return TicketClassification(
kategorie="other",
prioritaet="medium",
zusammenfassung="Automatische Klassifikation fehlgeschlagen",
antwortvorschlag="",
confidence=0.0,
)
except APIError as e:
raise HTTPException(status_code=502, detail=f"LLM API error: {e.message}")
@app.post("/api/generate-stream")
async def generate_stream(req: TicketRequest):
"""Streaming-Endpoint für lange Antworten — z.B. ausführliche Kundenantworten."""
prompt = TICKET_CLASSIFICATION_PROMPT.format(
ticket_text=req.ticket_text,
customer_history="\n".join(req.customer_history[-5:]) or "Keine Historie"
)
async def event_stream():
try:
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
) as stream:
for text in stream.text_stream:
yield f"data: {json.dumps({'text': text})}\n\n"
yield "data: [DONE]\n\n"
except APIError:
yield f"data: {json.dumps({'error': 'LLM nicht erreichbar'})}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
Webhook-basierte Async-Verarbeitung
Nicht jeder KI-Task muss synchron laufen. Für alles, was länger als ~2 Sekunden dauert, ist asynchrone Verarbeitung besser. Der User wartet nicht — er bekommt das Ergebnis, sobald es fertig ist:
// api/webhooks/ticket-created.ts (Next.js API Route)
import { NextRequest, NextResponse } from "next/server";
import { Queue } from "bullmq";
import { createHmac } from "crypto";
const aiQueue = new Queue("ai-processing", {
connection: { host: "localhost", port: 6379 },
});
// Webhook-Endpoint: wird vom bestehenden System aufgerufen
export async function POST(req: NextRequest) {
// Webhook-Signatur verifizieren
const signature = req.headers.get("x-webhook-signature");
const body = await req.text();
const expected = createHmac("sha256", process.env.WEBHOOK_SECRET!)
.update(body)
.digest("hex");
if (signature !== expected) {
return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
}
const payload = JSON.parse(body);
// Job in die Queue — wird async vom Worker verarbeitet
await aiQueue.add(
"classify-ticket",
{
ticketId: payload.ticket_id,
ticketText: payload.text,
customerId: payload.customer_id,
callbackUrl: payload.callback_url,
},
{
attempts: 3,
backoff: { type: "exponential", delay: 2000 },
removeOnComplete: 1000,
removeOnFail: 5000,
}
);
return NextResponse.json({ status: "queued" }, { status: 202 });
}
// --- Worker (separater Prozess) ---
// workers/ai-worker.ts
import { Worker } from "bullmq";
const worker = new Worker(
"ai-processing",
async (job) => {
const { ticketId, ticketText, customerId, callbackUrl } = job.data;
// KI Service aufrufen
const classification = await fetch(
`${process.env.AI_SERVICE_URL}/api/classify-ticket`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
ticket_text: ticketText,
customer_id: customerId,
}),
}
);
const result = await classification.json();
// Ergebnis an bestehendes System zurücksenden
await fetch(callbackUrl, {
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${process.env.SYSTEM_API_KEY}`,
},
body: JSON.stringify({
ticket_id: ticketId,
classification: result,
processed_at: new Date().toISOString(),
}),
});
},
{ connection: { host: "localhost", port: 6379 }, concurrency: 10 }
);
worker.on("failed", (job, err) => {
console.error(`Job ${job?.id} failed after ${job?.attemptsMade} attempts:`, err);
});
Caching-Layer für LLM-Responses
LLM-Aufrufe sind teuer und langsam. Intelligentes Caching spart beides. Wir setzen auf einen zweistufigen Cache — exakte Treffer und semantische Treffer:
// lib/llm-cache.ts
import Redis from "ioredis";
import { createHash } from "crypto";
import { createEmbedding, semanticSearch } from "./rag-pipeline";
const redis = new Redis(process.env.REDIS_URL!);
interface CacheOptions {
ttl?: number; // TTL in Sekunden (Default: 1 Stunde)
semanticMatch?: boolean; // Auch ähnliche Queries matchen?
similarityThreshold?: number;
}
export async function cachedLlmCall<T>(
prompt: string,
llmCall: () => Promise<T>,
options: CacheOptions = {}
): Promise<T & { _cache: "hit" | "semantic_hit" | "miss" }> {
const { ttl = 3600, semanticMatch = true, similarityThreshold = 0.95 } = options;
// 1. Exakter Cache-Treffer
const exactKey = `llm:exact:${createHash("sha256").update(prompt).digest("hex").slice(0, 24)}`;
const exactHit = await redis.get(exactKey);
if (exactHit) {
return { ...JSON.parse(exactHit), _cache: "hit" as const };
}
// 2. Semantischer Cache-Treffer (optional)
if (semanticMatch) {
const similar = await semanticSearch(prompt, {
topK: 1,
sourceType: "llm_cache",
similarityThreshold,
});
if (similar.length > 0) {
const semanticKey = `llm:result:${similar[0].sourceId}`;
const semanticHit = await redis.get(semanticKey);
if (semanticHit) {
return { ...JSON.parse(semanticHit), _cache: "semantic_hit" as const };
}
}
}
// 3. Cache Miss — LLM aufrufen
const result = await llmCall();
// Ergebnis cachen
await redis.set(exactKey, JSON.stringify(result), "EX", ttl);
return { ...result, _cache: "miss" as const };
}
// --- Verwendung ---
const classification = await cachedLlmCall(
`Klassifiziere: ${ticketText}`,
() => classifyTicket(ticketText),
{ ttl: 7200, semanticMatch: true, similarityThreshold: 0.93 }
);
if (classification._cache !== "miss") {
console.log(`Cache ${classification._cache} — kein LLM-Call nötig`);
}
Performance-Zahlen aus der Praxis
Hier sind reale Messungen aus einem unserer Projekte — ein B2B-SaaS mit ~50.000 Support-Tickets pro Monat:
Latenz-Vergleich
| Methode | p50 Latenz | p99 Latenz | Kosten pro Call | |---|---|---|---| | Direkt (Claude API, synchron) | 1.800ms | 4.200ms | ~$0,008 | | Mit Redis-Cache (Hit) | 3ms | 12ms | $0,000 | | Mit Streaming (Time to first token) | 280ms | 650ms | ~$0,008 | | Async + Queue (aus User-Sicht) | 0ms* | 0ms* | ~$0,008 |
*User wartet nicht — Ergebnis kommt asynchron.
Token-Optimierung
Durch gezieltes Prompt Engineering konnten wir den Token-Verbrauch drastisch senken:
| Metrik | Vor Optimierung | Nach Optimierung | Einsparung | |---|---|---|---| | Input Tokens / Ticket | ~2.400 | ~850 | -65% | | Output Tokens / Ticket | ~600 | ~280 | -53% | | Monatliche API-Kosten (50k Tickets) | ~$4.800 | ~$1.650 | -66% | | Cache Hit Rate | 0% | 34% | — | | Effektive Kosten nach Cache | ~$4.800 | ~$1.090 | -77% |
Die größten Hebel: strukturierte Prompts statt freier Text, Few-Shot-Examples entfernt und durch klare Instruktionen ersetzt, Kundenhistorie auf die letzten 5 Tickets begrenzt statt alle zu senden.
Praxis-Beispiel: Support-Ticket-Workflow transformieren
Vorher: Manueller Workflow
- Ticket kommt rein (E-Mail, Formular, Chat)
- Support-Mitarbeiter liest das Ticket
- Mitarbeiter entscheidet manuell: Kategorie, Priorität, Zuständigkeit
- Mitarbeiter schreibt Antwort von Grund auf
- Durchschnittliche Bearbeitungszeit: 12 Minuten pro Ticket
Nachher: KI-gestützter Workflow
- Ticket kommt rein
- Webhook triggert automatische KI-Klassifikation
- Mitarbeiter sieht: Kategorie, Priorität, Zusammenfassung, vorgeschlagene Antwort
- Mitarbeiter prüft, passt ggf. an, sendet ab
- Durchschnittliche Bearbeitungszeit: 3 Minuten pro Ticket
Die Integration ins bestehende System sieht so aus — ein einziger API-Call reicht:
// Im bestehenden Ticket-System: neuen Ticket-Handler erweitern
import type { Ticket, TicketClassification } from "@/types";
async function onTicketCreated(ticket: Ticket): Promise<void> {
// Bestehende Logik bleibt komplett erhalten
await saveToDatabase(ticket);
await notifyTeam(ticket);
// NEU: KI-Klassifikation anstoßen (async, blockiert nichts)
try {
const response = await fetch(`${process.env.AI_SERVICE_URL}/api/classify-ticket`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
ticket_text: ticket.subject + "\n\n" + ticket.body,
customer_id: ticket.customerId,
customer_history: ticket.customerHistory ?? [],
}),
signal: AbortSignal.timeout(10_000), // 10s Timeout
});
if (response.ok) {
const classification: TicketClassification = await response.json();
// Nur anreichern, nicht überschreiben
await updateTicket(ticket.id, {
ai_category: classification.kategorie,
ai_priority: classification.prioritaet,
ai_summary: classification.zusammenfassung,
ai_suggested_reply: classification.antwortvorschlag,
ai_confidence: classification.confidence,
ai_classified_at: new Date(),
});
}
} catch (error) {
// KI nicht erreichbar? Kein Problem — Ticket funktioniert wie vorher
console.warn("AI classification failed, continuing without:", error);
}
}
Der entscheidende Punkt: das try/catch um den KI-Aufruf. Wenn die KI nicht erreichbar ist oder Unsinn liefert, funktioniert das Ticket-System exakt wie vorher. KI ist eine Bereicherung, kein Single Point of Failure.
Checkliste: Ist dein System bereit für KI?
Bevor du loslegst, prüfe diese 8 Punkte. Jeder „Nein"-Punkt ist kein Dealbreaker — aber du solltest wissen, was auf dich zukommt:
1. Hat dein System eine API? Dein bestehendes System muss Daten rein- und rauslassen können. REST, GraphQL, Webhooks — egal was, aber es muss eine Schnittstelle geben. Ohne API brauchst du erst eine.
2. Wie ist die Datenqualität? KI ist nur so gut wie die Daten, die du reinfütterst. Wenn deine Support-Tickets aus Einzeilern bestehen oder deine Produktbeschreibungen leer sind, wird die KI wenig liefern. Prüfe: Sind die relevanten Datenfelder befüllt, konsistent und aktuell?
3. Wie hoch ist dein Latenz-Budget? Ein LLM-API-Call dauert 1-4 Sekunden. Wenn dein User-Flow unter 200ms bleiben muss, brauchst du async Processing oder Caching. Wo im Flow kann der User warten, wo nicht?
4. Wie viele Requests erwartest du? 100 Requests pro Tag sind trivial. 100.000 pro Tag erfordern Caching, Queuing und Rate-Limit-Management. Rechne die API-Kosten bei voller Last durch.
5. DSGVO und Datenschutz geklärt? Personenbezogene Daten an externe APIs senden ist heikel. Prüfe: Welche Daten gehen raus? Kann anonymisiert/pseudonymisiert werden? Brauchst du einen Auftragsverarbeitungsvertrag (AVV) mit dem LLM-Anbieter? Reicht ein EU-Hosting?
6. Gibt es klare Erfolgskriterien? „Die KI soll besser sein als vorher" ist kein Kriterium. Definiere messbare Ziele: Bearbeitungszeit pro Ticket sinkt von 12 auf 5 Minuten. Klassifikationsgenauigkeit liegt bei mindestens 85%. Kosten pro verarbeitetem Dokument unter 0,02 EUR.
7. Wer pflegt die Prompts? Prompts sind Code — sie brauchen Versionierung, Testing und Iteration. Kläre vorab: Wer ist zuständig? Wie werden Prompt-Änderungen getestet, bevor sie live gehen? Hast du ein Evaluation-Framework?
8. Gibt es einen manuellen Fallback? Was passiert, wenn die KI ausfällt? Wenn der LLM-Anbieter eine Störung hat? Wenn das Ergebnis offensichtlich falsch ist? Dein System muss ohne KI genauso funktionieren wie vorher — KI-Features sind eine Verbesserung, keine Abhängigkeit.
Die wichtigsten Lektionen
Nach über einem Dutzend KI-Integrationen haben sich drei Prinzipien herauskristallisiert:
Starte mit einem einzigen Use Case. Nicht drei gleichzeitig. Nimm den Use Case mit dem klarsten ROI — meistens dort, wo heute am meisten manuelle, repetitive Arbeit anfällt.
Bau den Fallback zuerst. Bevor du den Happy Path baust, implementiere den Fehlerfall. Was passiert, wenn die KI nicht antwortet? Wenn sie Unsinn liefert? Wenn sie zu langsam ist? Erst wenn der Fallback steht, bau den KI-Pfad.
Miss alles von Tag 1. Latenz, Token-Verbrauch, Cache Hit Rate, Kosten pro Request, Nutzerzufriedenheit. Ohne Daten kannst du nicht optimieren und nicht beweisen, dass die Integration einen Mehrwert liefert.
Nächster Schritt
Wenn du eine bestehende Anwendung hast und KI-Features einbauen willst — ohne Rewrite, ohne Risiko — dann lass uns reden. Im kostenlosen Discovery Call analysieren wir dein bestehendes System, identifizieren den besten Einstiegspunkt und zeigen dir, wie die Integration konkret aussieht.
Kein Slide-Deck, kein Buzzword-Bingo. Nur ein technisches Gespräch zwischen Engineers.