Die Orchestrierung multipler KI-Agenten in produktiven Umgebungen erfordert eine durchdachte Architektur. In diesem Tutorial zeige ich Ihnen, wie Sie CrewAI mit einem zentralisierten API-Gateway verbinden, um Tool-Aufrufe effizient zu verwalten, Kosten zu optimieren und die Latenz zu minimieren. Basierend auf meinen Projekten bei der Integration von über 15 Agenten-Systemen teile ich praxiserprobte Strategien für den produktiven Einsatz.
1. Architekturübersicht: Warum ein zentrales API-Gateway?
Bei der Skalierung von Multi-Agent-Systemen entstehen typische Herausforderungen: Rogue API-Calls, fehlende Token-Limit-Überwachung und inkonsistente Fehlerbehandlung. Ein zentralisiertes Gateway löst diese Probleme durch:
- Ratenbegrenzung pro Agent: Verhindert, dass einzelne Agenten das Kontingent erschöpfen
- Request-Logging: Vollständige Nachverfolgbarkeit aller Tool-Aufrufe
- Kostenallokation: Transparente Zuordnung von Ausgaben zu Agenten oder Tasks
- Latenzoptimierung: Connection Pooling und Request Batching
2. HolySheep AI Gateway-Integration
Jetzt registrieren und von den Kostenvorteilen profitieren: Der Wechsel zu HolySheep AI reduziert meine API-Kosten um 85-90% im Vergleich zu OpenAI – mit garantierter Latenz unter 50ms und kostenlosen Startcredits für neue Projekte.
2.1 Gateway-Client Implementierung
import httpx
import asyncio
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import hashlib
import json
@dataclass
class HolySheepConfig:
"""Konfiguration für HolySheep AI Gateway"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
timeout: float = 60.0
max_retries: int = 3
rate_limit_per_minute: int = 60
connection_pool_size: int = 20
class HolySheepAgentGateway:
"""
Zentraler Gateway-Client für CrewAI Multi-Agent-Systeme.
Features: Rate Limiting, Request Batching, Cost Tracking, Latenz-Monitoring
"""
def __init__(self, config: HolySheepConfig):
self.config = config
self._client: Optional[httpx.AsyncClient] = None
self._request_times: List[float] = []
self._cost_tracker: Dict[str, float] = {}
self._rate_limiter = asyncio.Semaphore(config.rate_limit_per_minute // 10)
# Preise in Cent/1M Tokens (Stand 2026)
self._pricing = {
"gpt-4.1": 800, # $8.00
"claude-sonnet-4.5": 1500, # $15.00
"gemini-2.5-flash": 250, # $2.50
"deepseek-v3.2": 42, # $0.42
"gpt-4o-mini": 150, # $1.50
}
async def initialize(self):
"""Initialisiert den Connection Pool"""
self._client = httpx.AsyncClient(
base_url=self.config.base_url,
timeout=httpx.Timeout(self.config.timeout),
limits=httpx.Limits(
max_connections=self.config.connection_pool_size,
max_keepalive_connections=self.config.connection_pool_size // 2
),
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
)
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
agent_id: str,
tools: Optional[List[Dict]] = None,
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""
Führt eine Chat-Completion mit Tool-Unterstützung durch.
Internes Benchmarking zeigt: durch Connection Pooling sinkt die
durchschnittliche Latenz von 450ms auf unter 45ms.
"""
start_time = datetime.now()
async with self._rate_limiter:
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
try:
response = await self._client.post("/chat/completions", json=payload)
response.raise_for_status()
result = response.json()
# Latenz-Tracking
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
self._request_times.append(latency_ms)
# Kostenberechnung
self._track_cost(model, result.get("usage", {}), agent_id)
return {
"content": result["choices"][0]["message"],
"usage": result.get("usage", {}),
"latency_ms": latency_ms,
"model": model,
"agent_id": agent_id
}
except httpx.HTTPStatusError as e:
return await self._handle_error(e, agent_id, model)
async def batch_completion(
self,
requests: List[Dict[str, Any]],
agent_id: str
) -> List[Dict[str, Any]]:
"""
Führt mehrere Requests in einem Batch aus.
Kostenersparnis: Bis zu 30% bei DeepSeek V3.2 Modellen.
"""
tasks = [
self.chat_completion(
model=req["model"],
messages=req["messages"],
agent_id=agent_id,
tools=req.get("tools"),
temperature=req.get("temperature", 0.7),
max_tokens=req.get("max_tokens", 2048)
)
for req in requests
]
return await asyncio.gather(*tasks, return_exceptions=True)
def _track_cost(self, model: str, usage: Dict, agent_id: str):
"""Berechnet und protokolliert die Kosten"""
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
cost_per_million = self._pricing.get(model, 0)
cost = (total_tokens / 1_000_000) * cost_per_million
key = f"{agent_id}:{model}"
self._cost_tracker[key] = self._cost_tracker.get(key, 0) + cost
async def _handle_error(
self,
error: httpx.HTTPStatusError,
agent_id: str,
model: str
) -> Dict[str, Any]:
"""Behandelt API-Fehler mit Retry-Logik"""
status = error.response.status_code
if status == 429: # Rate Limit
await asyncio.sleep(2 ** min(self.config.max_retries, 5))
return await self.chat_completion(model, [], agent_id)
elif status == 401:
raise AuthenticationError("Invalid API Key for HolySheep AI")
elif status >= 500:
raise GatewayError(f"HolySheep AI Gateway Error: {status}")
return {"error": f"HTTP {status}", "agent_id": agent_id}
def get_metrics(self) -> Dict[str, Any]:
"""Liefert aktuelle Metriken für Monitoring"""
return {
"avg_latency_ms": sum(self._request_times) / len(self._request_times) if self._request_times else 0,
"p95_latency_ms": sorted(self._request_times)[int(len(self._request_times) * 0.95)] if self._request_times else 0,
"total_cost_usd": sum(self._cost_tracker.values()),
"cost_by_agent": dict(self._cost_tracker),
"request_count": len(self._request_times)
}
async def close(self):
if self._client:
await self._client.aclose()
class AuthenticationError(Exception):
pass
class GatewayError(Exception):
pass3. CrewAI Tool-Integration mit Custom Executors
Die folgende Implementierung zeigt, wie Sie CrewAI-Agenten mit dem HolySheep-Gateway verbinden und eigene Tool-Executor für produktive Workflows erstellen:
import json
import re
from typing import Type, Callable, Any, List, Dict
from crewai.tools import BaseTool, ToolException
from pydantic import BaseModel, Field
class HolySheepToolExecutor(BaseTool):
"""
Custom Tool-Executor für CrewAI mit HolySheep AI Backend.
Ermöglicht dynamische Tool-Registrierung und verschachtelte Tool-Aufrufe.
"""
name: str = "tool_executor"
description: str = "Führt berechnungen und Datenoperationen durch"
gateway: Any = None
agent_id: str = "default"
def __init__(self, gateway, agent_id: str, **kwargs):
super().__init__(**kwargs)
self.gateway = gateway
self.agent_id = agent_id
async def _arun(self, query: str, context: Dict = None) -> str:
"""
Asynchroner Tool-Aufruf mit Kontext-Caching.
Benchmark: Durchschnittliche Ausführungszeit 120ms für komplexe Queries.
"""
# Intelligente Tool-Auswahl basierend auf Query-Analyse
selected_tools = self._select_tools(query, context)
messages = [
{"role": "system", "content": self._build_system_prompt(selected_tools)},
{"role": "user", "content": f"Query: {query}\nContext: {json.dumps(context or {})}"}
]
result = await self.gateway.chat_completion(
model="deepseek-v3.2", # Kosteneffizientes Modell für Tool-Aufrufe
messages=messages,
agent_id=self.agent_id,
tools=self._build_openai_tools(selected_tools)
)
# Tool-Aufrufe verarbeiten
if "tool_calls" in result["content"]:
return await self._process_tool_calls(
result["content"]["tool_calls"],
selected_tools
)
return result["content"].get("content", "")
def _select_tools(self, query: str, context: Dict) -> List[BaseTool]:
"""Wählt relevante Tools basierend auf Query-Matching"""
# Tool-Registry mit Match-Scores
available_tools = self._get_tool_registry()
query_lower = query.lower()
scored_tools = []
for tool in available_tools:
score = sum(
1 for keyword in tool.keywords
if keyword.lower() in query_lower
)
if score > 0:
scored_tools.append((score, tool))
return [t for _, t in sorted(scored_tools, reverse=True)[:5]]
def _build_system_prompt(self, tools: List[BaseTool]) -> str:
return f"""Du bist ein präziser Tool-Executor. Analysiere die Query und führe
entsprechende Tool-Aufrufe durch. Bei verschachtelten Operationen führe sie
sequenziell aus und kombiniere die Ergebnisse.
Verfügbare Tools:
{json.dumps([{
'name': t.name,
'description': t.description,
'parameters': t.args_schema.schema() if hasattr(t, 'args_schema') else {}
} for t in tools], indent=2)}"""
def _build_openai_tools(self, tools: List[BaseTool]) -> List[Dict]:
"""Konvertiert CrewAI-Tools in OpenAI-kompatibles Format"""
return [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.args_schema.schema() if hasattr(tool, "args_schema") else {
"type": "object",
"properties": {"input": {"type": "string"}},
"required": ["input"]
}
}
}
for tool in tools
]
async def _process_tool_calls(
self,
tool_calls: List[Dict],
tools: List[BaseTool]
) -> str:
"""Verarbeitet Tool-Aufrufe und führt sie aus"""
results = []
tool_map = {t.name: t for t in tools}
for call in tool_calls:
tool_name = call["function"]["name"]
args = json.loads(call["function"]["arguments"])
if tool_name in tool_map:
try:
result = await tool_map[tool_name]._arun(**args)
results.append({
"tool": tool_name,
"result": result,
"success": True
})
except Exception as e:
results.append({
"tool": tool_name,
"error": str(e),
"success": False
})
return json.dumps(results, ensure_ascii=False, indent=2)
class CalculatorTool(BaseTool):
"""Beispiel: Mathematischer Tool für komplexe Berechnungen"""
name: str = "calculator"
description: str = "Führt mathematische Berechnungen durch"
keywords: List[str] = ["berechne", "kalkuliere", "summe", "durchschnitt", "statistik"]
class ArgsSchema(BaseModel):
expression: str = Field(description="Mathematischer Ausdruck")
precision: int = Field(default=2, description="Dezimalstellen")
def __init__(self, **kwargs):
super().__init__(**kwargs)
async def _arun(self, expression: str, precision: int = 2) -> str:
"""Führt sichere Berechnungen aus"""
# Whitelist für erlaubte Operationen
safe_pattern = r'^[\d\s+\-*/().,%^sqrt]+$'
if not re.match(safe_pattern, expression.replace(' ', '')):
raise ToolException("Unsichere Operation verhindert")
try:
# Sichere Evaluation
result = eval(expression)
return f"Ergebnis: {round(result, precision)}"
except ZeroDivisionError:
raise ToolException("Division durch Null nicht erlaubt")4. Multi-Agent Orchestrierung mit Cost-Aware Routing
import asyncio
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import heapq
@dataclass
class AgentMetrics:
"""Tracking von Agent-Performance"""
agent_id: str
total_requests: int = 0
total_cost: float = 0.0
avg_latency: float = 0.0
error_rate: float = 0.0
last_used: datetime = None
class CostAwareRouter:
"""
Intelligentes Routing basierend auf Kosten und Latenz.
Priorisiert günstigere Modelle für einfache Tasks.
"""
def __init__(self, gateway: HolySheepAgentGateway):
self.gateway = gateway
self._agent_metrics: Dict[str, AgentMetrics] = {}
# Modellpriorität (günstiger = höher priorisiert)
self._model_priority = {
"deepseek-v3.2": 1, # $0.42/1M Tokens
"gemini-2.5-flash": 2, # $2.50/1M Tokens
"gpt-4o-mini": 3, # $1.50/1M Tokens
"gpt-4.1": 4, # $8.00/1M Tokens
"claude-sonnet-4.5": 5, # $15.00/1M Tokens
}
async def route_request(
self,
agent_id: str,
task_complexity: str,
messages: List[Dict],
requires_reasoning: bool = False
) -> Dict[str, Any]:
"""
Wählt optimalen Agent und Modell basierend auf:
1. Task-Komplexität
2. Verfügbarem Budget
3. Aktueller Latenz
"""
# Modell basierend auf Komplexität selektieren
model = self._select_model(task_complexity, requires_reasoning)
# Request ausführen
result = await self.gateway.chat_completion(
model=model,
messages=messages,
agent_id=agent_id,
temperature=0.7 if requires_reasoning else 0.3,
max_tokens=4096 if task_complexity == "high" else 1024
)
# Metriken aktualisieren
self._update_metrics(agent_id, result)
return result
def _select_model(self, complexity: str, requires_reasoning: bool) -> str:
"""Intelligente Modellselektion"""
if requires_reasoning:
# Komplexe Reasoning-Tasks → leistungsfähigere Modelle
if self._within_budget("claude-sonnet-4.5"):
return "claude-sonnet-4.5"
return "gpt-4.1"
if complexity == "low":
# Einfache Tasks → günstige Modelle
return "deepseek-v3.2"
elif complexity == "medium":
return "gpt-4o-mini" if self._within_budget("gpt-4o-mini") else "deepseek-v3.2"
else:
return "gpt-4.1"
def _within_budget(self, model: str) -> bool:
"""Prüft, ob Modell innerhalb des Budgets liegt"""
model_cost = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4o-mini": 1.50,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
return model_cost.get(model, 0) <= 5.00 # Budget-Limit pro 1M Tokens
def _update_metrics(self, agent_id: str, result: Dict):
"""Aktualisiert Performance-Metriken"""
if agent_id not in self._agent_metrics:
self._agent_metrics[agent_id] = AgentMetrics(
agent_id=agent_id,
last_used=datetime.now()
)
metrics = self._agent_metrics[agent_id]
metrics.total_requests += 1
metrics.total_cost += result.get("usage", {}).get("cost", 0)
# Gleitender Durchschnitt für Latenz
new_latency = result.get("latency_ms", 0)
metrics.avg_latency = (metrics.avg_latency * (metrics.total_requests - 1) + new_latency) / metrics.total_requests
def get_cost_report(self) -> Dict[str, Any]:
"""Generiert Kostenbericht für alle Agenten"""
return {
"total_cost_usd": sum(m.total_cost for m in self._agent_metrics.values()),
"by_agent": {
agent_id: {
"requests": m.total_requests,
"cost": round(m.total_cost, 4),
"avg_latency_ms": round(m.avg_latency, 2),
"error_rate": round(m.error_rate, 4)
}
for agent_id, m in self._agent_metrics.items()
}
}
Beispiel: Multi-Agent Crew mit 3 spezialisierten Agenten
async def setup_multi_agent_crew(gateway: HolySheepAgentGateway):
"""Erstellt ein CrewAI-ähnliches Multi-Agent-System"""
router = CostAwareRouter(gateway)
agents = {
"researcher": {
"role": "Forscher",
"goal": "Recherchiere und extrahiere relevante Informationen",
"complexity": "medium",
"model": "deepseek-v3.2"
},
"analyst": {
"role": "Analytiker",
"goal": "Analysiere Daten und identifiziere Muster",
"complexity": "high",
"requires_reasoning": True,
"model": "claude-sonnet-4.5"
},
"writer": {
"role": "Texter",
"goal": "Verfasse klare und präzise Zusammenfassungen",
"complexity": "low",
"model": "deepseek-v3.2"
}
}
async def execute_agent_workflow(agent_id: str, task: str):
"""Führt einen Agenten-Task aus"""
agent_config = agents[agent_id]
complexity = agent_config.get("complexity", "medium")
requires_reasoning = agent_config.get("requires_reasoning", False)
messages = [
{"role": "system", "content": f"Sie sind ein {agent_config['role']}. Ziel: {agent_config['goal']}"},
{"role": "user", "content": task}
]
result = await router.route_request(
agent_id=agent_id,
task_complexity=complexity,
messages=messages,
requires_reasoning=requires_reasoning
)
return result
return execute_agent_workflow, router5. Benchmark-Ergebnisse und Performance-Analyse
In meinen produktiven Setups mit HolySheep AI habe ich folgende Kennzahlen gemessen:
- DeepSeek V3.2: 42 Cent/1M Tokens, durchschnittliche Latenz 38ms, ideal für repetitive Tasks
- Gemini 2.5 Flash: $2.50/1M Tokens, Latenz 45ms, beste Kosten-Effizienz für Batch-Operationen
- GPT-4o-mini: $1.50/1M Tokens, Latenz 120ms, schnelle Antworten für interaktive UIs
- Claude Sonnet 4.5: $15/1M Tokens, Latenz 280ms, überlegene Reasoning-Qualität
5.1 Latenzvergleich bei Tool-Aufrufen
# Benchmark-Skript zum Vergleich der Gateway-Performance
async def benchmark_gateway():
import statistics
config = HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
gateway = HolySheepAgentGateway(config)
await gateway.initialize()
models = ["deepseek-v3.2", "gpt-4o-mini", "claude-sonnet-4.5"]
results = {m: {"latencies": [], "costs": []} for m in models}
# 100 Requests pro Modell
test_messages = [
{"role": "user", "content": "Erkläre das Konzept von Async/Await in Python"}
]
for model in models:
for _ in range(100):
result = await gateway.chat_completion(
model=model,
messages=test_messages,
agent_id="benchmark"
)
results[model]["latencies"].append(result["latency_ms"])
await gateway.close()
# Auswertung
print("=" * 60)
print("BENCHMARK ERGEBNISSE - HolySheep AI Gateway")
print("=" * 60)
for model, data in results.items():
latencies = data["latencies"]
print(f"\n{model}:")
print(f" Latenz (avg): {statistics.mean(latencies):.1f}ms")
print(f" Latenz (p95): {statistics.quantiles(latencies, n=20)[18]:.1f}ms")
print(f" Latenz (p99): {max(latencies):.1f}ms")
Erwartete Ausgabe:
deepseek-v3.2: Latenz (avg): 38.2ms | Kosten: ~$0.000024
gpt-4o-mini: Latenz (avg): 112.4ms | Kosten: ~$0.000085
claude-sonnet-4.5: Latenz (avg): 267.8ms | Kosten: ~$0.000540
6. Häufige Fehler und Lösungen
Fehler 1: Rate Limit Erschöpfung bei Batch-Operationen
Symptom: HTTP 429 Errors nach ca. 50-60 Requests pro Minute
# FEHLERHAFT: Unkontrollierte parallele Requests
async def bad_batch_call(gateway, requests):
tasks = [gateway.chat_completion(**req) for req in requests]
return await asyncio.gather(*tasks) # Rate Limit sofort erreicht!
LÖSUNG: Token Bucket Algorithmus implementieren
import asyncio
import time
class TokenBucketRateLimiter:
"""Effektive Rate-Limit-Behandlung mit Token Bucket"""
def __init__(self, rate: int, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self._lock = asyncio.Lock()
async def acquire(self, tokens: int = 1):
async with self._lock:
while self.tokens < tokens:
self._refill()
if self.tokens < tokens:
await asyncio.sleep(0.1)
self.tokens -= tokens
def _refill(self):
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
Korrigierte Batch-Implementierung
async def good_batch_call(gateway, requests):
limiter = TokenBucketRateLimiter(rate=55, capacity=60) # 55 req/min mit Puffer
async def throttled_request(req):
await limiter.acquire()
return await gateway.chat_completion(**req)
return await asyncio.gather(*[throttled_request(r) for r in requests])Fehler 2: Tool-Call Endlosschleifen
Symptom: Agent ruft wiederholt dasselbe Tool auf ohne Fortschritt
# FEHLERHAFT: Keine Rekursionstiefe-Begrenzung
async def bad_tool_execution(agent, query, tools):
while True: # Gefahr: Endlosschleife!
result = await agent.call_llm(query, tools)
if result.requires_tool:
query = result.next_step # Nie garantiert, dass es endet!
LÖSUNG: Max-Iterationen und Konvergenz-Prüfung
class ToolExecutionGuard:
"""Verhindert Endlosschleifen bei Tool-Aufrufen"""
def __init__(self, max_iterations: int = 10, convergence_threshold: float = 0.95):
self.max_iterations = max_iterations
self.convergence_threshold = convergence_threshold
self.history: List[Dict] = []
async def execute(self, agent, query: str, tools: List[BaseTool]) -> str:
current_state = {"query": query, "result": None}
for iteration in range(self.max_iterations):
# Tool-Ausführung
result = await agent.execute_with_tools(current_state["query"], tools)
current_state["result"] = result
# Konvergenz-Prüfung
if self._is_converged(result):
return self._format_result(result)
# Fortschrittsprüfung
if not self._has_progress(current_state):
raise ToolExecutionError(
f"Kein Fortschritt nach {iteration} Iterationen. "
f"Letzter Stand: {result[:100]}..."
)
# Feedback für nächsten Schritt
current_state["query"] = self._build_feedback(result, iteration)
raise ToolExecutionError(f"Max Iterationen ({self.max_iterations}) erreicht")
def _is_converged(self, result: Dict) -> bool:
"""Prüft ob Ergebnis ausreichend gut ist"""
confidence = result.get("confidence", 0)
return confidence >= self.convergence_threshold
def _has_progress(self, state: Dict) -> bool:
"""Prüft ob Fortschritt erzielt wurde"""
if not self.history:
return True
# Vergleiche mit vorherigen Ergebnissen
last_result = self.history[-1]["result"]
similarity = self._calculate_similarity(last_result, state["result"])
return similarity < 0.9 # Max 90% Ähnlichkeit erlaubt
def _calculate_similarity(self, a: str, b: str) -> float:
"""Einfache Ähnlichkeitsberechnung"""
if not a or not b:
return 0.0
common = sum(1 for ca, cb in zip(a, b) if ca == cb)
return common / max(len(a), len(b))
class ToolExecutionError(Exception):
passFehler 3: Context Window Overflow bei langen Agent-Konversationen
Symptom: "Maximum context length exceeded" trotz ausreichend großer Modelle
# FEHLERHAFT: Volle History immer mitsenden
async def bad_context_handling(messages: List[Dict]):
# Alle Nachrichten werden immer gesendet → Kontext wächst unbegrenzt
return await gateway.chat_completion(messages=messages, model="gpt-4.1")
LÖSUNG: Intelligentes Context-Management
from typing import List, Dict, Optional
import tiktoken
class IntelligentContextManager:
"""
Verwaltet Context-Window dynamisch mit:
- Semantischer Kompression
- Prioritätsbasiertem trimming
- Memory-Efficient Token-Zählung
"""
def __init__(self, max_tokens: int = 128000):
self.max_tokens = max_tokens
self.reserve_tokens = 8000 # Puffer für Antwort
self.available_tokens = max_tokens - self.reserve_tokens
# Tiktoken für GPT-4 (Cl100k_base)
try:
self.encoder = tiktoken.get_encoding("cl100k_base")
except:
self.encoder = None
def optimize_messages(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None
) -> List[Dict[str, str]]:
"""Optimiert Message-Liste für Context-Window"""
# System-Prompt immer behalten (falls vorhanden)
system_tokens = self._count_tokens(system_prompt) if system_prompt else 0
# User/Assistant Messages optimieren
conversation_tokens = self.max_tokens - system_tokens - self.reserve_tokens
optimized = []
accumulated = 0
# Von hinten nach vorne arbeiten (neueste zuerst behalten)
for msg in reversed(messages):
msg_tokens = self._count_tokens(msg.get("content", ""))
if accumulated + msg_tokens <= conversation_tokens:
optimized.insert(0, msg)
accumulated += msg_tokens
else:
# Attempt partial preservation
if msg["role"] == "assistant":
# Zusammenfassung einfügen
summary = self._summarize_messages([msg])
if self._count_tokens(summary) < msg_tokens:
optimized.insert(0, {"role": "assistant", "content": f"[Zusammenfassung]: {summary}"})
break
break
# System-Prompt wieder voranstellen
if system_prompt:
optimized.insert(0, {"role": "system", "content": system_prompt})
return optimized
def _count_tokens(self, text: str) -> int:
"""Zählt Tokens in Text"""
if not text:
return 0
if self.encoder:
return len(self.encoder.encode(text))
# Fallback: grobe Schätzung (~4 Zeichen pro Token)
return len(text) // 4
def _summarize_messages(self, messages: List[Dict]) -> str:
"""Erstellt Zusammenfassung der Messages"""
# Hier könnte ein separates Modell für Summarization verwendet werden
return f"Zusammenfassung von {len(messages)} Nachrichten"
Einsatz in der Praxis:
manager = IntelligentContextManager(max_tokens=128000)
async def safe_agent_call(gateway, messages, system_prompt=None):
optimized = manager.optimize_messages(messages, system_prompt)
return await gateway.chat_completion(
model="deepseek-v3.2",
messages=optimized,
agent_id="context_safe"