Aufrufschleifen und Token-Kostenkontrolle für benutzerdefinierte KI-Agenten nach der Demo-Phase
makedream٢٨ يوليو ٢٠٢٦
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Das Erstellen eines KI-Agenten mit Unternehmensdaten auf Basis eines Frameworks verläuft bis zur Prototypenphase meist reibungslos. Das eigentliche Problem beginnt in dem Moment, in dem dieser Code in der Produktionsumgebung bereitgestellt wird. Ein Agent, der auf dem Bildschirm einwandfrei funktionierte, verfängt sich durch eine einzige unerwartete Eingabe in einer endlosen Aufrufschleife, was am Wochenende zu API-Kosten von mehreren Tausend Euro führen kann.
Ein Großteil der Störungen in der Praxis entsteht nicht durch die Grenzen des LLM-Modells selbst, sondern durch fehlendes Zustandsmanagement und fehlende äußere Kontrollmechanismen. Man hat zwar auf Anweisung des Managements ein benutzerdefiniertes Tool entwickelt, muss nun aber den Zustand beenden, in dem man den ganzen Tag damit verbringt, unvorhersehbare Agenten-Halluzinationen zu debuggen und unzählige Token-Abrechnungen zu verarbeiten, anstatt sich auf die eigentliche Arbeit zu konzentrieren.
Sobald ein Agent die Demo-Umgebung verlässt, stößt er auf drei Hindernisse: Blackbox-Operationen, nicht-deterministische Schleifen und das Risiko von Datenlecks.
Wenn die Ein- und Ausgaben der einzelnen Schritte – wie der Denkprozess (Chain-of-Thought), externe Werkzeugaufrufe (Tool Calls) und die Suche in der Vektordatenbank – nicht protokolliert werden, lässt sich bei Problemen die Ursache nicht ermitteln. Allein die Nachverfolgung, an welcher Stelle der Prompt fehlerhaft war, kann einen ganzen Tag in Anspruch nehmen.
Noch gravierender sind Endlosschleifen. Agenten mit einer ReAct-Struktur (Reasoning + Acting) senden dieselbe Anfrage immer wieder neu, wenn das Ergebnis der Werkzeugausführung zweideutig ist.
┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘
Wenn die Abbruchbedingung nicht erreicht wird und die API kontinuierlich mit denselben Argumenten aufgerufen wird, ist das Token-Budget eines ganzen Monats innerhalb weniger Minuten erschöpft. Kaskadierende Ausfälle, bei denen sich Haupt- und Sub-Agenten gegenseitig aufrufen und erneut versuchen, machen mehr als 30 % aller Gesamtsystemausfälle aus.
Sammelt sich der Gesprächsverlauf immer weiter an und füllt das Kontextfenster, sinkt zudem die Leistungsfähigkeit des Modells. Darüber hinaus führen im Python-Code als String fest kodierte Prompts dazu, dass selbst bei der Korrektur einer einfachen Formulierung das gesamte System neu gebaut und bereitgestellt werden muss.
Es ist notwendig, die Erfassungspunkte zu definieren, indem lokale Debugging-Lösungen und Observability-Tools für die Produktion voneinander getrennt werden. In der Entwicklungsphase werden die RAG-Embedding-Qualität und die Werkzeugaufrufe mit dem OpenTelemetry-basierten Arize Phoenix überprüft. In der Produktionsumgebung wird Langfuse mit ClickHouse als Backend eingebunden, um Trace-Logs und den Token-Verbrauch in Echtzeit zu überwachen.
┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘
API-Kosten, die durch sich wiederholende identische Anfragen entstehen, werden durch eine Semantic-Caching-Schicht abgefangen.
Laut von AWS veröffentlichten Analysedaten von 60.000 Anfragen konnten die LLM-Inferenzkosten um bis zu 86 % gesenkt und die Antwortlatenz um 88 % verbessert werden, wenn ein semantischer Cache mit einem Kosinus-Ähnlichkeitsstandard eingesetzt wurde. Auch das RouteLLM-Framework von LMSYS reduzierte die Kosten um 85 %, indem es Anfragen je nach Schwierigkeitsgrad der Frage auf hochpreisige und kostengünstige Modelle aufteilte.
Das semantische Caching wird in folgender Reihenfolge aufgebaut:
Um Endlosschleifen grundlegend zu verhindern, muss ein Circuit-Breaker-Muster zur Überprüfung des Agentenzustands direkt in den Pipeline-Steuerungsfluss integriert werden. Verlässt man sich auf die vom Framework bereitgestellten Standard-Retry-Optionen, führt ein Ausnahmesituation-Fehler zum Absturz des gesamten Dienstes. Es ist sicherer, den Werkzeugnamen und die Argumentwerte mittels SHA-256 zu hashen, in einer Liste zu speichern und die Sitzung erzwungen auf einen anderen Knoten umzuleiten, wenn derselbe Hash dreimal hintereinander auftritt.
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. Wenn die Gesamtzahl der Ausführungen 5 überschreitet, zum festgelegten Fallback-Knoten wechseln
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. Wenn dasselbe Tool mit denselben Argumenten 3 Mal hintereinander aufgerufen wird, sofort blockieren
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. Prüfen auf Schlüsselwort für reguläre Beendigung
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
Unabhängig vom nicht-deterministischen Ausgabezustand des LLM überprüft dieser Router in der Python-Ausführungsumgebung die Bedingungen, wodurch das Eintreten in eine Schleife aufgrund von Halluzinationen physisch verhindert wird.
Belässt man Prompts im Code, lassen sich Regressionsfehler (Regression Failures) nicht erfassen, bei denen nach einer Modellanpassung zuvor funktionierende Features nicht mehr arbeiten. Prompts sollten aus dem Python-Code ausgelagert und als eigenständige YAML-Dateien verwaltet werden.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
Solche getrennten Prompts werden in der CI/CD-Pipeline automatisch getestet, indem die Open-Source-Evaluierungsframeworks DeepEval und Pytest kombiniert werden. Mit den G-Eval-Metriken als Richtwert werden die Antwortwerte des Modells vor der Bereitstellung validiert.
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="Genauigkeit und Einhaltung des Schemas",
criteria="Beantwortet die LLM-Antwort die Frage genau und hält sie das angeforderte JSON-Format perfekt ein?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# Abbruch des Builds, wenn der festgelegte Schwellenwert nicht erreicht wird
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
Das Evaluierungssystem wird in drei Schritten integriert:
deepeval test run automatisch ausgeführt wird, wenn nach einer Codeänderung ein PR erstellt wird.`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
Um Datenlecks von Unternehmensdaten zu verhindern, sind Guardrails erforderlich, die personenbezogene Daten (PII) anonymisieren, bevor die API-Anfrage nach außen geht. Die Integration der Microsoft Presidio-Engine im Gateway maskiert automatisch Sozialversicherungsnummern, E-Mail-Adressen, Telefonnummern und Mitarbeiter-IDs.
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
Auch bei der Einbindung von RAG-Suchen muss sichergestellt werden, dass nur Dokumente abgerufen werden, die der Berechtigungsstufe des Benutzers entsprechen. Vektordatenbanken wie Qdrant oder Pinecone verwalten die Berechtigungstrennung bei Ähnlichkeitssuchen über Metadaten-Filterung (Metadata Filtering).
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
Der grundlegende Schutz vor Prompt-Injections besteht darin, den System-Prompt-Kanal und den Benutzereingabe-Kanal vollständig voneinander zu trennen. Gefährliche Aktionen wie das Ändern von Datenbanken oder das Aufrufen externer APIs sollten aus Sicherheitsgründen einen manuellen Freigabeprozess (Human-in-the-Loop) durchlaufen oder in ihrer Ausführung auf eine isolierte Sandbox beschränkt werden.
Um SaaS-Module zu ersetzen und als internes System dauerhaft zu betreiben, sind regelmäßige Überprüfungsroutinen erforderlich.
| Intervall | Überprüfungspunkt für den Betrieb | Detaillierte Aufgabe |
|---|---|---|
| Täglich | Fehlerrate und Token-Verbrauch | HTTP 5xx-Fehlerrate und Token-Verbrauch nach Abteilung im Langfuse-Dashboard prüfen |
| Täglich | Circuit-Breaker-Blockierungsverlauf | Werkzeugnamen und Argumentmuster von Sitzungen erfassen und korrigieren, die durch Schleifenerkennung unterbrochen wurden |
| Wöchentlich | Fehlgeschlagene RAG-Suchanfragen extrahieren | Anfragen mit einem Ähnlichkeitswert unter 0,6 herausfiltern und dem Test-Set korrekter Antworten hinzufügen |
| Wöchentlich | Fehlalarme bei der PII-Maskierung überprüfen | Presidio-Verarbeitungsprotokolle stichprobenartig prüfen, um fehlende Maskierungen zu identifizieren |
| Monatlich | CI/CD-Evaluierungskriterien aktualisieren | Automatische Testfälle anpassen, um Geschäftsänderungen widerzuspiegeln |
Wenn Sie zwischen dem Hosting eines eigenen Modells (vLLM-basiert) und dem Abonnieren einer externen API abwägen, berechnen Sie dies auf der Grundlage des täglichen Datenverkehrs.
Eine AWS EC2 g5.2xlarge-Instanz mit einer NVIDIA A10G GPU kostet ca. 880 $ pro Monat. Im Gegensatz dazu liegen die API-Preise für GPT-4o bei etwa 2,50 pro 1 Million Ausgabe-Tokens. Liegt der tägliche Traffic unter 50 Millionen Tokens, ist das API-Abonnement unter Berücksichtigung des Aufwands für DevOps-Engineers und der GPU-Fixkosten vorteilhafter. Erst wenn der tägliche Traffic 100 Millionen Tokens überschreitet oder eine vollständige Isolierung im internen Netzwerk zwingend erforderlich ist, ist der Wechsel zu einem eigenen vLLM-basierten Hosting sinnvoll.
Um für Ausfälle des Hauptmodells gewappnet zu sein, empfiehlt sich eine 4-stufige Backup-Routing-Struktur.
┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘
Die Optimierung eines benutzerdefinierten Agentensystems erfolgt in 90-Tage-Blöcken.
| Zeitraum | Anwendungsziel | Konkrete Aufgaben |
|---|---|---|
| Tag 1–30 | Transparenz und Datenschutz | Presidio-Maskierungsengine installieren und Langfuse-Logging-SDK einbinden |
| Tag 31–60 | Schleifenblockierung und Kostensenkung | Redis Semantic Caching anwenden und Python-basierten Circuit Breaker verbinden |
| Tag 61–90 | Testautomatisierung und Zugriffskontrolle | Prompt-Git-Verwaltung, DeepEval CI/CD-Anbindung und Qdrant RBAC-Metadaten anwenden |
Im ersten Monat richten Sie eine Observability-Lösung ein, um Datenlecks personenbezogener Daten zu verhindern und alle Anfragen zu protokollieren. Im zweiten Monat binden Sie ein Caching für sich wiederholende Prompts sowie einen Router zur Schleifenblockierung ein, um unerwartete Ausgaben zu unterbinden. Im letzten Monat vervollständigen Sie die Prompt-Versionsverwaltung und das automatische Testevaluierungssystem, sodass ein stabiler Betrieb ohne Sorge vor Fehlfunktionen des Agenten bei jedem Deployment möglich wird.