Implementierung von Multi-Agenten-Systemen: Architekturmuster und Designprinzipien

from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatAnthropic
from typing import TypedDict, Annotated, Sequence
import operator

# Definiere den Agentenstatus
class AgentState(TypedDict):
    messages: Annotated[Sequence[str], operator.add]
    current_task: str
    results: dict
    error_count: int

# Initialisiere Modelle
coordinator_llm = ChatAnthropic(model="claude-sonnet-4.5", temperature=0)
worker_llm = ChatAnthropic(model="claude-sonnet-4.5", temperature=0.3)

def coordinator_node(state: AgentState):
    """Koordinator-Agent delegiert Aufgaben an Worker"""
    try:
        messages = state["messages"]
        task = state["current_task"]
        
        # Analysiere Aufgabe und bestimme Worker
        prompt = f"""Analysiere diese Aufgabe und bestimme, welcher spezialisierte Agent sie bearbeiten sollte:
        Aufgabe: {task}
        Verfügbare Agenten: data_extraction, analysis, validation
        Gib nur den Agentennamen zurück."""
        
        response = coordinator_llm.invoke(prompt)
        assigned_agent = response.content.strip()
        
        return {
            "messages": messages + [f"Koordinator hat Aufgabe an {assigned_agent} zugewiesen"],
            "current_task": task,
            "results": {"assigned_to": assigned_agent}
        }
    except Exception as e:
        return {
            "messages": messages + [f"Koordinator-Fehler: {str(e)}"],
            "error_count": state.get("error_count", 0) + 1
        }

def worker_node(state: AgentState):
    """Worker-Agent führt zugewiesene Aufgaben aus"""
    try:
        task = state["current_task"]
        assigned_to = state["results"].get("assigned_to")
        
        # Führe Aufgabe basierend auf Spezialisierung aus
        prompt = f"""Du bist ein {assigned_to}-Agent. Führe diese Aufgabe aus:
        {task}
        Liefere ein detailliertes Ergebnis."""
        
        response = worker_llm.invoke(prompt)
        result = response.content
        
        return {
            "messages": state["messages"] + [f"Worker abgeschlossen: {result[:100]}..."],
            "results": {**state["results"], "worker_output": result}
        }
    except Exception as e:
        return {
            "messages": state["messages"] + [f"Worker-Fehler: {str(e)}"],
            "error_count": state.get("error_count", 0) + 1
        }

def should_retry(state: AgentState) -> str:
    """Bestimme, ob Aufgabe wiederholt werden soll"""
    if state.get("error_count", 0) > 3:
        return "end"
    if "worker_output" in state.get("results", {}):
        return "end"
    return "coordinator"

# Baue den Graph
workflow = StateGraph(AgentState)

# Füge Knoten hinzu
workflow.add_node("coordinator", coordinator_node)
workflow.add_node("worker", worker_node)

# Füge Kanten hinzu
workflow.set_entry_point("coordinator")
workflow.add_edge("coordinator", "worker")
workflow.add_conditional_edges(
    "worker",
    should_retry,
    {
        "coordinator": "coordinator",
        "end": END
    }
)

# Kompiliere den Graph
app = workflow.compile()

# Führe aus
initial_state = {
    "messages": [],
    "current_task": "Extrahiere Kundenstimmung aus Support-Tickets",
    "results": {},
    "error_count": 0
}

result = app.invoke(initial_state)
print(f"Endergebnis: {result['results']}")

from crewai import Agent, Task, Crew, Process
from langchain.chat_models import ChatAnthropic
from langchain.tools import Tool

# Initialisiere LLM
llm = ChatAnthropic(model="claude-sonnet-4.5", temperature=0.3)

# Definiere spezialisierte Agenten
researcher = Agent(
    role="Forschungsanalyst",
    goal="Informationen aus Dokumenten sammeln und analysieren",
    backstory="""Du bist ein Experte darin, Erkenntnisse aus Daten zu extrahieren.
    Du zeichnest dich darin aus, Muster und relevante Informationen zu finden.""",
    verbose=True,
    allow_delegation=False,
    llm=llm
)

writer = Agent(
    role="Content-Writer",
    goal="Überzeugende, präzise Inhalte basierend auf Forschung erstellen",
    backstory="""Du bist ein erfahrener Autor, der komplexe
    Informationen in klare, ansprechende Inhalte verwandelt.""",
    verbose=True,
    allow_delegation=False,
    llm=llm
)

reviewer = Agent(
    role="Qualitätsprüfer",
    goal="Genauigkeit, Konsistenz und Qualität sicherstellen",
    backstory="""Du bist ein akribischer Editor, der Fehler aufspürt
    und hohe Qualität der Ausgabe sicherstellt.""",
    verbose=True,
    allow_delegation=True,
    llm=llm
)

