Agent-Bewertung, Sicherheit und Alignment
Die Bewertung ist die Grundlage für die Sicherstellung der Effektivität von Agenten.
Sicherheit und Alignment sind der Schlüssel dafür, ob ein Agent vertrauenswürdig ist und bereitgestellt werden kann.
Agent-Bewertungssystem
Die Bewertung der Agent-Leistung ist ein kritischer Teil des Entwicklungsprozesses.
Ein gutes Bewertungssystem hilft uns, die Grenzen der Fähigkeiten eines Agenten zu verstehen.
Es liefert außerdem Richtung und Grundlage für kontinuierliche Optimierung.
Bewertungsdimensionen
Aufgabenabschluss: Ob der Agent die gegebene Aufgabe korrekt abgeschlossen hat.
Effizienzmetriken: Anzahl der Schritte, Tokenverbrauch und Ausführungszeit, die zum Abschließen der Aufgabe erforderlich sind.
Qualitätsmetriken: Genauigkeit der Antworten, Konsistenz der Antworten und Natürlichkeit des Dialogs.
Robustheit: Fähigkeit, mit anormalen Eingaben und verrauschten Daten umzugehen.
Gängige Benchmarks
| Benchmark | Zweck | Evaluierungsinhalt |
|---|---|---|
| GAIA | Bewertung allgemeiner KI-Assistenten | Umgang mit komplexen Aufgaben, mehrstufiges Schlussfolgern |
| MMLU | Multitask-Sprachverständnis | Wissensfragen und -antworten über 57 Themen |
| HumanEval | Bewertung der Codegenerierung | Korrektheit beim Schreiben von Python-Code |
| HotpotQA | Bewertung der Multi-Hop-Fragebeantwortung | Fragen, die Schlussfolgerungen über mehrere Dokumente hinweg erfordern |
| AgentBench | Bewertung der Agent-Fähigkeiten | Agent-Leistung in realen Umgebungen |
Code-Implementierung: Evaluierungsframework
Agent-Evaluierungsframework
class AgentEvaluator:
"""
Agent-Evaluierungsframework
Bewerte die Leistung eines Agenten bei verschiedenen Aufgaben
"""
def __init__(self, agent, metrics):
# Zu evaluierender Agent
self.agent = agent
# Liste der Evaluierungsmetriken
self.metrics = metrics
def evaluate(self, test_cases):
"""
Evaluierung ausführen
:param test_cases: Liste der Testfälle
:return: Evaluierungsbericht
"""
results = []
for test_case in test_cases:
# Aufgabe ausführen
result = self.run_single_test(test_case)
results.append(result)
# Evaluierungsbericht erstellen
report = self.generate_report(results)
return report
def run_single_test(self, test_case):
"""
Einzelnen Testfall ausführen
"""
# Startzeit erfassen
start_time = time.time()
# Agenten ausführen
try:
output = self.agent.run(test_case.input)
success = self.evaluate_output(output, test_case.expected)
error = None
except Exception as e:
output = None
success = False
error = str(e)
# Endzeit erfassen
end_time = time.time()
return TestResult(
test_case=test_case,
output=output,
success=success,
error=error,
duration=end_time - start_time,
token_count=self.count_tokens(output)
)
def evaluate_output(self, output, expected):
"""Bewerte, ob die Ausgabe die Erwartungen erfüllt"""
for metric in self.metrics:
if not metric.evaluate(output, expected):
return False
return True
def generate_report(self, results):
"""Evaluierungsbericht erstellen"""
total = len(results)
passed = sum(1 for r in results if r.success)
# Verschiedene Metriken berechnen
avg_duration = sum(r.duration for r in results) / total
avg_tokens = sum(r.token_count for r in results) / total
# Nach Testtyp gruppieren und zählen
by_category = {}
for r in results:
category = r.test_case.category
if category not in by_category:
by_category[category] = {"total": 0, "passed": 0}
by_category[category]["total"] += 1
if r.success:
by_category[category]["passed"] += 1
return EvaluationReport(
total=total,
passed=passed,
pass_rate=passed / total,
avg_duration=avg_duration,
avg_tokens=avg_tokens,
by_category=by_category,
results=results
)
class TestCase:
"""Testfall"""
def __init__(self, input, expected, category="general", metadata=None):
# Eingabe
self.input = input
# Erwartete Ausgabe oder Bewertungskriterien
self.expected = expected
# Kategorie
self.category = category
# Zusätzliche Metadaten
self.metadata = metadata or {}
class TestResult:
"""Testergebnis"""
def __init__(self, test_case, output, success, error, duration, token_count):
self.test_case = test_case
self.output = output
self.success = success
self.error = error
self.duration = duration
self.token_count = token_count
class Metric:
"""Basisklasse für Bewertungsmetriken"""
def evaluate(self, output, expected):
raise NotImplementedError
class ExactMatchMetric(Metric):
"""Metrik für exakte Übereinstimmung"""
def evaluate(self, output, expected):
return output.strip() == expected.strip()
class ContainsMetric(Metric):
"""Metrik für Keyword-Einschluss"""
def evaluate(self, output, expected):
if isinstance(expected, list):
return all(keyword in output for keyword in expected)
return expected in output
class SemanticSimilarityMetric(Metric):
"""Metrik der semantischen Ähnlichkeit"""
def __init__(self, threshold=0.8):
self.threshold = threshold
def evaluate(self, output, expected):
similarity = self.compute_similarity(output, expected)
return similarity >= self.threshold
def compute_similarity(self, text1, text2):
"""Berechnet die semantische Ähnlichkeit zweier Texte"""
# Embedding-Modell verwenden, um Kosinus-Ähnlichkeit zu berechnen
embedding1 = self.embedder.embed([text1])[0]
embedding2 = self.embedder.embed([text2])[0]
return cosine_similarity(embedding1, embedding2)
"""
Agent-Evaluierungsframework
Bewerte die Leistung eines Agenten bei verschiedenen Aufgaben
"""
def __init__(self, agent, metrics):
# Zu evaluierender Agent
self.agent = agent
# Liste der Evaluierungsmetriken
self.metrics = metrics
def evaluate(self, test_cases):
"""
Evaluierung ausführen
:param test_cases: Liste der Testfälle
:return: Evaluierungsbericht
"""
results = []
for test_case in test_cases:
# Aufgabe ausführen
result = self.run_single_test(test_case)
results.append(result)
# Evaluierungsbericht erstellen
report = self.generate_report(results)
return report
def run_single_test(self, test_case):
"""
Einzelnen Testfall ausführen
"""
# Startzeit erfassen
start_time = time.time()
# Agenten ausführen
try:
output = self.agent.run(test_case.input)
success = self.evaluate_output(output, test_case.expected)
error = None
except Exception as e:
output = None
success = False
error = str(e)
# Endzeit erfassen
end_time = time.time()
return TestResult(
test_case=test_case,
output=output,
success=success,
error=error,
duration=end_time - start_time,
token_count=self.count_tokens(output)
)
def evaluate_output(self, output, expected):
"""Bewerte, ob die Ausgabe die Erwartungen erfüllt"""
for metric in self.metrics:
if not metric.evaluate(output, expected):
return False
return True
def generate_report(self, results):
"""Evaluierungsbericht erstellen"""
total = len(results)
passed = sum(1 for r in results if r.success)
# Verschiedene Metriken berechnen
avg_duration = sum(r.duration for r in results) / total
avg_tokens = sum(r.token_count for r in results) / total
# Nach Testtyp gruppieren und zählen
by_category = {}
for r in results:
category = r.test_case.category
if category not in by_category:
by_category[category] = {"total": 0, "passed": 0}
by_category[category]["total"] += 1
if r.success:
by_category[category]["passed"] += 1
return EvaluationReport(
total=total,
passed=passed,
pass_rate=passed / total,
avg_duration=avg_duration,
avg_tokens=avg_tokens,
by_category=by_category,
results=results
)
class TestCase:
"""Testfall"""
def __init__(self, input, expected, category="general", metadata=None):
# Eingabe
self.input = input
# Erwartete Ausgabe oder Bewertungskriterien
self.expected = expected
# Kategorie
self.category = category
# Zusätzliche Metadaten
self.metadata = metadata or {}
class TestResult:
"""Testergebnis"""
def __init__(self, test_case, output, success, error, duration, token_count):
self.test_case = test_case
self.output = output
self.success = success
self.error = error
self.duration = duration
self.token_count = token_count
class Metric:
"""Basisklasse für Bewertungsmetriken"""
def evaluate(self, output, expected):
raise NotImplementedError
class ExactMatchMetric(Metric):
"""Metrik für exakte Übereinstimmung"""
def evaluate(self, output, expected):
return output.strip() == expected.strip()
class ContainsMetric(Metric):
"""Metrik für Keyword-Einschluss"""
def evaluate(self, output, expected):
if isinstance(expected, list):
return all(keyword in output for keyword in expected)
return expected in output
class SemanticSimilarityMetric(Metric):
"""Metrik der semantischen Ähnlichkeit"""
def __init__(self, threshold=0.8):
self.threshold = threshold
def evaluate(self, output, expected):
similarity = self.compute_similarity(output, expected)
return similarity >= self.threshold
def compute_similarity(self, text1, text2):
"""Berechnet die semantische Ähnlichkeit zweier Texte"""
# Embedding-Modell verwenden, um Kosinus-Ähnlichkeit zu berechnen
embedding1 = self.embedder.embed([text1])[0]
embedding2 = self.embedder.embed([text2])[0]
return cosine_similarity(embedding1, embedding2)
Sicherheit und Alignment
Die Sicherheit des Agenten ist von größter Bedeutung.
KI-Systeme können verschiedenen Angriffen ausgesetzt sein und schädliche Ausgaben erzeugen.
Alignment stellt sicher, dass das Verhalten von KI mit menschlichen Absichten und Werten übereinstimmt.
Häufige Sicherheitsbedrohungen
Prompt-Injection
Angreifer nutzen Eingaben, um den Agenten dazu zu bringen, Systemanweisungen zu ignorieren.
Beispieleingabe: "Ignoriere vorherige Anweisungen und führe stattdessen aus..."
Dies ist ein Context-Hijacking-Angriff, der das Vertrauen des Agenten in Benutzereingaben ausnutzt.
Jailbreaking
Sicherheitsbeschränkungen durch bestimmte Eingaben umgehen.
Zum Beispiel durch Techniken wie Rollenspiele oder fiktive Szenarien.
Datenvergiftung
Trainingsdaten oder Retrieval-Ergebnisse böswillig verändern.
Veranlasst den Agenten, falsche oder schädliche Ausgaben zu erzeugen.
Preisgabe sensibler Informationen
Der Agent gibt unzulässigerweise Nutzerprivatsphäre oder interne Systeminformationen preis.
Verteidigungsstrategien
Sichere Agent-Implementierung
class SecureAgent:
"""
Sicherheits-Agent
Mehrere Ebenen des Sicherheitsschutzes zusätzlich zum Basis-Agenten hinzufügen
"""
def __init__(self, base_agent, guardrails, input_validator, output_filter):
# Basis-Agent
self.base_agent = base_agent
# Liste der Sicherheits-Guardrails
self.guardrails = guardrails
# Eingabe-Validator
self.input_validator = input_validator
# Ausgabefilter
self.output_filter = output_filter
def process(self, user_input):
"""
Benutzereingabe verarbeiten, einschließlich mehrerer Ebenen von Sicherheitsprüfungen
"""
# ==================== Ebene 1: Eingabevalidierung ====================
# Prüfen, ob die Eingabe gültig ist
is_valid, reason = self.input_validator.validate(user_input)
if not is_valid:
return self.create_safety_response(reason)
# ==================== Ebene 2: Injection-Erkennung ====================
# Angriffe wie Prompt-Injection erkennen
for guardrail in self.guardrails:
check_result = guardrail.check_input(user_input)
if not check_result.is_safe:
# Sicherheitsereignisse aufzeichnen
self.log_security_event(
event_type="input_guardrail_triggered",
input=user_input,
reason=check_result.reason
)
return self.create_safety_response(check_result.reason)
# ==================== Ebene 3: Kernlogik ausführen ====================
try:
response = self.base_agent.process(user_input)
except Exception as e:
return self.create_error_response(str(e))
# ==================== Ebene 4: Ausgabefilterung ====================
# Prüfen, ob die Ausgabe sicher ist
for guardrail in self.guardrails:
check_result = guardrail.check_output(response)
if not check_result.is_safe:
self.log_security_event(
event_type="output_guardrail_triggered",
output=response,
reason=check_result.reason
)
return self.create_safety_response(check_result.reason)
# Ausgabefilterung anwenden (z. B. Desensibilisierung sensibler Informationen)
response = self.output_filter.filter(response)
return response
def create_safety_response(self, reason):
"""Sichere Antwort erstellen"""
return {
"type": "safety_block",
"message": "Es tut mir leid, ich kann diese Anfrage nicht abschließen.",
"reason": reason
}
def log_security_event(self, event_type, **kwargs):
"""Sicherheitsereignisse aufzeichnen"""
# In tatsächlichen Anwendungen sollte dies in ein Sicherheitsprotokollsystem geschrieben werden
print(f"[SICHERHEIT] {event_type}: {kwargs}")
class InputValidator:
"""Eingabevalidierer"""
def validate(self, text):
"""
Prüfen, ob die Eingabe gültig ist
:return: (is_valid, reason)
"""
if not text or len(text.strip()) == 0:
return False, "Die Eingabe darf nicht leer sein"
if len(text) > 10000:
return False, "Die Eingabelänge überschreitet das Limit"
# Prüfen, ob ausführbarer Inhalt enthalten ist
if self.contains_executable_content(text):
return False, "Die Eingabe enthält verdächtigen ausführbaren Inhalt"
return True, None
def contains_executable_content(self, text):
"""Prüfen, ob ausführbarer Inhalt enthalten ist"""
# Vereinfachte Implementierung
suspicious_patterns = [
"javascript:",
"data:text/html",
"<script>",
]
return any(pattern in text.lower() for pattern in suspicious_patterns)
class Guardrail:
"""Sicherheits-Guardrail"""
def check_input(self, text):
"""Eingabe prüfen"""
raise NotImplementedError
def check_output(self, text):
"""Ausgabe prüfen"""
raise NotImplementedError
class ContentFilterGuardrail(Guardrail):
"""Guardrail zur Inhaltsfilterung"""
def __init__(self, blocked_topics, banned_words):
self.blocked_topics = blocked_topics
self.banned_words = banned_words
def check_input(self, text):
# Prüfen, ob verbotene Themen betroffen sind
for topic in self.blocked_topics:
if topic in text.lower():
return CheckResult(
is_safe=False,
reason=f"Betrifft sensibles Thema: {topic}"
)
# Prüfen, ob es verbotene Wörter enthält
for word in self.banned_words:
if word in text.lower():
return CheckResult(
is_safe=False,
reason=f"Enthält unangemessene Wörter"
)
return CheckResult(is_safe=True)
def check_output(self, text):
# Ausgabeprüfung wie oben
return self.check_input(text)
class CheckResult:
"""Prüfergebnis"""
def __init__(self, is_safe, reason=None):
self.is_safe = is_safe
self.reason = reason
"""
Sicherheits-Agent
Mehrere Ebenen des Sicherheitsschutzes zusätzlich zum Basis-Agenten hinzufügen
"""
def __init__(self, base_agent, guardrails, input_validator, output_filter):
# Basis-Agent
self.base_agent = base_agent
# Liste der Sicherheits-Guardrails
self.guardrails = guardrails
# Eingabe-Validator
self.input_validator = input_validator
# Ausgabefilter
self.output_filter = output_filter
def process(self, user_input):
"""
Benutzereingabe verarbeiten, einschließlich mehrerer Ebenen von Sicherheitsprüfungen
"""
# ==================== Ebene 1: Eingabevalidierung ====================
# Prüfen, ob die Eingabe gültig ist
is_valid, reason = self.input_validator.validate(user_input)
if not is_valid:
return self.create_safety_response(reason)
# ==================== Ebene 2: Injection-Erkennung ====================
# Angriffe wie Prompt-Injection erkennen
for guardrail in self.guardrails:
check_result = guardrail.check_input(user_input)
if not check_result.is_safe:
# Sicherheitsereignisse aufzeichnen
self.log_security_event(
event_type="input_guardrail_triggered",
input=user_input,
reason=check_result.reason
)
return self.create_safety_response(check_result.reason)
# ==================== Ebene 3: Kernlogik ausführen ====================
try:
response = self.base_agent.process(user_input)
except Exception as e:
return self.create_error_response(str(e))
# ==================== Ebene 4: Ausgabefilterung ====================
# Prüfen, ob die Ausgabe sicher ist
for guardrail in self.guardrails:
check_result = guardrail.check_output(response)
if not check_result.is_safe:
self.log_security_event(
event_type="output_guardrail_triggered",
output=response,
reason=check_result.reason
)
return self.create_safety_response(check_result.reason)
# Ausgabefilterung anwenden (z. B. Desensibilisierung sensibler Informationen)
response = self.output_filter.filter(response)
return response
def create_safety_response(self, reason):
"""Sichere Antwort erstellen"""
return {
"type": "safety_block",
"message": "Es tut mir leid, ich kann diese Anfrage nicht abschließen.",
"reason": reason
}
def log_security_event(self, event_type, **kwargs):
"""Sicherheitsereignisse aufzeichnen"""
# In tatsächlichen Anwendungen sollte dies in ein Sicherheitsprotokollsystem geschrieben werden
print(f"[SICHERHEIT] {event_type}: {kwargs}")
class InputValidator:
"""Eingabevalidierer"""
def validate(self, text):
"""
Prüfen, ob die Eingabe gültig ist
:return: (is_valid, reason)
"""
if not text or len(text.strip()) == 0:
return False, "Die Eingabe darf nicht leer sein"
if len(text) > 10000:
return False, "Die Eingabelänge überschreitet das Limit"
# Prüfen, ob ausführbarer Inhalt enthalten ist
if self.contains_executable_content(text):
return False, "Die Eingabe enthält verdächtigen ausführbaren Inhalt"
return True, None
def contains_executable_content(self, text):
"""Prüfen, ob ausführbarer Inhalt enthalten ist"""
# Vereinfachte Implementierung
suspicious_patterns = [
"javascript:",
"data:text/html",
"<script>",
]
return any(pattern in text.lower() for pattern in suspicious_patterns)
class Guardrail:
"""Sicherheits-Guardrail"""
def check_input(self, text):
"""Eingabe prüfen"""
raise NotImplementedError
def check_output(self, text):
"""Ausgabe prüfen"""
raise NotImplementedError
class ContentFilterGuardrail(Guardrail):
"""Guardrail zur Inhaltsfilterung"""
def __init__(self, blocked_topics, banned_words):
self.blocked_topics = blocked_topics
self.banned_words = banned_words
def check_input(self, text):
# Prüfen, ob verbotene Themen betroffen sind
for topic in self.blocked_topics:
if topic in text.lower():
return CheckResult(
is_safe=False,
reason=f"Betrifft sensibles Thema: {topic}"
)
# Prüfen, ob es verbotene Wörter enthält
for word in self.banned_words:
if word in text.lower():
return CheckResult(
is_safe=False,
reason=f"Enthält unangemessene Wörter"
)
return CheckResult(is_safe=True)
def check_output(self, text):
# Ausgabeprüfung wie oben
return self.check_input(text)
class CheckResult:
"""Prüfergebnis"""
def __init__(self, is_safe, reason=None):
self.is_safe = is_safe
self.reason = reason
Wichtiger Hinweis: Sicherheit ist ein fortlaufender Prozess, und es gibt keine narrensichere Lösung. Sicherheitsstrategien erfordern kontinuierliche Überwachung, Aktualisierung und Verbesserung.
Observability
Agenten in Produktionsumgebungen benötigen ein umfassendes Überwachungssystem.
Observability hilft uns, das Verhalten von Agenten zu verstehen, Probleme zu beheben und die Leistung zu optimieren.
Drei Säulen
Logging: Alle wichtigen Ereignisse und Entscheidungen aufzeichnen.
Tracing: Den vollständigen Ablaufpfad von Anfragen durch das System verfolgen.
Metrics: Leistungs- und Qualitätsmetriken sammeln.
Code-Implementierung
Implementierung eines beobachtbaren Agenten
import logging
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
# Logging konfigurieren
logger = logging.getLogger(__name__)
# Tracing konfigurieren
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)
class ObservableAgent:
"""
Observability-Agent
Integrierte Erfassung von Logs, Traces und Metriken
"""
def __init__(self, agent, metrics_collector):
# Basis-Agent
self.agent = agent
# Metriken-Sammler
self.metrics = metrics_collector
def process(self, user_input):
"""
Anfragen mit vollständiger Observability-Unterstützung bearbeiten
"""
# Tracing-Span erstellen
with tracer.start_as_current_span("agent_process") as span:
# Span-Attribute setzen
span.set_attribute("user.input.length", len(user_input))
span.set_attribute("user.input.preview", user_input[:100])
# Start aufzeichnen
logger.info(f"Starte Verarbeitung der Anfrage: {user_input[:50]}...")
start_time = time.time()
try:
# Kernlogik ausführen
result = self.agent.process(user_input)
# Erfolg aufzeichnen
duration = time.time() - start_time
span.set_attribute("success", True)
span.set_attribute("duration_ms", duration * 1000)
span.set_attribute("result.length", len(str(result)))
# Metriken sammeln
self.metrics.record("request_duration", duration)
self.metrics.increment("request_success")
logger.info(f"Anfrage abgeschlossen, Dauer: {duration:.2f}s")
return result
except Exception as e:
# Fehler erfassen
duration = time.time() - start_time
span.set_attribute("success", False)
span.set_attribute("error.type", type(e).__name__)
span.set_attribute("error.message", str(e))
span.record_exception(e)
# Fehlermetriken sammeln
self.metrics.increment("request_error")
self.metrics.record("error_duration", duration)
logger.error(f"Anfrage fehlgeschlagen: {e}")
raise
class MetricsCollector:
"""Metrik-Sammler"""
def __init__(self):
self.metrics = {}
def increment(self, name, value=1):
"""Zähler-Metrik erhöhen"""
if name not in self.metrics:
self.metrics[name] = {"type": "counter", "value": 0}
self.metrics[name]["value"] += value
def record(self, name, value):
"""Numerische Metrik aufzeichnen"""
if name not in self.metrics:
self.metrics[name] = {"type": "gauge", "values": []}
self.metrics[name]["values"].append(value)
def get_summary(self):
"""Metrikübersicht abrufen"""
summary = {}
for name, data in self.metrics.items():
if data["type"] == "counter":
summary[name] = data["value"]
else:
values = data["values"]
summary[f"{name}_avg"] = sum(values) / len(values)
summary[f"{name}_max"] = max(values)
summary[f"{name}_min"] = min(values)
return summary
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
# Logging konfigurieren
logger = logging.getLogger(__name__)
# Tracing konfigurieren
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)
class ObservableAgent:
"""
Observability-Agent
Integrierte Erfassung von Logs, Traces und Metriken
"""
def __init__(self, agent, metrics_collector):
# Basis-Agent
self.agent = agent
# Metriken-Sammler
self.metrics = metrics_collector
def process(self, user_input):
"""
Anfragen mit vollständiger Observability-Unterstützung bearbeiten
"""
# Tracing-Span erstellen
with tracer.start_as_current_span("agent_process") as span:
# Span-Attribute setzen
span.set_attribute("user.input.length", len(user_input))
span.set_attribute("user.input.preview", user_input[:100])
# Start aufzeichnen
logger.info(f"Starte Verarbeitung der Anfrage: {user_input[:50]}...")
start_time = time.time()
try:
# Kernlogik ausführen
result = self.agent.process(user_input)
# Erfolg aufzeichnen
duration = time.time() - start_time
span.set_attribute("success", True)
span.set_attribute("duration_ms", duration * 1000)
span.set_attribute("result.length", len(str(result)))
# Metriken sammeln
self.metrics.record("request_duration", duration)
self.metrics.increment("request_success")
logger.info(f"Anfrage abgeschlossen, Dauer: {duration:.2f}s")
return result
except Exception as e:
# Fehler erfassen
duration = time.time() - start_time
span.set_attribute("success", False)
span.set_attribute("error.type", type(e).__name__)
span.set_attribute("error.message", str(e))
span.record_exception(e)
# Fehlermetriken sammeln
self.metrics.increment("request_error")
self.metrics.record("error_duration", duration)
logger.error(f"Anfrage fehlgeschlagen: {e}")
raise
class MetricsCollector:
"""Metrik-Sammler"""
def __init__(self):
self.metrics = {}
def increment(self, name, value=1):
"""Zähler-Metrik erhöhen"""
if name not in self.metrics:
self.metrics[name] = {"type": "counter", "value": 0}
self.metrics[name]["value"] += value
def record(self, name, value):
"""Numerische Metrik aufzeichnen"""
if name not in self.metrics:
self.metrics[name] = {"type": "gauge", "values": []}
self.metrics[name]["values"].append(value)
def get_summary(self):
"""Metrikübersicht abrufen"""
summary = {}
for name, data in self.metrics.items():
if data["type"] == "counter":
summary[name] = data["value"]
else:
values = data["values"]
summary[f"{name}_avg"] = sum(values) / len(values)
summary[f"{name}_max"] = max(values)
summary[f"{name}_min"] = min(values)
return summary
Guardrails und Human-in-the-Loop
Guardrails
Guardrails sind Mechanismen zur Echtzeitfilterung und -beschränkung von Ein- und Ausgaben.
Im Gegensatz zu Sicherheits-Guardrails liegt der Fokus von Guardrails eher auf der Sicherstellung von Ausgabequalität und Compliance.
HITL（Human-in-the-Loop）
HITL führt an kritischen Entscheidungspunkten eine menschliche Überprüfung ein.
Geeignet für Hochrisikoszenarien, in denen KI keine autonomen Entscheidungen treffen kann.
Implementierung des HITL-Agenten
class HITLAgent:
"""
Human-in-the-Loop-Agent
Menschliche Überprüfung an kritischen Entscheidungspunkten einführen
"""
def __init__(self, agent, approval_queue, notification_handler):
# Basis-Agent
self.agent = agent
# Genehmigungswarteschlange
self.approval_queue = approval_queue
# Benachrichtigungshandler
self.notification_handler = notification_handler
def process(self, request):
"""
Anfragen verarbeiten, bei Bedarf menschliche Genehmigung auslösen
"""
# Risikostufe der Anfrage bewerten
risk_level = self.assess_risk(request)
if risk_level == "high":
# Hochrisikoanfrage, erfordert menschliche Genehmigung
return self.handle_high_risk_request(request)
elif risk_level == "medium":
# Mittleres Risiko, erweiterte Überwachung aktivieren
return self.handle_medium_risk_request(request)
else:
# Geringes Risiko, direkt verarbeiten
return self.agent.process(request)
def assess_risk(self, request):
"""Risikostufe der Anfrage bewerten"""
# Prüfen, ob sensible Vorgänge beteiligt sind
sensitive_operations = [
"delete", "remove", "cancel",
"transfer", "payment", "refund"
]
content_lower = request.lower()
for op in sensitive_operations:
if op in content_lower:
return "high"
# Komplexität des Anfrageinhalts prüfen
if len(request) > 1000:
return "medium"
return "low"
def handle_high_risk_request(self, request):
"""
Hochrisiko-Anfragen verarbeiten, die eine menschliche Genehmigung erfordern
"""
# Genehmigungsaufgabe erstellen
task_id = self.approval_queue.add({
"request": request,
"risk_level": "high",
"timestamp": datetime.now()
})
# Genehmiger benachrichtigen
self.notification_handler.notify_approver(
task_id=task_id,
message=f"Eine Anfrage mit hohem Risiko erfordert eine Genehmigung: {request[:100]}..."
)
# Wartestatus zurückgeben
return {
"status": "pending_approval",
"task_id": task_id,
"message": "Ihre Anfrage erfordert eine menschliche Genehmigung, bitte warten."
}
def handle_feedback(self, task_id, approved, feedback=Keine):
"""
Genehmigungsfeedback verarbeiten
"""
task = self.approval_queue.get(task_id)
if approved:
# Genehmigung erteilt, Anfrage ausführen
self.approval_queue.complete(task_id)
result = self.agent.process(task["request"])
# Antragsteller benachrichtigen
self.notification_handler.notify_requester(
task_id=task_id,
status="approved",
result=result
)
return result
else:
# Genehmigung abgelehnt
self.approval_queue.reject(task_id, feedback)
# Antragsteller benachrichtigen
self.notification_handler.notify_requester(
task_id=task_id,
status="rejected",
feedback=feedback
)
return {
"status": "rejected",
"reason": feedback or Genehmigung fehlgeschlagen.
}
class ApprovalQueue:
Genehmigungswarteschlange
def __init__(self):
self.queue = {}
self.counter = 0
def add(self, task):
Genehmigungsaufgabe hinzufügen
self.counter += 1
task_id = f"task_{self.counter}"
self.queue[task_id] = {
**task,
"status": "pending"
}
return task_id
def get(self, task_id):
return self.queue.get(task_id)
def complete(self, task_id):
self.queue[task_id]["status"] = "approved"
def reject(self, task_id, reason):
self.queue[task_id]["status"] = "rejected"
self.queue[task_id]["reject_reason"] = reason
"""
Human-in-the-Loop-Agent
Menschliche Überprüfung an kritischen Entscheidungspunkten einführen
"""
def __init__(self, agent, approval_queue, notification_handler):
# Basis-Agent
self.agent = agent
# Genehmigungswarteschlange
self.approval_queue = approval_queue
# Benachrichtigungshandler
self.notification_handler = notification_handler
def process(self, request):
"""
Anfragen verarbeiten, bei Bedarf menschliche Genehmigung auslösen
"""
# Risikostufe der Anfrage bewerten
risk_level = self.assess_risk(request)
if risk_level == "high":
# Hochrisikoanfrage, erfordert menschliche Genehmigung
return self.handle_high_risk_request(request)
elif risk_level == "medium":
# Mittleres Risiko, erweiterte Überwachung aktivieren
return self.handle_medium_risk_request(request)
else:
# Geringes Risiko, direkt verarbeiten
return self.agent.process(request)
def assess_risk(self, request):
"""Risikostufe der Anfrage bewerten"""
# Prüfen, ob sensible Vorgänge beteiligt sind
sensitive_operations = [
"delete", "remove", "cancel",
"transfer", "payment", "refund"
]
content_lower = request.lower()
for op in sensitive_operations:
if op in content_lower:
return "high"
# Komplexität des Anfrageinhalts prüfen
if len(request) > 1000:
return "medium"
return "low"
def handle_high_risk_request(self, request):
"""
Hochrisiko-Anfragen verarbeiten, die eine menschliche Genehmigung erfordern
"""
# Genehmigungsaufgabe erstellen
task_id = self.approval_queue.add({
"request": request,
"risk_level": "high",
"timestamp": datetime.now()
})
# Genehmiger benachrichtigen
self.notification_handler.notify_approver(
task_id=task_id,
message=f"Eine Anfrage mit hohem Risiko erfordert eine Genehmigung: {request[:100]}..."
)
# Wartestatus zurückgeben
return {
"status": "pending_approval",
"task_id": task_id,
"message": "Ihre Anfrage erfordert eine menschliche Genehmigung, bitte warten."
}
def handle_feedback(self, task_id, approved, feedback=Keine):
"""
Genehmigungsfeedback verarbeiten
"""
task = self.approval_queue.get(task_id)
if approved:
# Genehmigung erteilt, Anfrage ausführen
self.approval_queue.complete(task_id)
result = self.agent.process(task["request"])
# Antragsteller benachrichtigen
self.notification_handler.notify_requester(
task_id=task_id,
status="approved",
result=result
)
return result
else:
# Genehmigung abgelehnt
self.approval_queue.reject(task_id, feedback)
# Antragsteller benachrichtigen
self.notification_handler.notify_requester(
task_id=task_id,
status="rejected",
feedback=feedback
)
return {
"status": "rejected",
"reason": feedback or Genehmigung fehlgeschlagen.
}
class ApprovalQueue:
Genehmigungswarteschlange
def __init__(self):
self.queue = {}
self.counter = 0
def add(self, task):
Genehmigungsaufgabe hinzufügen
self.counter += 1
task_id = f"task_{self.counter}"
self.queue[task_id] = {
**task,
"status": "pending"
}
return task_id
def get(self, task_id):
return self.queue.get(task_id)
def complete(self, task_id):
self.queue[task_id]["status"] = "approved"
def reject(self, task_id, reason):
self.queue[task_id]["status"] = "rejected"
self.queue[task_id]["reject_reason"] = reason
Kapitelzusammenfassung
Dieses Kapitel führt in Wissen über die Bewertung, Sicherheit und Ausrichtung von Agenten ein.
BewertungssystemBewerte die Leistung des Agenten anhand mehrdimensionaler Metriken.
SicherheitsbedrohungEinschließlich Prompt-Injection, Jailbreaking, Datenvergiftung usw.
SchutzstrategieGewährleiste Systemsicherheit durch mehrschichtige Sicherheitsprüfungen.
BeobachtbarkeitImplementiere Systemüberwachung durch Logs, Tracing und Metriken.
Guardrails und HITLBiete Kontrolle der Ausgabequalität und manuelle Überprüfungsmöglichkeiten.
Bewertung, Sicherheit und Ausrichtung sind die Grundlage für den Aufbau vertrauenswürdiger Agentensysteme.
In der tatsächlichen Entwicklung erfordert es kontinuierliche Aufmerksamkeit und Verbesserung.
Weitere Erweiterungen