Moderne KI-Anwendungen erfordern zunehmend komplexe Workflows, die weit über einzelne Modellaufrufe hinausgehen. Multi-Agent-Systeme ermöglichen die Orchestrierung mehrerer spezialisierter Agenten, die gemeinsam komplexe Aufgaben bewältigen. In diesem Tutorial zeige ich Ihnen, wie Sie ein produktionsreifes Multi-Agent-System mit HolySheep AI implementieren – inklusive intelligenter Task-Allokation, zustandsbehafteter Kommunikation und robuster Fehlerbehandlung.
Architekturüberblick: Das Agent-Netzwerk
Ein effektives Multi-Agent-System basiert auf drei Kernkomponenten:
- Task Router: Analysiert eingehende Requests und verteilt sie an passende Agenten
- State Manager: Verwaltet den globalen Zustand und ermöglicht State-Sharing zwischen Agenten
- Error Supervisor: Überwacht die Ausführung und implementiert Recovery-Strategien
Die folgende Architektur zeigt das Zusammenspiel dieser Komponenten:
┌─────────────────────────────────────────────────────────────┐
│ Multi-Agent Orchestrator │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Router │───▶│ Agent │───▶│ State │ │
│ │ Agent │ │ Pool │ │ Store │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ │ ┌────┴────┐ │ │
│ │ │ Error │◀────────┘ │
│ │ │Supervisor│ │
│ │ └─────────┘ │
│ │ │
└───────┼─────────────────────────────────────────────────────┘
│
▼
┌───────────────────┐
│ HolySheep AI API │
│ (Multi-Provider) │
└───────────────────┘Task-Allokation: Intelligente Verteilung mit_priorisierten Agenten
Die Task-Allokation ist der kritischste Aspekt eines Multi-Agent-Systems. Ich implementiere einen Prioritäts-basierten Router, der Aufgaben basierend auf Komplexität, Domäne und aktueller Systemlast verteilt.
import aiohttp
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import hashlib
import time
class AgentCapability(Enum):
REASONING = "reasoning"
CREATIVE = "creative"
ANALYTICAL = "analytical"
CODING = "coding"
GENERAL = "general"
@dataclass
class Agent:
id: str
name: str
capabilities: List[AgentCapability]
max_concurrent: int = 3
current_load: int = 0
avg_latency_ms: float = 0.0
cost_per_1k_tokens: float = 0.0
@dataclass
class Task:
id: str
description: str
required_capabilities: List[AgentCapability]
priority: int # 1-5, lower = higher priority
estimated_tokens: int
metadata: Dict = field(default_factory=dict)
class IntelligentTaskRouter:
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.agents: Dict[str, Agent] = {}
self.task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self.state_store: Dict[str, any] = {}
self._init_agents()
def _init_agents(self):
"""Initialisiere Agenten-Pool mit HolySheep-kompatiblen Modellen"""
self.agents = {
"reasoner": Agent(
id="agent_reasoner",
name="Deep Reasoning Agent",
capabilities=[AgentCapability.REASONING, AgentCapability.ANALYTICAL],
max_concurrent=2,
cost_per_1k_tokens=0.42 # DeepSeek V3.2
),
"creative": Agent(
id="agent_creative",
name="Creative Agent",
capabilities=[AgentCapability.CREATIVE],
max_concurrent=3,
cost_per_1k_tokens=0.42
),
"coder": Agent(
id="agent_coder",
name="Coding Agent",
capabilities=[AgentCapability.CODING, AgentCapability.REASONING],
max_concurrent=2,
cost_per_1k_tokens=2.50 # Gemini 2.5 Flash
),
"general": Agent(
id="agent_general",
name="General Purpose Agent",
capabilities=[AgentCapability.GENERAL],
max_concurrent=5,
cost_per_1k_tokens=8.0 # GPT-4.1
)
}
async def allocate_task(self, task: Task) -> Optional[Agent]:
"""Finde optimalen Agent basierend auf Fähigkeiten und Last"""
candidates = []
for agent in self.agents.values():
# Prüfe Kapazität
if agent.current_load >= agent.max_concurrent:
continue
# Prüfe Fähigkeiten-Match
capability_match = any(
cap in agent.capabilities
for cap in task.required_capabilities
)
if capability_match:
# Berechne Fitness-Score
load_factor = 1 - (agent.current_load / agent.max_concurrent)
latency_factor = 1 / (agent.avg_latency_ms / 1000 + 1)
priority_bonus = (6 - task.priority) * 0.1
fitness = (
0.4 * capability_match +
0.3 * load_factor +
0.2 * latency_factor +
0.1 * priority_bonus
)
candidates.append((fitness, agent))
if candidates:
candidates.sort(key=lambda x: x[0], reverse=True)
return candidates[0][1]
return None
async def execute_task(self, task: Task) -> Dict:
"""Führe Task mit ausgewähltem Agent aus"""
agent = await self.allocate_task(task)
if not agent:
return {
"status": "queued",
"task_id": task.id,
"message": "Alle Agenten ausgelastet, Task wird gequeued"
}
agent.current_load += 1
start_time = time.time()
try:
result = await self._call_holysheep_api(task, agent)
# Update Performance-Metriken
latency = (time.time() - start_time) * 1000
agent.avg_latency_ms = (agent.avg_latency_ms * 0.7) + (latency * 0.3)
# Speichere Ergebnis im State Store
self.state_store[task.id] = {
"result": result,
"agent_id": agent.id,
"latency_ms": latency,
"timestamp": time.time()
}
return {
"status": "completed",
"task_id": task.id,
"agent_id": agent.id,
"result": result,
"latency_ms": latency
}
finally:
agent.current_load = max(0, agent.current_load - 1)
async def _call_holysheep_api(self, task: Task, agent: Agent) -> Dict:
"""API-Call zu HolySheep mit Fehlerbehandlung"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Wähle Modell basierend auf Agent-Typ
model_map = {
"agent_reasoner": "deepseek-v3.2",
"agent_creative": "deepseek-v3.2",
"agent_coder": "gemini-2.5-flash",
"agent_general": "gpt-4.1"
}
payload = {
"model": model_map.get(agent.id, "deepseek-v3.2"),
"messages": [
{"role": "system", "content": f"You are {agent.name}"},
{"role": "user", "content": task.description}
],
"temperature": 0.7,
"max_tokens": min(task.estimated_tokens, 4096)
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
return {
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"model": data.get("model")
}
else:
error_text = await response.text()
raise Exception(f"API Error {response.status}: {error_text}")
Benchmark-Initialisierung
async def benchmark_router():
router = IntelligentTaskRouter("YOUR_HOLYSHEEP_API_KEY")
tasks = [
Task(
id=f"task_{i}",
description=f"Analysiere komplexe Query {i} mit logischer Schlussfolgerung",
required_capabilities=[AgentCapability.REASONING, AgentCapability.ANALYTICAL],
priority=2,
estimated_tokens=1000
) for i in range(100)
]
start = time.time()
results = await asyncio.gather(*[router.execute_task(t) for t in tasks])
duration = time.time() - start
completed = sum(1 for r in results if r["status"] == "completed")
print(f"Benchmark Results:")
print(f" Total Tasks: {len(tasks)}")
print(f" Completed: {completed}")
print(f" Duration: {duration:.2f}s")
print(f" Throughput: {len(tasks)/duration:.2f} tasks/s")
return {"duration": duration, "throughput": len(tasks)/duration}
asyncio.run(benchmark_router())
State-Sharing: Verteilter Zustand zwischen Agenten
Multi-Agent-Systeme benötigen einen konsistenten Zustandsspeicher, der es allen Agenten ermöglicht, Informationen auszutauschen. Ich implementiere einen Thread-sicheren State-Store mit Change-Notification.
import asyncio
from typing import Any, Dict, List, Callable, Optional
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import threading
import json
@dataclass
class StateEntry:
key: str
value: Any
version: int
timestamp: datetime
agent_id: str
ttl_seconds: Optional[int] = None
class StateManager:
"""Thread-safe State-Sharing für Multi-Agent-Systeme"""
def __init__(self):
self._state: Dict[str, StateEntry] = {}
self._lock = asyncio.Lock()
self._version_lock = threading.Lock()
self._version_counter = 0
self._watchers: Dict[str, List[Callable]] = defaultdict(list)
self._context_stack: Dict[str, List[Dict]] = defaultdict(list)
def _next_version(self) -> int:
with self._version_lock:
self._version_counter += 1
return self._version_counter
async def set(self, key: str, value: Any, agent_id: str,
ttl_seconds: Optional[int] = None) -> StateEntry:
"""Setze State mit automatischer Versionierung"""
async with self._lock:
entry = StateEntry(
key=key,
value=value,
version=self._next_version(),
timestamp=datetime.now(),
agent_id=agent_id,
ttl_seconds=ttl_seconds
)
self._state[key] = entry
# Benachrichtige Watcher
for callback in self._watchers.get(key, []):
await callback(key, value, entry.version)
# Benachrichtige globale Watcher
for callback in self._watchers.get("*", []):
await callback(key, value, entry.version)
return entry
async def get(self, key: str) -> Optional[Any]:
"""Lese aktuellen State"""
async with self._lock:
entry = self._state.get(key)
if entry:
return entry.value
return None
async def get_with_meta(self, key: str) -> Optional[Dict]:
"""Lese State mit Metadaten"""
async with self._lock:
entry = self._state.get(key)
if entry:
return {
"value": entry.value,
"version": entry.version,
"timestamp": entry.timestamp.isoformat(),
"agent_id": entry.agent_id
}
return None
async def compare_and_set(self, key: str, expected_version: int,
new_value: Any, agent_id: str) -> bool:
"""Atomare CAS-Operation für Concurrency-Control"""
async with self._lock:
entry = self._state.get(key)
if entry and entry.version == expected_version:
entry.value = new_value
entry.version = self._next_version()
entry.timestamp = datetime.now()
entry.agent_id = agent_id
return True
return False
async def push_context(self, agent_id: str, context: Dict) -> int:
"""Pushe Kontext auf Agent-spezifischen Stack"""
async with self._lock:
stack = self._context_stack[agent_id]
stack.append(context)
return len(stack)
async def pop_context(self, agent_id: str) -> Optional[Dict]:
"""Popt Kontext vom Agent-Stack"""
async with self._lock:
stack = self._context_stack.get(agent_id, [])
if stack:
return stack.pop()
return None
async def watch(self, key_pattern: str, callback: Callable):
"""Registriere Watcher für Key-Änderungen"""
self._watchers[key_pattern].append(callback)
async def snapshot(self) -> Dict[str, Any]:
"""Erstelle konsistenten Snapshot aller States"""
async with self._lock:
return {
key: {
"value": entry.value,
"version": entry.version,
"agent_id": entry.agent_id
}
for key, entry in self._state.items()
}
class MultiAgentSession:
"""Orchestriert mehrere Agenten mit gemeinsamem State"""
def __init__(self, state_manager: StateManager, api_key: str):
self.state = state_manager
self.api_key = api_key
self.participants: List[str] = []
self.session_id = f"session_{int(asyncio.get_event_loop().time() * 1000)}"
async def register_agent(self, agent_id: str) -> Dict:
"""Registriere Agent für diese Session"""
self.participants.append(agent_id)
# Initialisiere Agent-spezifischen State
await self.state.set(
f"session:{self.session_id}:agent:{agent_id}",
{"status": "active", "joined_at": datetime.now().isoformat()},
agent_id
)
# Aktualisiere Session-State
session_state = await self.state.get(f"session:{self.session_id}") or {}
session_state["participants"] = self.participants
await self.state.set(
f"session:{self.session_id}",
session_state,
"system"
)
return {"agent_id": agent_id, "session_id": self.session_id}
async def shared_knowledge(self, agent_id: str, knowledge_key: str,
value: Any) -> bool:
"""Teile Wissen mit allen Session-Teilnehmern"""
full_key = f"session:{self.session_id}:shared:{knowledge_key}"
# Atomares Update mit Version-Check
current = await self.state.get_with_meta(full_key)
if current:
success = await self.state.compare_and_set(
full_key,
current["version"],
{"value": value, "contributor": agent_id},
agent_id
)
else:
await self.state.set(full_key, {"value": value, "contributor":
Verwandte Ressourcen
Verwandte Artikel