# Definiere Aufgaben
research_task = Task(
    description="""Recherchiere das Thema: KI-Multi-Agenten-Systeme.
    Sammle Schlüsselkonzepte, Vorteile und Implementierungsmuster.
    Konzentriere dich auf produktionsreife Ansätze.""",
    agent=researcher,
    expected_output="Umfassende Forschungszusammenfassung mit Schlüsselergebnissen"
)

writing_task = Task(
    description="""Schreibe unter Verwendung der Forschungsergebnisse einen technischen Blogpost
    über die Implementierung von Multi-Agenten-Systemen. Beinhalte:
    - Architekturmuster
    - Code-Beispiele
    - Best Practices
    Zielgruppe: Software-Ingenieure.""",
    agent=writer,
    expected_output="Vollständiger Blogpost-Entwurf (800-1000 Wörter)"
)

review_task = Task(
    description="""Überprüfe den Blogpost auf:
    - Technische Genauigkeit
    - Klarheit und Lesbarkeit
    - Vollständigkeit
    Gib spezifische Verbesserungsvorschläge.""",
    agent=reviewer,
    expected_output="Überprüfter Inhalt mit Feedback und Korrekturen"
)

# Erstelle Crew mit hierarchischem Prozess
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # Aufgaben werden nacheinander ausgeführt
    verbose=2
)

# Führe den Workflow aus
try:
    result = crew.kickoff()
    print(f"\n\nEndausgabe:\n{result}")
except Exception as e:
    print(f"Fehler bei der Crew-Ausführung: {str(e)}")
    # Implementiere hier Fallback-Logik

import redis
import json
from typing import Dict, Any
import asyncio
from datetime import datetime

class AgentCommunicationHub:
    """Ereignisgesteuertes Kommunikationssystem für Multi-Agenten-Koordination"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_client = redis.from_url(redis_url)
        self.pubsub = self.redis_client.pubsub()
        self.agent_id = None
        
    def register_agent(self, agent_id: str, capabilities: list):
        """Registriere Agent mit seinen Fähigkeiten"""
        self.agent_id = agent_id
        agent_info = {
            "id": agent_id,
            "capabilities": capabilities,
            "status": "active",
            "registered_at": datetime.utcnow().isoformat()
        }
        self.redis_client.hset(
            "agents",
            agent_id,
            json.dumps(agent_info)
        )
        # Abonniere agenten-spezifischen Kanal
        self.pubsub.subscribe(f"agent:{agent_id}")
        
    def publish_event(self, event_type: str, data: Dict[Any, Any]):
        """Veröffentliche Ereignis zum Kommunikationsbus"""
        event = {
            "type": event_type,
            "source": self.agent_id,
            "data": data,
            "timestamp": datetime.utcnow().isoformat()
        }
        self.redis_client.publish(
            f"events:{event_type}",
            json.dumps(event)
        )
        
    def request_task(self, task_type: str, task_data: Dict[Any, Any]):
        """Fordere einen anderen Agenten auf, eine Aufgabe auszuführen"""
        task = {
            "id": f"task_{datetime.utcnow().timestamp()}",
            "type": task_type,
            "requester": self.agent_id,
            "data": task_data,
            "status": "pending"
        }
        # Speichere Aufgabe in Redis
        self.redis_client.hset(
            "tasks",
            task["id"],
            json.dumps(task)
        )
        # Veröffentliche Aufgabenanfrage
        self.publish_event("task_request", task)
        return task["id"]
        
    async def listen_for_events(self, callback):
        """Höre auf Ereignisse und rufe Callback auf"""
        for message in self.pubsub.listen():
            if message["type"] == "message":
                try:
                    event = json.loads(message["data"])
                    await callback(event)
                except Exception as e:
                    print(f"Fehler bei der Verarbeitung des Ereignisses: {e}")
                    
    def get_agent_status(self, agent_id: str) -> Dict[Any, Any]:
        """Erhalte Status eines anderen Agenten"""
        agent_data = self.redis_client.hget("agents", agent_id)
        if agent_data:
            return json.loads(agent_data)
        return None

# Beispielverwendung
async def main():
    # Initialisiere Kommunikationshub für mehrere Agenten
    coordinator_hub = AgentCommunicationHub()
    coordinator_hub.register_agent(
        "coordinator_1",
        ["task_delegation", "result_aggregation"]
    )
    
    worker_hub = AgentCommunicationHub()
    worker_hub.register_agent(
        "worker_1",
        ["data_processing", "analysis"]
    )
    
    # Koordinator fordert Aufgabe an
    task_id = coordinator_hub.request_task(
        "data_analysis",
        {"dataset": "customer_feedback.csv", "metric": "sentiment"}
    )
    
    # Worker verarbeitet Ereignisse
    async def handle_event(event):
        if event["type"] == "task_request":
            print(f"Worker hat Aufgabe erhalten: {event['data']['id']}")
            # Verarbeite Aufgabe und veröffentliche Ergebnis
            worker_hub.publish_event(
                "task_complete",
                {"task_id": event['data']['id'], "result": "Analyse abgeschlossen"}
            )
    
    await worker_hub.listen_for_events(handle_event)

# asyncio.run(main())