Aufrufschleifen und Token-Kostenkontrolle für benutzerdefinierte KI-Agenten nach der Demo-Phase

Computing/Software19 min read

TuBrief Editorial Written with AI assistance from the source video. The video is the authority. Source video: "Build vs. Buy" verschiebt sich in Richtung "Eigenentwicklung" · Maximilian Schwarzmüller

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.


Drei Hürden in der Produktionsumgebung

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.


Ein Beobachtungssystem zur Unterbrechung von Endlosschleifen und Kostenexplosionen

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:

  1. Redis Vector Search Backend vorbereiten: Starten Sie eine Redis-Instanz, um den Eingabe-Prompt in einen Embedding-Vektor umzuwandeln und zu speichern.
  2. Ähnlichkeitsbeurteilung und Cache-Rückgabe: Wenn die Kosinus-Ähnlichkeit zwischen dem neu eingehenden Prompt-Vektor und den bereits gespeicherten Vektoren 0.92 oder höher ist, wird die Antwort direkt aus dem Cache zurückgegeben, ohne die externe LLM aufzurufen.
  3. Cache-Miss speichern: Nur wenn die Ähnlichkeit unter 0.92 liegt, wird die externe API aufgerufen, und die empfangene Antwort wird neu in Redis aktualisiert.

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.


Trennung von Prompt-Versionen und CI/CD-Automatisierung

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:

  • role: "system"
    content: |
    Sie sind ein Agent zur Analyse interner Unternehmensdatenbanken.
    Sie müssen gemäß dem angegebenen JSON-Schema antworten. Generieren Sie zusätzliche Fragen, falls Informationen fehlen.
  • role: "user"
    content: "{{ user_input }}"
    `

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

1. Evaluierungskriterien definieren

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
)

2. Pytest-basierter Regressionstest

@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:

  1. Evaluierungsdatenmenge erstellen: Erstellen Sie ein Set aus Hauptdomänenfragen und den dazugehörigen korrekten Antworten und legen Sie den zulässigen Schwellenwert (Threshold) fest.
  2. GitHub Actions Workflow einbinden: Konfigurieren Sie die Action so, dass der Befehl deepeval test run automatisch ausgeführt wird, wenn nach einer Codeänderung ein PR erstellt wird.
  3. Shadow-Tests anwenden: Vor dem Mergen in den Main-Branch werden echte Traffic-Anfragen asynchron dupliziert und gesendet, um Antwortqualität und Latenzzeit mit der Vorgängerversion zu vergleichen, bevor die finale Bereitstellung erfolgt.

`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
`


Anonymisierung und Zugriffskontrolle

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

1. Analyzer Engine und Regel für Mitarbeiter-IDs registrieren

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)

2. Anonymizer Engine initialisieren

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.


Infrastruktur-Lebenszyklus und stufenweise Konsistenzsicherung

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 pro1MillionEingabe−Tokensund10,00pro 1 Million Eingabe-Tokens und 10,00 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 │ └─────────────────────────────────────────────────────────────────────────┘


Ablauf der Betriebsstabilisierung

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.

Related Video

"Build vs. Buy" verschiebt sich in Richtung "Eigenentwicklung"

Maximilian Schwarzmüller10:07

More from the community

Comments

Log in to leave a comment.

No comments yet.