Written with AI assistance from the source video. The video is the authority.
Log in to leave a comment
No posts yet
Jedes Mal, wenn ein neues Modell erscheint, verzögert sich die Einführung nicht wegen der Benchmark-Ergebnisse. Es sind die Legacy-Strukturen, in denen das SDK eines bestimmten Anbieters fest im Geschäftslogik-Kern verdrahtet ist, sowie die unvorhersehbaren API-Rechnungen, die einem im Weg stehen.
Fügt man dem noch den Einwand des Sicherheitsteams hinzu, dass „Kundendaten, die in ausländische Regionen abfließen, gegen Datenschutzgesetze verstoßen“, verbringen Ingenieure jede Nacht mit Überstunden und dem Korrigieren von Prüfberichten. Das Ändern des Endpunkts ist der allerletzte Schritt. Es ist sicherer, zunächst anhand echter Betriebslogs Zahlen zu ermitteln, um die Budgetfreigabe zu erhalten, das Audit durch Prompt-Maskierung zu bestehen und erst dann den Code auszutauschen.
Wenn man einen Haushaltsplan allein auf Basis der Preistabelle des Anbieters einreicht, droht bei Traffic-Spitzen eine saftige Rechnung. Zunächst müssen die Aufruf-Logs der letzten 30 Tage des laufenden Servers geparst werden, um den tatsächlichen Token-Verbrauch zu ermitteln.
Wer CloudWatch Logs Insights nutzt, kann mit einer AWS-CLI-Query innerhalb von 10 Sekunden den Token-Verbrauch pro Aufruf sowie das tägliche Gesamtvolumen ermitteln.
`bash
#!/usr/bin/env bash
set -euo pipefail
LOG_GROUP_NAME="/aws/backend/legacy-llm-service"
START_TIME=(date +%s)
QUERY_STRING='fields @timestamp, usage.prompt_tokens as p_tok, usage.completion_tokens as c_tok
| filter ispresent(p_tok) and ispresent(c_tok)
| stats count(@timestamp) as total_requests,
sum(p_tok) as sum_prompt_tokens,
sum(c_tok) as sum_completion_tokens,
avg(p_tok) as avg_prompt_tokens,
avg(c_tok) as avg_completion_tokens,
percentile(p_tok + c_tok, 95) as p95_total_tokens'
QUERY_ID=$(aws logs start-query
--log-group-name "$LOG_GROUP_NAME"
--start-time "$START_TIME"
--end-time "$END_TIME"
--query-string "$QUERY_STRING"
--output text --query 'queryId')
sleep 5
aws logs get-query-results --query-id "$QUERY_ID"
--output json | jq -r '.results[] | map({(.field): .value}) | add'
`
Das Preismodell von GPT-6 Astra liegt einheitlich bei 10 US-Dollar pro 10 Millionen Tokens (1,00 USD pro 1 Million Tokens), ohne Unterscheidung zwischen Eingabe und Ausgabe. Die Formel ist einfach:
C_{ ext{monthly}} ( ext{USD}) = rac{T_{ ext{total}}}{1.000.000} imes 1,00Durch das Einsetzen der aus den Logs extrahierten Werte in die folgende Tabellenstruktur entsteht eine vorzeigbare Tabelle für den Bericht.
| Szenario | Tägliche Anfragen () | Summe Tokens pro Anfrage () | Gesamte monatliche Tokens () | Astra-Stückpreis (1M Tokens) | Voraussichtliche monatliche Kosten (USD) | Umgerechneter Betrag (Basis 1.350 KRW) |
|---|---|---|---|---|---|---|
| Konservativ (Low) | 35.000 | 1.200 | 1.260.000.000 | $1,00 | $1.260,00 | 1.701.000 KRW |
| Standard (Base) | 50.000 | 1.200 | 1.800.000.000 | $1,00 | $1.800,00 | 2.430.000 KRW |
| Spitze (Peak) | 75.000 | 1.500 | 3.375.000.000 | $1,00 | $3.375,00 | 4.556.250 KRW |
Beim Erstellen der Tabellenkalkulation trägt man in B1 die täglichen Aufrufe (50000), in B2 die durchschnittlichen Eingabe-Tokens (800), in B3 die durchschnittlichen Ausgabe-Tokens (400) und in B4 den Wechselkurs (1350) ein. In Zelle B5 fügt man =30*B1*(B2+B3) ein, in B6 =(B5/10000000)*10 und in B7 =B6*B4. Verglichen mit dem bisherigen Anbieter, der 5 US-Dollar pro 1M Tokens berechnete, sinken die Kosten um 80 %. Im Standardszenario werden monatlich 7.200 US-Dollar (ca. 9,72 Mio. KRW) eingespart, womit kein Grund mehr für eine Ablehnung im Budgetfreigabe-Prozess besteht.
Strukturen, bei denen das SDK eines bestimmten Modells direkt innerhalb der Geschäftslogik aufgerufen wird, sollten vermieden werden. Dies führt dazu, dass vom Anbieter definierte Fehlertypen auf das gesamte System übergreifen. Wenn man das Interface vorab mit Pythons typing.Protocol festlegt und die Implementierung in einen Adapter kapselt, muss der Aufrufcode nicht angerührt werden.
`python
from typing import Protocol, List, Optional
from pydantic import BaseModel, Field
class LLMMessage(BaseModel):
role: str = Field(..., description="system, user, assistant")
content: str
class LLMUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class LLMResponse(BaseModel):
content: str
usage: LLMUsage
model_name: str
provider: str
class LLMClientAdapter(Protocol):
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
...
`
Der bestehende Fable-Endpunkt und der neue Astra-Endpunkt werden jeweils in eigene Adapter-Klassen gekapselt. Die Laufzeit-Umschaltung erfolgt durch die LLMFactory über eine einzige Umgebungsvariable.
`python
import httpx
import os
from typing import List, Optional
from core.llm_protocol import LLMClientAdapter, LLMMessage, LLMResponse, LLMUsage
class AstraAdapter:
def init(self, api_key: str, base_url: str = "https://api.astra.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=httpx.Timeout(connect=5.0, read=90.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
payload = {
"model": "gpt-6-astra",
"messages": [msg.model_dump() for msg in messages],
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
resp = await self.client.post("/chat/completions", json=payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["choices"][0]["message"]["content"],
usage=LLMUsage(
prompt_tokens=data["usage"]["prompt_tokens"],
completion_tokens=data["usage"]["completion_tokens"],
total_tokens=data["usage"]["total_tokens"]
),
model_name=data["model"],
provider="astra"
)
class FableAdapter:
def init(self, api_key: str, endpoint_url: str):
self.api_key = api_key
self.endpoint_url = endpoint_url
self.client = httpx.AsyncClient(
headers={"X-Fable-Key": self.api_key},
timeout=httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)
)
async def generate_completion(
self,
messages: List[LLMMessage],
temperature: float = 0.2,
max_tokens: Optional[int] = None
) -> LLMResponse:
fable_payload = {
"prompt_sequence": [{"speaker": m.role, "text": m.content} for m in messages],
"gen_params": {"temp": temperature, "limit": max_tokens or 1024}
}
resp = await self.client.post(self.endpoint_url, json=fable_payload)
resp.raise_for_status()
data = resp.json()
return LLMResponse(
content=data["result"]["generated_text"],
usage=LLMUsage(
prompt_tokens=data["meta"]["tokens_in"],
completion_tokens=data["meta"]["tokens_out"],
total_tokens=data["meta"]["tokens_in"] + data["meta"]["tokens_out"]
),
model_name="fable-legacy",
provider="fable"
)
class LLMFactory:
@staticmethod
def get_adapter() -> LLMClientAdapter:
provider = os.getenv("LLM_PROVIDER", "astra").lower()
if provider == "astra":
return AstraAdapter(
api_key=os.environ["ASTRA_API_KEY"],
base_url=os.getenv("ASTRA_BASE_URL", "https://api.astra.ai/v1")
)
elif provider == "fable":
return FableAdapter(
api_key=os.environ["FABLE_API_KEY"],
endpoint_url=os.environ["FABLE_ENDPOINT_URL"]
)
raise ValueError(f"Unsupported LLM provider: {provider}")
`
Fehler durch beschädigte Antwortformate bei einem Modellwechsel sollten vorab durch Unit-Tests abgefangen werden. Der folgende Test validiert das Pydantic-Schema sowie Mock-Antworten.
`python
import pytest
from core.llm_protocol import LLMMessage, LLMResponse
from core.adapters import AstraAdapter
@pytest.mark.asyncio
async def test_astra_response_contract_compliance(monkeypatch):
mock_payload = {
"id": "chatcmpl-astra-001",
"model": "gpt-6-astra",
"choices": [
{"message": {"role": "assistant", "content": "Normaler Ergebnisswert"}}
],
"usage": {
"prompt_tokens": 120,
"completion_tokens": 45,
"total_tokens": 165
}
}
class MockResponse:
def raise_for_status(self): pass
def json(self): return mock_payload
async def mock_post(*args, **kwargs):
return MockResponse()
adapter = AstraAdapter(api_key="test-key")
monkeypatch.setattr(adapter.client, "post", mock_post)
messages = [LLMMessage(role="user", content="Test-Anfrage")]
result: LLMResponse = await adapter.generate_completion(messages)
assert isinstance(result, LLMResponse)
assert result.provider == "astra"
assert result.model_name == "gpt-6-astra"
assert result.content == "Normaler Ergebnisswert"
assert result.usage.prompt_tokens == 120
assert result.usage.completion_tokens == 45
assert result.usage.total_tokens == 165
`
Wenn die Aufrufziele im Einstiegsmodul der Geschäftslogik auf LLMFactory.get_adapter().generate_completion(...) gebündelt werden, dauert die Umstellung nicht einmal eine Stunde. Sollte das neue Modell ausfallen, genügt ein Zurücksetzen der Umgebungsvariable auf LLM_PROVIDER=fable.
Die in den Werbematerialien der Anbieter angegebenen Zahlen zur Reduzierung von Halluzinationen müssen mit unternehmensinternen Daten verifiziert werden. Dazu werden 50 Datensätze von Rückerstattungsrichtlinien oder Zahlungsbedingungen als Gold-Dataset zusammengefasst und mit der Open-Source-Bibliothek DeepEval verglichen.
`python
import json
import asyncio
from deepeval.test_case import LLMTestCase
from deepeval.metrics import HallucinationMetric
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage
async def run_batch_evaluation():
with open("eval_dataset_50.json", "r", encoding="utf-8") as f:
cases = json.load(f)
astra = AstraAdapter(api_key="astra-key")
fable = FableAdapter(api_key="fable-key", endpoint_url="https://api.fable.internal/v1")
hallucination_metric = HallucinationMetric(threshold=0.1)
astra_scores, fable_scores = [], []
for case in cases:
msg = [LLMMessage(role="user", content=case["input"])]
astra_res = await astra.generate_completion(msg)
tc_astra = LLMTestCase(input=case["input"], actual_output=astra_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_astra)
astra_scores.append(hallucination_metric.score)
fable_res = await fable.generate_completion(msg)
tc_fable = LLMTestCase(input=case["input"], actual_output=fable_res.content, retrieval_context=case["retrieval_context"])
hallucination_metric.measure(tc_fable)
fable_scores.append(hallucination_metric.score)
avg_fable = sum(fable_scores) / len(fable_scores)
avg_astra = sum(astra_scores) / len(astra_scores)
delta = ((avg_fable - avg_astra) / avg_fable) * 100
print(f"Empirical Hallucination Reduction: {delta:.2f}%")
if name == "main":
asyncio.run(run_batch_evaluation())
`
Die Messergebnisse werden mittels der Funktion scipy.stats.ttest_rel einem verbundenen t-Test (Paired t-test) unterzogen.
Erst wenn der p-Wert bei einer Stichprobengröße von 50 unter 0,05 sinkt, wird davon ausgegangen, dass die Halluzinationen tatsächlich zurückgegangen sind.
Ein im operativen Betrieb weitaus gefährlicheres Problem als Fehlercodes sind schleifende Antwortzeiten. Ohne Timeout läuft der Verbindungspool des Backends im Handumdrehen leer. Der folgende Circuit-Breaker-Code bricht den Aufruf bei 5 aufeinanderfolgenden Fehlern sofort ab und leitet auf das vorherige Modell um.
`python
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError
from core.adapters import AstraAdapter, FableAdapter
from core.llm_protocol import LLMMessage, LLMResponse
logger = logging.getLogger(name)
astra_breaker = CircuitBreaker(fail_max=5, reset_timeout=30)
class ResilientLLMClient:
def init(self, primary: AstraAdapter, fallback: FableAdapter):
self.primary = primary
self.fallback = fallback
async def execute(self, messages: list[LLMMessage]) -> LLMResponse:
try:
return await self._call_primary(messages)
except (CircuitBreakerError, Exception) as exc:
logger.warning("Primary degraded [%s]. Executing fallback.", str(exc))
return await self._call_fallback(messages)
async def _call_primary(self, messages: list[LLMMessage]) -> LLMResponse:
if astra_breaker.current_state == "open":
raise CircuitBreakerError("Circuit breaker OPEN")
try:
response = await self.primary.generate_completion(messages)
astra_breaker.success()
return response
except Exception as err:
astra_breaker.fail()
raise err
async def _call_fallback(self, messages: list[LLMMessage]) -> LLMResponse:
return await self.fallback.generate_completion(messages)
`
Wenn das Connect-Timeout 5 Sekunden und das Read-Timeout 90 Sekunden überschreitet, wird dies unverzüglich als Fehlschlag gewertet. Sobald sich 5 Fehlschläge häufen, öffnet sich der Circuit Breaker und blockiert externe Aufrufe für 30 Sekunden sofort (mit 0 Sekunden Verzögerung). Dies ist die minimale Schutzlinie, um zu verhindern, dass hängende Aufrufe den gesamten Server lahmlegen.
Wenn Benutzereingaben unmaskiert an eine externe API in einer ausländischen Region gesendet werden, verstößt dies gegen das Datenschutzgesetz (wie z. B. Bestimmungen zur internationalen Datenübermittlung). Die Maskierung muss im In-Memory-Schritt abgeschlossen sein, bevor die Daten das Netzwerk verlassen. Da einfache 13stellige reguläre Ausdrücke auch Rechnungs- oder Bestellnummern löschen, wird ein gewichteter Prüfsummenalgorithmus hinzugefügt.
ext{Checksum} = 11 - left( left( sum_{i=1}^{12} d_i imes w_i ight) mod 11 ight)`python
import re
from typing import List
from core.llm_protocol import LLMMessage
class PIIMaskingPipeline:
EMAIL_REGEX = re.compile(r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+.[a-zA-Z0-9-.]+')
INTERNAL_IP_REGEX = re.compile(
r'\b(?:10.\d{1,3}.\d{1,3}.\d{1,3}|'
r'172.(?:1[6-9]|2\d|3[0-1]).\d{1,3}.\d{1,3}|'
r'192.168.\d{1,3}.\d{1,3})\b'
)
RRN_CANDIDATE_REGEX = re.compile(r'\b(\d{6})[- ]?(\d{7})\b')
@classmethod
def is_valid_rrn(cls, front: str, back: str) -> bool:
full = front + back
if len(full) != 13 or not full.isdigit():
return False
weights = [2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5]
s = sum(int(full[i]) * weights[i] for i in range(12))
remainder = (11 - (s % 11)) % 10
return remainder == int(full[12])
@classmethod
def mask_rrn(cls, text: str) -> str:
def replace(match):
front, back = match.group(1), match.group(2)
if cls.is_valid_rrn(front, back):
return "[RESIDENT_ID_MASKED]"
return match.group(0)
return cls.RRN_CANDIDATE_REGEX.sub(replace, text)
@classmethod
def sanitize(cls, text: str) -> str:
text = cls.mask_rrn(text)
text = cls.EMAIL_REGEX.sub("[EMAIL_MASKED]", text)
text = cls.INTERNAL_IP_REGEX.sub("[INTERNAL_IP_MASKED]", text)
return text
@classmethod
def sanitize_messages(cls, messages: List[LLMMessage]) -> List[LLMMessage]:
return [
LLMMessage(role=m.role, content=cls.sanitize(m.content))
for m in messages
]
`
Es muss überprüft werden, ob der Vertrag mit dem Anbieter eine Zero-Data-Retention-Klausel (ZDR) enthält, die das Wiederverwenden von Eingabe-Prompts zum Training untersagt. In den CloudWatch-Logs wird das Speichern des Prompt-Inhalts unterdrückt; es verbleiben lediglich trace_id, Latenz und Token-Zahlen.
Dem Sicherheitsteam oder dem Datenschutzbeauftragten (DSB) wird die folgende Datenfluss-Tabelle vorgelegt:
| Verarbeitungsabschnitt | Datenfluss | Übertragene Daten | Sicherheits-Kontrollmechanismus | Compliance-Standard |
|---|---|---|---|---|
| Abschnitt 1 | Client → Gateway | Ursprünglicher Prompt | TLS 1.3, JWT-Autorisierung | Transportverschlüsselung |
| Abschnitt 2 | In-Memory-Maskierungs-Middleware | Original → Maskierung von Identifikatoren | Prüfsummenbasierte Ersetzung, sofortige Speicherlöschung | Grundsatz der Datensparsamkeit |
| Abschnitt 3 | Backend-Adapter → Astra API | Maskierter Payload | Einspeisung von Schlüsseln über AWS Secrets Manager | Datenschutzbestimmungen zur Datenübertragung |
| Abschnitt 4 | Astra API → Backend-Adapter | Generierter Modeltext | Gewährleistung von Zero Data Retention, TLS | DPA-Unternehmensbedingungen |
| Abschnitt 5 | Backend-Logger → CloudWatch Logs | Token-Metriken und Latenz | Unterdrückung der Inhaltsaufzeichnung, beschränkt auf Metadaten | Gesetzliche Protokollierungspflichten |
Bei der Beantragung der Sicherheitsprüfung werden 1) die Angaben zu Auftragsverarbeitern in der Datenschutzerklärung, 2) eine Kopie des DPA-Vertrags, 3) die Testergebnisse der Prüfsummen-Maskierungs-Middleware und 4) die Dokumentation der Timeout- und Circuit-Breaker-Einstellungen gebündelt eingereicht. Dokumente mit beigefügtem Infrastruktur-Schutzcode verkürzen die technische Prüfphase spürbar.