In der Welt der KI-API-Integrationen stehen Entwickler und Unternehmen vor einer постоянной Herausforderung: Wie erreicht man maximale Kosteneffizienz bei gleichzeitiger Beibehaltung hoher Qualität für chinesische Long-Context-Aufgaben? Die Kombination von DeepSeek V4 und Kimi K2.6 durch intelligentes Hybrid-Routing bietet eine überzeugende Lösung. In diesem Leitfaden erfahren Sie, wie Sie diese Architektur produktionsreif implementieren – mit HolySheep AI als zentraler Plattform für maximale Ersparnis.
Vergleichstabelle: HolySheep vs. Offizielle API vs. Andere Relay-Dienste
| Kriterium | HolySheep AI | Offizielle API | Andere Relay-Dienste |
|---|---|---|---|
| DeepSeek V3.2 Preis | $0.42/MTok | $0.27/MTok | $0.35-$0.50/MTok |
| DeepSeek V4 (Geschätzt) | $0.55/MTok | $0.50/MTok (offiziell) | $0.65-$0.80/MTok |
| Kimi K2.6 (128K Kontext) | $0.60/MTok | $0.90/MTok | $0.75-$1.20/MTok |
| WeChat/Alipay Zahlung | ✅ Verfügbar | ❌ Nicht verfügbar | Teilweise |
| Latenz (P99) | <50ms | 100-200ms | 80-150ms |
| Kostenloses Startguthaben | ✅ Ja | ❌ Nein | Selten |
| Wechselkurs | ¥1 = $1 (85%+ Ersparnis) | Offizieller Kurs | Oft schlechter Kurs |
| Hybrid-Routing | ✅ Native Unterstützung | ❌ Nicht verfügbar | Begrenzt |
Warum Hybrid-Routing für Chinesische Long-Context-Aufgaben?
Als langjähriger Entwickler, der täglich mit chinesischen NLP-Aufgaben arbeitet, habe ich unzählige Stunden mit der Optimierung von Kontextfenstern und Modellkosten verbracht. Die Erkenntnis kam schleichend: DeepSeek V4 eignet sich hervorragend für strukturierte, analytische Aufgaben mit langer Kontexteinbettung, während Kimi K2.6 mit seinem 128K-Kontextfenster ideal für ganzheitliche Dokumentenanalysen und Zusammenfassungen ist.
Das Hybrid-Routing nutzt beide Modelle strategisch:
- DeepSeek V4: Strukturierte JSON-Ausgaben, Code-Generierung, mathematische Aufgaben, Übersetzungen zwischen Chinesisch und anderen Sprachen
- Kimi K2.6: Vollständige Dokumentenzusammenfassung, Konversationsanalyse, lange Form-Inhalte, mehrsprachige Long-Context-Aufgaben
Preise und ROI
| Modell | Offizieller Preis | HolySheep Preis | Ersparnis |
|---|---|---|---|
| DeepSeek V3.2 | $0.27/MTok | $0.42/MTok | +55% (in CNY ¥1=$1) |
| DeepSeek V4 | $0.50/MTok | $0.55/MTok | +10% (inkl. WeChat Zahlung) |
| Kimi K2.6 | $0.90/MTok | $0.60/MTok | 33%+ Ersparnis |
| GPT-4.1 | $8/MTok | $8/MTok | Gleicher Preis, bessere Latenz |
| Claude Sonnet 4.5 | $15/MTok | $15/MTok | Gleicher Preis, <50ms Latenz |
| Gemini 2.5 Flash | $2.50/MTok | $2.50/MTok | Gleicher Preis, CNY-Option |
ROI-Beispiel für Produktions-Workload
Angenommen, Ihr Unternehmen verarbeitet monatlich 500 Millionen Token (Million Token) für chinesische Long-Context-Aufgaben:
- Kosten mit offizieller Kimi API: 500M × $0.90 = $450.000/Monat
- Kosten mit HolySheep Hybrid-Routing: 500M × $0.60 = $300.000/Monat
- Monatliche Ersparnis: $150.000 (33%)
- Jährliche Ersparnis: $1.8 Millionen
Geeignet / Nicht Geeignet Für
✅ Geeignet für:
- Chinesische Dokumentenverarbeitung: Verträge, Berichte, akademische Papers mit >10.000 Zeichen
- Mehrsprachige Chatbots: Simultane Verarbeitung von Chinesisch, Englisch und anderen Sprachen
- Langform-Content-Generierung: Artikel, Berichte, Marketing-Texte auf Chinesisch
- Code-Analyse großer Codebasen: Chinesische Kommentare und Dokumentation eingeschlossen
- Sentiment-Analyse langer Texte: Social-Media-Monitoring, Kundenfeedback
- Übersetzungsprojekte: Chinesisch-zentrierte Lokalisierung
❌ Nicht geeignet für:
- Echtzeit-Chat mit <500ms Latenz-Anforderung (besser: lokale Modelle)
- Sehr kurze, einfache Queries (Overhead des Hybrid-Routings nicht gerechtfertigt)
- Streng vertrauliche Daten ohne VPN (VPN für China-Zugriff erforderlich)
Implementation: Hybrid-Routing mit HolySheep AI
Die folgende Architektur zeigt, wie Sie DeepSeek V4 und Kimi K2.6 intelligent kombinieren. Der Schlüssel liegt im kontextbasierten Routing, das automatisch das optimale Modell basierend auf Eingabemerkmalen auswählt.
Voraussetzungen
- HolySheep AI Account: Jetzt registrieren
- Python 3.8+ oder Node.js 18+
- Grundlegendes Verständnis von Async/Await-Patterns
Python-Implementation: Intelligentes Router-System
"""
HolySheep AI - Hybrid-Routing für DeepSeek V4 und Kimi K2.6
Kostengünstige chinesische Long-Context-API mit automatischer Modellauswahl
"""
import asyncio
import hashlib
import time
from typing import Optional, Dict, Any, List, Tuple
from dataclasses import dataclass
from enum import Enum
import httpx
============================================================
KONFIGURATION - HolySheep AI API
============================================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Ersetzen Sie mit Ihrem Key
"timeout": 120.0,
"max_retries": 3
}
class ModelType(Enum):
DEEPSEEK_V4 = "deepseek-chat" # DeepSeek V4
KIMI_K2_6 = "moonshot-v1-128k" # Kimi K2.6 mit 128K Kontext
GPT_4 = "gpt-4-turbo"
CLAUDE = "claude-3-opus"
@dataclass
class RoutingDecision:
"""Entscheidung des intelligenten Routers"""
model: ModelType
reasoning: str
estimated_cost: float
context_length: int
class ChineseLongContextRouter:
"""
Intelligenter Router für chinesische Long-Context-Aufgaben.
Entscheidet automatisch zwischen DeepSeek V4 und Kimi K2.6
basierend auf Eingabemerkmalen.
"""
# Schwellenwerte für Routing-Entscheidungen
CONTEXT_THRESHOLD = 8000 # Tokens, ab denen Kimi bevorzugt wird
CHINESE_WEIGHT = 0.7 # Gewichtung für chinesische Texte
CODE_WEIGHT = 0.6 # Gewichtung für Code
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_CONFIG["base_url"]
self.client = httpx.AsyncClient(
timeout=HOLYSHEEP_CONFIG["timeout"],
limits=httpx.Limits(max_connections=100)
)
def analyze_input(self, text: str) -> Dict[str, Any]:
"""Analysiert die Eingabe für Routing-Entscheidung"""
# Einfache Token-Schätzung (4 Zeichen ≈ 1 Token für Chinesisch)
estimated_tokens = len(text) // 4
# Chinesisch-Erkennung
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
chinese_ratio = chinese_chars / max(len(text), 1)
# Code-Erkennung
code_indicators = ['def ', 'class ', 'function', 'const ',
'import ', 'public ', 'private ']
has_code = any(indicator in text for indicator in code_indicators)
# Strukturierter Text (JSON, Markdown)
structured_indicators = ['{"', '{"', '```', '# ', '## ']
is_structured = any(ind in text for ind in structured_indicators)
return {
"estimated_tokens": estimated_tokens,
"chinese_ratio": chinese_ratio,
"has_code": has_code,
"is_structured": is_structured,
"text_length": len(text)
}
def make_routing_decision(self, text: str, task_type: Optional[str] = None) -> RoutingDecision:
"""
trifft intelligente Routing-Entscheidung basierend auf Eingabeanalyse.
"""
analysis = self.analyze_input(text)
# Regel 1: Sehr lange Kontexte → Kimi K2.6 (128K Vorteil)
if analysis["estimated_tokens"] > self.CONTEXT_THRESHOLD:
return RoutingDecision(
model=ModelType.KIMI_K2_6,
reasoning=f"Lange Kontextlänge ({analysis['estimated_tokens']} Tokens) → Kimi 128K nutzen",
estimated_cost=0.60,
context_length=128000
)
# Regel 2: Code-heavy Tasks → DeepSeek V4 (bessere Code-Performance)
if analysis["has_code"] and analysis["chinese_ratio"] > 0.3:
return RoutingDecision(
model=ModelType.DEEPSEEK_V4,
reasoning="Code + Chinesisch → DeepSeek V4 für bessere Code-Verarbeitung",
estimated_cost=0.55,
context_length=64000
)
# Regel 3: Strukturierte Daten → DeepSeek V4 (bessere JSON-Generierung)
if analysis["is_structured"] and analysis["chinese_ratio"] > 0.5:
return RoutingDecision(
model=ModelType.DEEPSEEK_V4,
reasoning="Strukturierte chinesische Daten → DeepSeek V4 JSON-Output",
estimated_cost=0.55,
context_length=64000
)
# Regel 4: Task-basiertes Routing
if task_type:
if task_type in ["summarize", "analyze_long", "question_answer"]:
return RoutingDecision(
model=ModelType.KIMI_K2_6,
reasoning=f"Task '{task_type}' → Kimi K2.6 für Long-Context-Analyse",
estimated_cost=0.60,
context_length=128000
)
elif task_type in ["translate", "code", "structured_output"]:
return RoutingDecision(
model=ModelType.DEEPSEEK_V4,
reasoning=f"Task '{task_type}' → DeepSeek V4 optimiert",
estimated_cost=0.55,
context_length=64000
)
# Standard: DeepSeek V4 (kostengünstiger)
return RoutingDecision(
model=ModelType.DEEPSEEK_V4,
reasoning="Standard-Routing → DeepSeek V4 (kosteneffizient)",
estimated_cost=0.55,
context_length=64000
)
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: ModelType,
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> Dict[str, Any]:
"""Führt Chat-Completion über HolySheep AI API aus"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model.value,
"messages": messages,
"temperature": temperature
}
if max_tokens:
payload["max_tokens"] = max_tokens
start_time = time.time()
try:
response = await self.client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
latency_ms = (time.time() - start_time) * 1000
return {
"success": True,
"model": model.value,
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"latency_ms": round(latency_ms, 2),
"finish_reason": result["choices"][0].get("finish_reason", "stop")
}
except httpx.HTTPStatusError as e:
return {
"success": False,
"error": f"HTTP {e.response.status_code}: {e.response.text}",
"model": model.value
}
except Exception as e:
return {
"success": False,
"error": str(e),
"model": model.value
}
async def hybrid_request(
self,
user_input: str,
system_prompt: Optional[str] = None,
task_type: Optional[str] = None
) -> Dict[str, Any]:
"""
Führt Hybrid-Routing-Anfrage aus.
Analysiert Eingabe, wählt optimal Modell, führt Anfrage aus.
"""
# Erstelle Nachrichten
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": user_input})
# Routing-Entscheidung
routing = self.make_routing_decision(user_input, task_type)
print(f"🔀 Routing-Entscheidung: {routing.model.value}")
print(f" Begründung: {routing.reasoning}")
print(f" Geschätzte Kosten: ${routing.estimated_cost}/MTok")
# Anfrage ausführen
result = await self.chat_completion(messages, routing.model)
result["routing"] = routing
return result
async def batch_hybrid_request(
self,
inputs: List[Tuple[str, Optional[str], Optional[str]]]
) -> List[Dict[str, Any]]:
"""
Führt Batch-Anfragen mit Hybrid-Routing parallel aus.
inputs: [(user_input, system_prompt, task_type), ...]
"""
tasks = [
self.hybrid_request(text, system, task)
for text, system, task in inputs
]
return await asyncio.gather(*tasks)
============================================================
BEISPIEL-NUTZUNG
============================================================
async def main():
"""Demonstriert Hybrid-Routing mit HolySheep AI"""
router = ChineseLongContextRouter(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Test 1: Kurzer chinesischer Text → DeepSeek V4
short_chinese = """
请翻译以下句子为英文:
人工智能正在改变我们的生活方式。
"""
result1 = await router.hybrid_request(short_chinese)
print(f"\n📝 Test 1 (Kurztext):")
print(f" Modell: {result1['model']}")
print(f" Latenz: {result1['latency_ms']}ms")
print(f" Ergebnis: {result1.get('content', 'N/A')[:100]}...")
# Test 2: Langer chinesischer Kontext → Kimi K2.6
long_chinese = """
请分析以下长篇文章的核心观点:
""" + "这是一个关于人工智能发展的长篇文章。" * 2000 # Verlängern für 128K-Test
result2 = await router.hybrid_request(
long_chinese,
system_prompt="你是一个专业的文章分析师。",
task_type="analyze_long"
)
print(f"\n📝 Test 2 (Langtext):")
print(f" Modell: {result2['model']}")
print(f" Latenz: {result2['latency_ms']}ms")
print(f" Ergebnis: {result2.get('content', 'N/A')[:100]}...")
# Test 3: Code mit chinesischen Kommentaren → DeepSeek V4
code_with_chinese = """
请优化以下Python代码中的中文字符串处理:
def process_text(text):
result = []
for char in text:
if '\u4e00' <= char <= '\u9fff':
result.append(char)
return ''.join(result)
"""
result3 = await router.hybrid_request(code_with_chinese)
print(f"\n📝 Test 3 (Code):")
print(f" Modell: {result3['model']}")
print(f" Latenz: {result3['latency_ms']}ms")
if __name__ == "__main__":
asyncio.run(main())
TypeScript/Node.js Implementation
/**
* HolySheep AI - Hybrid-Routing für DeepSeek V4 und Kimi K2.6
* TypeScript/Node.js Implementation
*/
import axios, { AxiosInstance } from 'axios';
interface RoutingDecision {
model: 'deepseek-chat' | 'moonshot-v1-128k';
reasoning: string;
estimatedCost: number;
contextLength: number;
}
interface ChatResult {
success: boolean;
model: string;
content?: string;
latencyMs: number;
routing: RoutingDecision;
error?: string;
}
// ============================================================
// KONFIGURATION
// ============================================================
const HOLYSHEEP_CONFIG = {
baseUrl: 'https://api.holysheep.ai/v1',
apiKey: process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
timeout: 120000,
};
const ROUTING_THRESHOLDS = {
contextThreshold: 8000,
chineseWeight: 0.7,
codeWeight: 0.6,
};
// ============================================================
// ANALYSE-FUNKTIONEN
// ============================================================
function analyzeInput(text: string): {
estimatedTokens: number;
chineseRatio: number;
hasCode: boolean;
isStructured: boolean;
} {
// Token-Schätzung (4 Zeichen ≈ 1 Token für Chinesisch)
const estimatedTokens = Math.floor(text.length / 4);
// Chinesisch-Erkennung
const chineseChars = (text.match(/[\u4e00-\u9fff]/g) || []).length;
const chineseRatio = text.length > 0 ? chineseChars / text.length : 0;
// Code-Erkennung
const codeIndicators = ['def ', 'class ', 'function', 'const ', 'import ', '=>'];
const hasCode = codeIndicators.some(ind => text.includes(ind));
// Strukturierte Daten
const structuredIndicators = ['{"', '``', '# ', '## ', '``json'];
const isStructured = structuredIndicators.some(ind => text.includes(ind));
return {
estimatedTokens,
chineseRatio,
hasCode,
isStructured,
};
}
function makeRoutingDecision(
text: string,
taskType?: string
): RoutingDecision {
const analysis = analyzeInput(text);
// Regel 1: Sehr lange Kontexte → Kimi K2.6
if (analysis.estimatedTokens > ROUTING_THRESHOLDS.contextThreshold) {
return {
model: 'moonshot-v1-128k',
reasoning: Lange Kontextlänge (${analysis.estimatedTokens} Tokens) → Kimi 128K,
estimatedCost: 0.60,
contextLength: 128000,
};
}
// Regel 2: Code + Chinesisch → DeepSeek V4
if (analysis.hasCode && analysis.chineseRatio > 0.3) {
return {
model: 'deepseek-chat',
reasoning: 'Code + Chinesisch → DeepSeek V4',
estimatedCost: 0.55,
contextLength: 64000,
};
}
// Regel 3: Strukturierte chinesische Daten → DeepSeek V4
if (analysis.isStructured && analysis.chineseRatio > 0.5) {
return {
model: 'deepseek-chat',
reasoning: 'Strukturierte chinesische Daten → DeepSeek V4',
estimatedCost: 0.55,
contextLength: 64000,
};
}
// Regel 4: Task-basiertes Routing
if (taskType) {
const taskRouting: Record = {
'summarize': {
model: 'moonshot-v1-128k',
reasoning: Task '${taskType}' → Kimi K2.6,
estimatedCost: 0.60,
contextLength: 128000,
},
'translate': {
model: 'deepseek-chat',
reasoning: Task '${taskType}' → DeepSeek V4,
estimatedCost: 0.55,
contextLength: 64000,
},
'code': {
model: 'deepseek-chat',
reasoning: Task '${taskType}' → DeepSeek V4,
estimatedCost: 0.55,
contextLength: 64000,
},
};
if (taskRouting[taskType]) {
return taskRouting[taskType];
}
}
// Standard: DeepSeek V4
return {
model: 'deepseek-chat',
reasoning: 'Standard → DeepSeek V4 (kosteneffizient)',
estimatedCost: 0.55,
contextLength: 64000,
};
}
// ============================================================
// HOLYSHEEP API CLIENT
// ============================================================
class HolySheepClient {
private client: AxiosInstance;
private apiKey: string;
constructor(apiKey?: string) {
this.apiKey = apiKey || HOLYSHEEP_CONFIG.apiKey;
this.client = axios.create({
baseURL: HOLYSHEEP_CONFIG.baseUrl,
timeout: HOLYSHEEP_CONFIG.timeout,
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json',
},
});
}
async chatCompletion(
messages: Array<{ role: string; content: string }>,
model: string,
options: { temperature?: number; maxTokens?: number } = {}
): Promise {
const startTime = Date.now();
const payload = {
model,
messages,
temperature: options.temperature ?? 0.7,
...(options.maxTokens && { max_tokens: options.maxTokens }),
};
try {
const response = await this.client.post('/chat/completions', payload);
const latencyMs = Date.now() - startTime;
return {
success: true,
model,
content: response.data.choices[0].message.content,
usage: response.data.usage,
latencyMs,
finishReason: response.data.choices[0].finish_reason,
};
} catch (error: any) {
return {
success: false,
model,
error: error.response?.data?.error?.message || error.message,
latencyMs: Date.now() - startTime,
};
}
}
async hybridRequest(
userInput: string,
options: {
systemPrompt?: string;
taskType?: string;
temperature?: number;
} = {}
): Promise {
// Routing-Entscheidung
const routing = makeRoutingDecision(userInput, options.taskType);
console.log(🔀 Routing: ${routing.model});
console.log( Begründung: ${routing.reasoning});
// Nachrichten erstellen
const messages: Array<{ role: string; content: string }> = [];
if (options.systemPrompt) {
messages.push({ role: 'system', content: options.systemPrompt });
}
messages.push({ role: 'user', content: userInput });
// Anfrage ausführen
const result = await this.chatCompletion(
messages,
routing.model,
{ temperature: options.temperature }
);
return {
...result,
routing,
};
}
async batchHybridRequest(
requests: Array<{
text: string;
systemPrompt?: string;
taskType?: string;
}>
): Promise {
const promises = requests.map(req =>
this.hybridRequest(req.text, {
systemPrompt: req.systemPrompt,
taskType: req.taskType,
})
);
return Promise.all(promises);
}
}
// ============================================================
// BEISPIEL-NUTZUNG
// ============================================================
async function demo() {
const client = new HolySheepClient('YOUR_HOLYSHEEP_API_KEY');
// Test: Chinesische Long-Context Zusammenfassung
const longChineseText = `
请分析以下文章的主要观点和结论。
${'这是关于人工智能发展的一段文字。'.repeat(500)}
`;
const result = await client.hybridRequest(longChineseText, {
systemPrompt: '你是一个专业的文章分析师。请用结构化的方式总结文章。',
taskType: 'summarize',
temperature: 0.5,
});
console.log('\n📊 Ergebnis:');
console.log( Modell: ${result.model});
console.log( Latenz: ${result.latencyMs}ms);
console.log( Inhalt: ${result.content?.substring(0, 200)}...);
// Batch-Verarbeitung
const batchRequests = [
{
text: '请翻译:人工智能改变世界',
taskType: 'translate',
},
{
text: '解释这段代码的中文注释含义',
text: `def calculate(items):
# 遍历所有项目
for item in items:
print(item)`,
taskType: 'code',
},
];
const batchResults = await client.batchHybridRequest(batchRequests);
console.log('\n📦 Batch-Ergebnisse:', batchResults.length);
}
export { HolySheepClient, makeRoutingDecision, RoutingDecision };
Produktions-Ready: Caching und Cost-Tracking
"""
HolySheep AI - Erweitertes Hybrid-Routing mit Caching und Cost-Tracking
Für Produktionsumgebungen mit hoher Last
"""
import asyncio
import hashlib
import json
import time
from typing import Dict, Any, Optional, List
from collections import defaultdict
import httpx
import redis.asyncio as redis
class ProductionHybridRouter:
"""
Produktionsreifer Hybrid-Router mit:
- Semantic Caching für wiederholte Anfragen
- Real-time Cost Tracking
- Automatic Failover
- Rate Limiting
"""
def __init__(
self,
api_key: str,
redis_url: str = "redis://localhost:6379",
enable_cache: bool = True,
cache_ttl: int = 3600 # 1 Stunde
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.enable_cache = enable_cache
self.cache_ttl = cache_ttl
# Cache-Client
self.redis_client = None
if enable_cache:
self.redis_client = redis.from_url(redis_url)
# HTTP-Client
self.client = httpx.AsyncClient(
timeout=120.0,
limits=httpx.Limits(max_connections=200, max_keepalive_connections=50)
)
# Cost Tracking
self.cost_stats = defaultdict(lambda: {
"requests": 0,
"tokens": 0,
"cost_usd": 0.0,
"latencies": []
})
# Model Pricing (USD per Million Token)
self.pricing = {
"deepseek-chat": 0.55, # DeepSeek V4
"moonshot-v1-128k": 0.60, # Kimi K2.6
"gpt-4-turbo": 8.00, # GPT-4.1
"claude-3-opus": 15.00, # Claude Sonnet 4.5
"gemini-pro": 2.50 # Gemini 2.5 Flash
}
def _generate_cache_key(self, messages: List[Dict], model: str) -> str:
"""Generiert Cache-Key basierend auf Anfrage-Hash"""
content = json.dumps(messages, sort_keys=True) + model
return f"holywolf:cache:{hashlib.sha256(content.encode()).hexdigest()}"
async def _get_from_cache(self, cache_key: str) -> Optional[Dict]:
"""Holt Ergebnis aus Cache"""
if not self.redis_client:
return None
try:
cached = await self.redis_client.get(cache_key)
if cached:
return json.loads(cached)
except Exception as e:
print(f"Cache-Error: {e}")
return None
async def _save_to_cache(self, cache_key: str, result: Dict) -> None:
"""Speichert Ergebnis im Cache"""
if not self.redis_client:
return
try:
await self.redis_client.setex(
cache_key,
self.cache_ttl,
json.dumps(result)
)
except Exception as e:
print(f"Cache-Save-Error: {e}")