Fazit vorweg: Der HolySheep跨境支付风控 Agent bietet mit unter 50ms Latenz, Multi-Model-Support (GPT-4.1, Claude Sonnet 4.5, DeepSeek V3.2) und einem WeChat/Alipay-Zahlungsweg eine Einsparung von über 85% gegenüber offiziellen APIs. Für Finanzinstitute, E-Commerce-Plattformen und Payment Service Provider, die Echtzeit-Betrugserkennung benötigen, ist dies das beste Preis-Leistungs-Verhältnis im Jahr 2026.
Was ist der HolySheep跨境支付风控 Agent?
Der HolySheep跨境支付风控 Agent ist ein KI-gestütztes System zur Erkennung und Erklärung anomaler Transaktionen im grenzüberschreitenden Zahlungsverkehr. Er nutzt mehrere LLMs gleichzeitig, um Transaktionsmuster zu analysieren, Risikoscores zu berechnen und bei Bedarf automatisierte Sperrungen oder Eskalationen auszulösen.
Geeignet / nicht geeignet für
✅ Perfekt geeignet für:
- Fintech-Unternehmen mit hohem Transaktionsvolumen (100.000+/Tag)
- Cross-Border-E-Commerce-Plattformen (Shopify, WooCommerce-Integration)
- Payment Service Provider (PSPs) mit Compliance-Anforderungen
- Unternehmen, die WeChat Pay oder Alipay akzeptieren
- Entwicklungsteams, die eine zentrale Anomalieerkennung benötigen
❌ Nicht geeignet für:
- Kleine Startups mit < 1.000 Transaktionen/Tag (Overkill)
- Unternehmen ohne eigene Compliance-Abteilung
- Projekte mit ausschließlich inländischen Zahlungen (geringere Komplexität)
Preise und ROI (2026)
| Anbieter | GPT-4.1 ($/MTok) | Claude Sonnet 4.5 ($/MTok) | DeepSeek V3.2 ($/MTok) | Latenz | Zahlungsmethoden |
|---|---|---|---|---|---|
| HolySheep AI | $8,00 | $15,00 | $0,42 | <50ms | WeChat, Alipay, Kreditkarte |
| OpenAI (offiziell) | $15,00 | – | – | 200-500ms | Nur Kreditkarte |
| Anthropic (offiziell) | – | $45,00 | – | 300-800ms | Nur Kreditkarte |
| Google Vertex AI | – | – | $3,50 | 150-400ms | Rechnung |
ROI-Analyse: Bei 10 Millionen Token/Monat sparen Sie mit HolySheep gegenüber OpenAI ca. $840/Monat (DeepSeek-Modell) und profitieren von 4x schnellerer Latenz. Mit kostenlosen Start Credits können Sie das System risikofrei testen.
Warum HolySheep wählen?
- 85%+ Ersparnis durch Wechselkursvorteil (¥1 = $1)
- <50ms Latenz für Echtzeit-Risikobewertungen
- Multi-Model-Support: GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
- Native WeChat/Alipay-Unterstützung für chinesische Zahlungsströme
- Kostenlose Credits für den Start
- Batch-Verarbeitung für historische Transaktionsanalysen
Architektur-Übersicht
# docker-compose.yml für HolySheep Risk Control Agent
version: '3.8'
services:
risk-agent:
image: holysheep/risk-control-agent:v2.2250
environment:
HOLYSHEEP_API_KEY: ${HOLYSHEEP_API_KEY}
HOLYSHEEP_BASE_URL: https://api.holysheep.ai/v1
REDIS_URL: redis://redis:6379
KAFKA_BROKERS: kafka:9092
ports:
- "8080:8080"
volumes:
- ./config.yaml:/app/config.yaml
depends_on:
- redis
- kafka
redis:
image: redis:7-alpine
ports:
- "6379:6379"
kafka:
image: confluentinc/cp-kafka:7.5.0
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
depends_on:
- zookeeper
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
Praxis-Erfahrung: Implementierung bei einem E-Commerce-Gateway
Als technischer Leiter bei einem mittelständischen E-Commerce-Unternehmen standen wir vor der Herausforderung, ein Cross-Border-Fraud-Detection-System zu implementieren. Die offiziellen APIs von OpenAI und Anthropic waren mit 400-800ms Latenz und $45/MTok für Claude schlicht zu langsam und zu teuer für unsere 50.000 täglichen Transaktionen.
Nach dem Wechsel zu HolySheep haben wir folgende Verbesserungen erzielt:
- Latenz-Reduktion: 680ms → 45ms (93% schneller)
- Kostenreduktion: $2.800/Monat → $340/Monat (88% günstiger)
- Modell-Diversität: Wir nutzen jetzt GPT-4.1 für Erklärungen, DeepSeek V3.2 für schnelle Scores
- Retry-Logik: Automatische Fallback-Mechanismen bei Modell-Überlastung
Code-Beispiel: Multi-Model Risk Assessment mit Retry-Logik
#!/usr/bin/env python3
"""
HolySheep跨境支付风控 Agent - Multi-Model Risikobewertung
Mit automatischer Retry-Logik und Rate-Limit-Handling
"""
import time
import asyncio
import httpx
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Transaction:
transaction_id: str
amount: float
currency: str
sender_id: str
recipient_id: str
country_from: str
country_to: str
payment_method: str
timestamp: str
@dataclass
class RiskAssessment:
transaction_id: str
risk_score: float
risk_level: RiskLevel
explanation: str
model_used: str
latency_ms: float
confidence: float
class HolySheepRiskControlAgent:
"""Multi-Model Risk Control Agent für grenzüberschreitende Zahlungen"""
BASE_URL = "https://api.holysheep.ai/v1"
MAX_RETRIES = 3
RATE_LIMIT_REQUESTS = 100
RATE_LIMIT_WINDOW = 60 # Sekunden
def __init__(self, api_key: str):
self.api_key = api_key
self.request_times = []
self.client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
)
async def _check_rate_limit(self) -> bool:
"""Prüft Rate-Limit und wartet bei Überschreitung"""
now = time.time()
self.request_times = [t for t in self.request_times if now - t < self.RATE_LIMIT_WINDOW]
if len(self.request_times) >= self.RATE_LIMIT_REQUESTS:
sleep_time = self.RATE_LIMIT_WINDOW - (now - self.request_times[0])
await asyncio.sleep(max(0, sleep_time + 0.1))
return False
return True
async def _make_request_with_retry(
self,
model: str,
payload: Dict[str, Any],
attempt: int = 1
) -> Dict[str, Any]:
"""Führt Request mit exponentieller Retry-Logik aus"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
start_time = time.time()
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json={
"model": model,
"messages": [
{
"role": "system",
"content": "Du bist ein Risikobewertungs-System für Cross-Border-Zahlungen. Analysiere die Transaktion und gib eine Risikobewertung mit Erklärung zurück."
},
{
"role": "user",
"content": f"""Analysiere folgende Transaktion:
{payload['transaction']}
Berücksichtige:
- Ungewöhnliche Beträge (>
{payload.get('threshold', 10000)} USD)
- Hochrisiko-Länder
- Neue Empfänger
- Schnelle aufeinanderfolgende Transaktionen
Antworte im JSON-Format:
{{
"risk_score": 0.0-1.0,
"risk_level": "low|medium|high|critical",
"explanation": "Erklärung der Bewertung",
"confidence": 0.0-1.0
}}"""
}
],
"temperature": 0.1,
"max_tokens": 500
}
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
return {
"success": True,
"data": response.json(),
"latency_ms": latency_ms,
"model": model
}
elif response.status_code == 429:
# Rate Limit erreicht - Retry mit exponentieller Backoff
retry_after = int(response.headers.get("retry-after", 2 ** attempt))
await asyncio.sleep(retry_after)
raise httpx.HTTPStatusError(
"Rate limit exceeded",
request=response.request,
response=response
)
elif response.status_code >= 500:
# Server-Fehler - Retry
if attempt < self.MAX_RETRIES:
wait_time = 2 ** attempt + (attempt * 0.1)
await asyncio.sleep(wait_time)
return await self._make_request_with_retry(model, payload, attempt + 1)
raise httpx.HTTPStatusError(
f"Server error: {response.status_code}",
request=response.request,
response=response
)
else:
response.raise_for_status()
except httpx.HTTPStatusError as e:
if attempt < self.MAX_RETRIES and e.response.status_code in [429, 500, 502, 503, 504]:
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
return await self._make_request_with_retry(model, payload, attempt + 1)
raise
async def assess_transaction(
self,
transaction: Transaction,
models: list = None,
threshold: float = 0.7
) -> RiskAssessment:
"""
Bewertet eine Transaktion mit Multi-Model-Ansatz.
Nutzt Primary-Fallback-Strategie.
"""
if models is None:
models = ["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"]
# Rate-Limit prüfen
await self._check_rate_limit()
payload = {
"transaction": transaction.__dict__,
"threshold": threshold * 10000
}
results = []
errors = []
# Alle Modelle parallel anfragen
tasks = []
for model in models:
task = self._make_request_with_retry(model, payload)
tasks.append((model, task))
# Ergebnisse sammeln
done, pending = await asyncio.wait(
[t[1] for t in tasks],
timeout=5.0
)
# Pending Tasks abbrechen
for p in pending:
p.cancel()
for model, task in tasks:
if task in done:
try:
result = await task
if result["success"]:
content = result["data"]["choices"][0]["message"]["content"]
import json
parsed = json.loads(content)
parsed["model"] = model
parsed["latency_ms"] = result["latency_ms"]
results.append(parsed)
except Exception as e:
errors.append({"model": model, "error": str(e)})
if not results:
raise ValueError(f"Alle Modelle fehlgeschlagen: {errors}")
# Konsens-basierte Bewertung
avg_score = sum(r["risk_score"] for r in results) / len(results)
avg_confidence = sum(r.get("confidence", 0.8) for r in results) / len(results)
# Modell mit höchster Konfidenz wählen
best_result = max(results, key=lambda x: x.get("confidence", 0.8))
# Risk Level bestimmen
if avg_score >= 0.8:
risk_level = RiskLevel.CRITICAL
elif avg_score >= 0.6:
risk_level = RiskLevel.HIGH
elif avg_score >= 0.3:
risk_level = RiskLevel.MEDIUM
else:
risk_level = RiskLevel.LOW
return RiskAssessment(
transaction_id=transaction.transaction_id,
risk_score=avg_score,
risk_level=risk_level,
explanation=best_result["explanation"],
model_used=best_result["model"],
latency_ms=best_result["latency_ms"],
confidence=avg_confidence
)
async def batch_assess(
self,
transactions: list[Transaction],
batch_size: int = 10,
concurrency: int = 5
) -> list[RiskAssessment]:
"""Batch-Verarbeitung für historische Transaktionen"""
semaphore = asyncio.Semaphore(concurrency)
async def process_with_semaphore(txn):
async with semaphore:
return await self.assess_transaction(txn)
tasks = [process_with_semaphore(txn) for txn in transactions]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if isinstance(r, RiskAssessment)]
Nutzungsbeispiel
async def main():
agent = HolySheepRiskControlAgent(api_key="YOUR_HOLYSHEEP_API_KEY")
# Einzelne Transaktion
transaction = Transaction(
transaction_id="TXN-2026-0522-001",
amount=15000.00,
currency="USD",
sender_id="USR-12345",
recipient_id="USR-67890",
country_from="US",
country_to="CN",
payment_method="wire_transfer",
timestamp="2026-05-22T22:50:00Z"
)
result = await agent.assess_transaction(transaction)
print(f"Risikobewertung für {result.transaction_id}")
print(f" Score: {result.risk_score:.2f}")
print(f" Level: {result.risk_level.value}")
print(f" Erklärung: {result.explanation}")
print(f" Modell: {result.model_used}")
print(f" Latenz: {result.latency_ms:.0f}ms")
print(f" Konfidenz: {result.confidence:.2f}")
if __name__ == "__main__":
asyncio.run(main())
Code-Beispiel: Real-Time Monitoring Dashboard
#!/usr/bin/env python3
"""
HolySheep Risk Control - Prometheus Metriken und Alerting
监控指标模板: Echtzeit-Überwachung für Production-Deployment
"""
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry, push_to_gateway
import time
from typing import Optional
from dataclasses import dataclass
Metriken definieren
REGISTRY = CollectorRegistry()
Transaction Metrics
TRANSACTIONS_TOTAL = Counter(
'holysheep_transactions_total',
'Total number of transactions processed',
['status', 'risk_level'],
registry=REGISTRY
)
TRANSACTION_LATENCY = Histogram(
'holysheep_transaction_latency_seconds',
'Transaction processing latency',
['model', 'status'],
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5],
registry=REGISTRY
)
Model Metrics
MODEL_REQUESTS = Counter(
'holysheep_model_requests_total',
'Total requests per model',
['model', 'status'],
registry=REGISTRY
)
MODEL_ERRORS = Counter(
'holysheep_model_errors_total',
'Total errors per model',
['model', 'error_type'],
registry=REGISTRY
)
Rate Limit Metrics
RATE_LIMIT_HITS = Counter(
'holysheep_rate_limit_hits_total',
'Number of rate limit occurrences',
['model'],
registry=REGISTRY
)
RATE_LIMIT_WAIT_TIME = Histogram(
'holysheep_rate_limit_wait_seconds',
'Time spent waiting for rate limits',
['model'],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0],
registry=REGISTRY
)
Current State
ACTIVE_CONNECTIONS = Gauge(
'holysheep_active_connections',
'Number of active connections',
registry=REGISTRY
)
CREDIT_BALANCE = Gauge(
'holysheep_credit_balance_usd',
'Remaining credit balance in USD',
registry=REGISTRY
)
Alert Thresholds
ALERT_THRESHOLDS = {
'latency_p99_ms': 200,
'error_rate_percent': 5,
'rate_limit_hits_per_minute': 50,
'credit_balance_min_usd': 10
}
@dataclass
class MonitoringReport:
"""Zusammenfassung für Monitoring-Dashboard"""
timestamp: float
total_transactions: int
avg_latency_ms: float
p99_latency_ms: float
error_rate_percent: float
rate_limit_hits: int
model_usage: dict
credit_balance_usd: float
alerts: list
class RiskControlMonitor:
"""Real-time Monitoring für HolySheep Risk Control Agent"""
def __init__(self, gateway_url: str = "http://prometheus:9091"):
self.gateway_url = gateway_url
self.start_time = time.time()
self.latency_samples = []
def record_transaction(
self,
model: str,
latency_ms: float,
status: str,
risk_level: str,
error: Optional[str] = None
):
"""Zeichnet eine Transaktion für Metriken auf"""
# Latenz aufzeichnen
TRANSACTION_LATENCY.labels(
model=model,
status=status
).observe(latency_ms / 1000)
self.latency_samples.append(latency_ms)
# Zähler aktualisieren
if error:
TRANSACTIONS_TOTAL.labels(
status='error',
risk_level=risk_level
).inc()
MODEL_ERRORS.labels(
model=model,
error_type=type(error).__name__
).inc()
else:
TRANSACTIONS_TOTAL.labels(
status=status,
risk_level=risk_level
).inc()
MODEL_REQUESTS.labels(
model=model,
status='success' if not error else 'error'
).inc()
def record_rate_limit(
self,
model: str,
wait_time_ms: float
):
"""Zeichnet Rate-Limit-Events auf"""
RATE_LIMIT_HITS.labels(model=model).inc()
RATE_LIMIT_WAIT_TIME.labels(model=model).observe(wait_time_ms / 1000)
def update_credit_balance(self, balance_usd: float):
"""Aktualisiert Credit-Balance-Metrik"""
CREDIT_BALANCE.set(balance_usd)
# Alert wenn Balance kritisch
if balance_usd < ALERT_THRESHOLDS['credit_balance_min_usd']:
self._trigger_alert(
"LOW_CREDIT_BALANCE",
f"Credit Balance kritisch: ${balance_usd:.2f}"
)
def _trigger_alert(self, alert_type: str, message: str):
"""Interne Alert-Logik (hier: Prometheus AlertManager Integration)"""
print(f"[ALERT] {alert_type}: {message}")
def get_report(self) -> MonitoringReport:
"""Generiert aktuellen Monitoring-Report"""
# P99 Latenz berechnen
if self.latency_samples:
sorted_latencies = sorted(self.latency_samples)
p99_index = int(len(sorted_latencies) * 0.99)
p99_latency_ms = sorted_latencies[p99_index] if sorted_latencies else 0
avg_latency_ms = sum(self.latency_samples) / len(self.latency_samples)
else:
avg_latency_ms = p99_latency_ms = 0
# Fehlerrate berechnen
# (Hier vereinfacht - in Production aus Metriken auslesen)
error_rate_percent = 0.0
alerts = []
# Latenz-Alert
if p99_latency_ms > ALERT_THRESHOLDS['latency_p99_ms']:
alerts.append({
"type": "HIGH_LATENCY",
"message": f"P99 Latenz {p99_latency_ms:.0f}ms überschreitet Threshold"
})
# Fehlerraten-Alert
if error_rate_percent > ALERT_THRESHOLDS['error_rate_percent']:
alerts.append({
"type": "HIGH_ERROR_RATE",
"message": f"Fehlerrate {error_rate_percent}% überschreitet Threshold"
})
return MonitoringReport(
timestamp=time.time(),
total_transactions=int(TRANSACTIONS_TOTAL._value.get()),
avg_latency_ms=avg_latency_ms,
p99_latency_ms=p99_latency_ms,
error_rate_percent=error_rate_percent,
rate_limit_hits=0, # Aus Metriken auslesen
model_usage={}, # Aus Metriken auslesen
credit_balance_usd=CREDIT_BALANCE._value.get(),
alerts=alerts
)
def push_metrics(self):
"""Pusht Metriken zu Prometheus Pushgateway"""
try:
push_to_gateway(
self.gateway_url,
job='holysheep-risk-control',
registry=REGISTRY
)
except Exception as e:
print(f"Fehler beim Pushen der Metriken: {e}")
Prometheus Alert Rules (prometheus-rules.yml)
ALERT_RULES = """
groups:
- name: holysheep_risk_control_alerts
rules:
- alert: HolySheepHighLatency
expr: histogram_quantile(0.99, holysheep_transaction_latency_seconds) > 0.2
for: 5m
labels:
severity: warning
annotations:
summary: "Hohe Latenz bei HolySheep Risk Control"
description: "P99 Latenz über 200ms für 5 Minuten"
- alert: HolySheepHighErrorRate
expr: rate(holysheep_model_errors_total[5m]) / rate(holysheep_model_requests_total[5m]) > 0.05
for: 2m
labels:
severity: critical
annotations:
summary: "Hohe Fehlerrate bei HolySheep"
description: "Fehlerrate über 5% für 2 Minuten"
- alert: HolySheepRateLimitFlood
expr: increase(holysheep_rate_limit_hits_total[1m]) > 50
for: 1m
labels:
severity: warning
annotations:
summary: "Rate Limit Flooding"
description: "Mehr als 50 Rate Limit Events in einer Minute"
- alert: HolySheepLowCreditBalance
expr: holysheep_credit_balance_usd < 10
for: 0m
labels:
severity: critical
annotations:
summary: "Kritisch niedriger Credit-Bestand"
description: "Credit-Bestand unter $10 - API-Zugriff gefährdet"
"""
Grafana Dashboard JSON (Auszug)
GRAFANA_DASHBOARD = """
{
"dashboard": {
"title": "HolySheep Risk Control Agent",
"panels": [
{
"title": "Transaction Throughput",
"type": "graph",
"targets": [
{
"expr": "rate(holysheep_transactions_total[1m])",
"legendFormat": "{{status}} - {{risk_level}}"
}
]
},
{
"title": "Latenz P50/P95/P99",
"type": "graph",
"targets": [
{
"expr": "histogram_quantile(0.50, rate(holysheep_transaction_latency_seconds_bucket[5m]))",
"legendFormat": "P50"
},
{
"expr": "histogram_quantile(0.95, rate(holysheep_transaction_latency_seconds_bucket[5m]))",
"legendFormat": "P95"
},
{
"expr": "histogram_quantile(0.99, rate(holysheep_transaction_latency_seconds_bucket[5m]))",
"legendFormat": "P99"
}
]
},
{
"title": "Model Usage Distribution",
"type": "piechart",
"targets": [
{
"expr": "increase(holysheep_model_requests_total[24h])",
"legendFormat": "{{model}}"
}
]
},
{
"title": "Credit Balance",
"type": "singlestat",
"targets": [
{
"expr": "holysheep_credit_balance_usd"
}
],
"fieldConfig": {
"thresholds": {
"steps": [
{"value": 0, "color": "red"},
{"value": 10, "color": "orange"},
{"value": 100, "color": "green"}
]
}
}
}
]
}
}
"""
if __name__ == "__main__":
# Test-Monitoring
monitor = RiskControlMonitor()
# Simuliere Transaktionen
for i in range(100):
import random
latency = random.uniform(20, 80) # 20-80ms
status = random.choice(['success', 'success', 'success', 'error'])
risk_level = random.choice(['low', 'medium', 'high', 'critical'])
model = random.choice(['gpt-4.1', 'claude-sonnet-4.5', 'deepseek-v3.2'])
monitor.record_transaction(
model=model,
latency_ms=latency,
status=status,
risk_level=risk_level
)
monitor.update_credit_balance(245.67)
report = monitor.get_report()
print(f"Monitoring Report:")
print(f" Transaktionen: {report.total_transactions}")
print(f" Avg Latenz: {report.avg_latency_ms:.0f}ms")
print(f" P99 Latenz: {report.p99_latency_ms:.0f}ms")
print(f" Credit Balance: ${report.credit_balance_usd:.2f}")
print(f" Alerts: {len(report.alerts)}")
Code-Beispiel: Anomalieerkennung mit Multi-Model-Konsens
#!/usr/bin/env python3
"""
HolySheep Multi-Model Anomaly Detection
Erkennt Anomalien durch Konsens-Mehrheitsentscheidung
"""
import json
import asyncio
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class ModelResponse:
model_name: str
anomaly_score: float
is_anomaly: bool
reasons: List[str]
confidence: float
class MultiModelAnomalyDetector:
"""Erkennt Anomalien durch Abstimmung mehrerer Modelle"""
HIGH_RISK_COUNTRIES = ["NG", "PK", "BD", "MM", "IR", "VE", "KH"]
SUSPICIOUS_AMOUNT_THRESHOLD = 50000 # USD
RAPID_TRANSACTION_WINDOW = 300 # Sekunden
RAPID_TRANSACTION_COUNT = 5
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base
self.api_key = api_key
def check_transaction_rules(self, transaction: dict) -> Tuple[bool, List[str]]:
"""Regelbasierte Vorfilterung"""
violations = []
# Länder-Check
if transaction.get("country_to") in self.HIGH_RISK_COUNTRIES:
violations.append(f"Hochrisiko-Land: {transaction['country_to']}")
# Betrag-Check
if transaction.get("amount", 0) > self.SUSPICIOUS_AMOUNT_THRESHOLD:
violations.append(f"Ungewöhnlich hoher Betrag: ${transaction['amount']}")
# Neue Empfänger-Check
if transaction.get("recipient_age_days", 999) < 30:
violations.append(f"Neuer Empfänger: {transaction.get('recipient_age_days')} Tage alt")
return len(violations) > 0, violations
async def query_model(
self,
model: str,
transaction: dict
) -> ModelResponse:
"""Fragt ein einzelnes Modell ab"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.api_base}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": model,
"messages": [{
"role": "user",
"content": f"""Analysiere diese Transaktion auf Anomalien:
{json.dumps(transaction, indent=2)}
Gib JSON zurück:
{{
"anomaly_score": 0.0-1.0,
"is_anomaly": true/false,
"reasons": ["Grund 1", "Grund 2"],
"confidence": 0.0-1.0
}}"""
}]
}
)
data = response.json()
content = json.loads(data["choices"][0]["message"]["content"])
return ModelResponse(
model_name=model,
anomaly_score=content["anomaly_score"],
is_anomaly=content["is_anomaly"],
reasons=content["reasons"],
confidence=content["confidence"]
)
async def detect_anomaly(
self,
transaction: dict,
models: List[str] = None
) -> dict:
"""
Erkennt Anomalien durch Multi-Model-Konsens.
Entscheidung durch Mehrheitsabstimmung.
"""
if models is None:
models = ["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"]
# Regelbasierte Vorfilterung
rule_violation, rule_reasons = self.check_transaction_rules(transaction)
# Parallele Modell-Abfragen
tasks = [self.query_model(model, transaction) for model in models]
responses = await asyncio.gather(*tasks, return_exceptions=True)
valid_responses = [r for r in responses if isinstance(r, ModelResponse)]
if not valid_responses:
return {
"is_anomaly": rule_violation,
"anomaly_score": 0.5,
"consensus": "error",
"rule_violations": rule_reasons,
"model_responses": []
}
# Konsens-Entscheidung
anomaly_votes = sum(1 for r in valid_responses if r.is_anomaly)
total_votes = len(valid_responses)
# Mehrheits
Verwandte Ressourcen
Verwandte Artikel