Multi-Agent System aufsetzen

Wissen

Ein einzelner Agent stößt schnell an seine Grenzen: Er muss planen, ausführen und die Qualität seiner eigenen Arbeit bewerten -- alles in einem Loop. In der Produktion teilst du diese Verantwortung auf spezialisierte Agents auf. Das Ergebnis ist ein Multi-Agent System, das robuster, effizienter und kontrollierbarer ist als ein Einzelkämpfer.

Die Planner-Executor-Reviewer-Architektur

Die bewährteste Architektur für Produktions-Agents besteht aus drei Rollen:

Warum nicht ein einziger Agent? Drei spezialisierte Agents sind in der Summe günstiger und besser als ein universeller Agent, der alles gleichzeitig können muss. Der Planner wird nur einmal aufgerufen (teuer, aber selten), der Executor läuft oft (günstig), und der Reviewer sorgt für Qualität, ohne den gesamten Kontext des Executors zu brauchen.

Inter-Agent-Kommunikation

Agents kommunizieren über strukturierte Nachrichten -- typischerweise JSON-Objekte mit klaren Schemas. Das ist kein Freitext-Chat, sondern ein definiertes Protokoll:

# Nachrichtenformat für Inter-Agent-Kommunikation
from dataclasses import dataclass, field
from typing import Literal
from datetime import datetime

@dataclass
class AgentMessage:
    sender: Literal["planner", "executor", "reviewer"]
    receiver: Literal["planner", "executor", "reviewer"]
    message_type: Literal["plan", "task", "result", "feedback", "approval"]
    payload: dict
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    correlation_id: str = ""  # Verknüpft zusammengehörige Nachrichten

Verstehen

Vollständiges Multi-Agent System in Python

Das folgende Beispiel zeigt ein produktionsfähiges Multi-Agent System. Der Planner erstellt einen Plan, der Executor arbeitet ihn ab, und der Reviewer prüft das Ergebnis.

import anthropic
import json
from dataclasses import dataclass, field, asdict
from typing import Literal
from datetime import datetime

client = anthropic.Anthropic()

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    message_type: str
    payload: dict
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class ProductionAgent:
    def __init__(self, name: str, model: str, system_prompt: str):
        self.name = name
        self.model = model
        self.system_prompt = system_prompt
        self.message_log: list[AgentMessage] = []

    def process(self, task: str) -> str:
        """Verarbeitet eine Aufgabe und gibt die Antwort zurück."""
        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system_prompt,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text

class MultiAgentOrchestrator:
    def __init__(self):
        self.planner = ProductionAgent(
            name="planner",
            model="claude-opus-5",
            system_prompt=(
                "Du bist ein Planungs-Agent. Analysiere Aufgaben und erstelle "
                "strukturierte Pläne im JSON-Format. Jeder Schritt muss klar, "
                "atomar und ausführbar sein. Format: "
                '{"steps": [{"id": 1, "action": "...", "expected_output": "..."}]}'
            )
        )
        self.executor = ProductionAgent(
            name="executor",
            model="claude-haiku-4-5",
            system_prompt=(
                "Du bist ein Ausführungs-Agent. Du bekommst einen einzelnen "
                "Schritt und führst ihn präzise aus. Gib nur das Ergebnis "
                "zurück, keine Erklärungen."
            )
        )
        self.reviewer = ProductionAgent(
            name="reviewer",
            model="claude-sonnet-5",
            system_prompt=(
                "Du bist ein Review-Agent. Prüfe Ergebnisse auf Korrektheit, "
                "Vollständigkeit und Qualität. Antworte mit JSON: "
                '{"approved": true/false, "feedback": "...", "score": 0-10}'
            )
        )

    def run(self, task: str) -> dict:
        # 1. Planner erstellt den Plan
        plan_raw = self.planner.process(f"Erstelle einen Plan für: {task}")
        plan = json.loads(plan_raw)

        results = []
        # 2. Executor arbeitet jeden Schritt ab
        for step in plan["steps"]:
            result = self.executor.process(
                f"Führe aus: {step['action']}\n"
                f"Erwartetes Ergebnis: {step['expected_output']}"
            )
            results.append({"step_id": step["id"], "result": result})

        # 3. Reviewer prüft das Gesamtergebnis
        review = self.reviewer.process(
            f"Aufgabe: {task}\n"
            f"Plan: {json.dumps(plan, ensure_ascii=False)}\n"
            f"Ergebnisse: {json.dumps(results, ensure_ascii=False)}"
        )

        return {
            "plan": plan,
            "results": results,
            "review": json.loads(review)
        }

# Nutzung
orchestrator = MultiAgentOrchestrator()
output = orchestrator.run("Analysiere die Performance der API-Endpoints")

TypeScript-Variante

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

interface AgentMessage {
  sender: string;
  receiver: string;
  messageType: string;
  payload: Record<string, unknown>;
  timestamp: string;
}

interface PlanStep {
  id: number;
  action: string;
  expected_output: string;
}

class ProductionAgent {
  constructor(
    private name: string,
    private model: string,
    private systemPrompt: string,
    public messageLog: AgentMessage[] = []
  ) {}

  async process(task: string): Promise<string> {
    const response = await client.messages.create({
      model: this.model,
      max_tokens: 2048,
      system: this.systemPrompt,
      messages: [{ role: "user", content: task }],
    });
    const block = response.content[0];
    return block.type === "text" ? block.text : "";
  }
}

class MultiAgentOrchestrator {
  private planner: ProductionAgent;
  private executor: ProductionAgent;
  private reviewer: ProductionAgent;

  constructor() {
    this.planner = new ProductionAgent(
      "planner",
      "claude-opus-5",
      "Du bist ein Planungs-Agent. Erstelle strukturierte Pläne als JSON."
    );
    this.executor = new ProductionAgent(
      "executor",
      "claude-haiku-4-5",
      "Du bist ein Ausführungs-Agent. Führe Schritte präzise aus."
    );
    this.reviewer = new ProductionAgent(
      "reviewer",
      "claude-sonnet-5",
      "Du bist ein Review-Agent. Prüfe Ergebnisse und antworte als JSON."
    );
  }

  async run(task: string) {
    // 1. Plan erstellen
    const planRaw = await this.planner.process(
      `Erstelle einen Plan für: ${task}`
    );
    const plan = JSON.parse(planRaw);

    // 2. Schritte ausführen
    const results = [];
    for (const step of plan.steps as PlanStep[]) {
      const result = await this.executor.process(
        `Führe aus: ${step.action}`
      );
      results.push({ stepId: step.id, result });
    }

    // 3. Review
    const review = await this.reviewer.process(
      `Aufgabe: ${task}\nErgebnisse: ${JSON.stringify(results)}`
    );

    return { plan, results, review: JSON.parse(review) };
  }
}

Anwenden

Reflektieren

Ein Multi-Agent System ist nicht automatisch besser als ein einzelner Agent. Der Mehrwert entsteht durch klare Rollentrennung, passende Modellwahl pro Rolle und strukturierte Kommunikation. In der Praxis startest du mit einem einzelnen Agent und teilst erst dann auf, wenn du merkst, dass eine Rolle zu komplex wird oder die Kosten zu hoch sind.