Agenten Orchestrierung: Best Practices für Multi-Agent-Systeme

Die Orchestrierung von KI-Agenten ist die Kunst, mehrere Agenten so zu koordinieren, dass sie gemeinsam komplexe Aufgaben lösen — zuverlässig, effizient und nachvollziehbar. Die einzelnen Agenten sind oft nicht das Problem. Das Problem ist die Koordination: Wer macht was? In welcher Reihenfolge? Was passiert bei Fehlern? Wie kontrollierst Du die Kosten?

TL;DR — Agenten-Orchestrierung in 90 Sekunden

Was ist Agenten-Orchestrierung?

Agenten-Orchestrierung ist die Koordination mehrerer KI-Agenten mit:

• Aufgabenteilung: Wer macht was? Welche Expertise braucht welcher Teil?
• Kommunikation: Wie tauschen Agenten Informationen? Über State, Messages oder Events?
• Kontrolle: Wer entscheidet den nächsten Schritt? Zentraler Orchestrator oder bedingte Routen?
• Fehlerbehandlung: Was passiert bei Fehlern? Retry, Fallback, Human-in-the-Loop?
• Ressourcenmanagement: Wie werden LLM-Calls optimiert? Caching, Model-Auswahl, Token-Limits?
• Observability: Wie verfolgst Du, was passiert? Logging, Tracing, Metriken?

Single-Agent vs. Multi-Agent — Wann brauchst Du Orchestrierung?

Single-Agent reicht bei: linearen Aufgaben, einer Expertise, keinem Iterationsbedarf, Aufgabe in einem LLM-Call lösbar.

Multi-Agent ist nötig bei: verschiedenen Expertisen (Recherche + Schreiben + Review), Iteration (Writer ↔ Reviewer), Parallelität, Aufgaben jenseits des Context-Limits, verschiedenen Tools, Human-in-the-Loop.

Orchestrierungs-Patterns — Im Detail

1. Hub-and-Spoke (Zentraler Orchestrator)

Ein zentraler Orchestrator-Agent entscheidet, welche Agenten wann ausgeführt werden.

         [Orchestrator]
        /     |      \
[Agent A] [Agent B] [Agent C]

Vorteile: Klarer Kontrollfluss, einfache Fehlerbehandlung, nachvollziehbar. Nachteile: Single Point of Failure, Orchestrator als Flaschenhals, zusätzliche LLM-Kosten.

def orchestrator(state):
    task = state["task"]
    # LLM entscheidet, welcher Agent benötigt wird
    response = llm.invoke(f"Welcher Agent für: {task}? (research/code/write)")
    task_type = response.content.strip().lower()
    return {"task_type": task_type}

def route_to_agent(state):
    return state["task_type"]

# In LangGraph:
workflow.add_conditional_edges("orchestrator", route_to_agent, {
    "research": "research_agent",
    "code": "code_agent",
    "write": "writer_agent"
})

2. Pipeline (Sequentiell)

Agenten arbeiten nacheinander, jeder gibt an den nächsten weiter. CrewAI’s Process.sequential verwendet dieses Pattern standardmäßig.

[Agent A] → [Agent B] → [Agent C] → [Output]

Vorteile: Einfach, vorhersehbar, jeder Agent bekommt alle vorherigen Ergebnisse. Nachteile: Keine Parallelität, ein fehlerhafter Agent blockiert alles.

Wann verwenden? Bei linearen Workflows ohne Verzweigungen (Recherche → Schreiben → Review).

3. DAG (Directed Acyclic Graph)

Agenten arbeiten in Abhängigkeiten, mit Parallelität wo möglich. Ein DAG erlaubt es, unabhängige Tasks parallel auszuführen und abhängige Tasks zu koordinieren.

[Agent A] ──→ [Agent C] ──→ [Agent E] (Merge)
     └──→ [Agent B] ──→ [Agent D] ──┘

Vorteile: Parallelität — unabhängige Tasks laufen gleichzeitig. Effizient — Gesamtzeit = längster Pfad. Nachteile: Komplex zu debuggen, Abhängigkeiten müssen klar definiert sein, Race Conditions bei Shared State.

Beispiel mit LangGraph (parallele Researcher):

class ResearchState(TypedDict):
    topic: str
    general_research: str
    technical_research: str
    combined_report: str

def general_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Allgemeine Infos zu: {state['topic']}")
    return {"general_research": response.content}

def technical_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Technische Details zu: {state['topic']}")
    return {"technical_research": response.content}

def merge_results(state: ResearchState) -> dict:
    combined = f"Allgemein:\n{state['general_research']}\n\nTechnisch:\n{state['technical_research']}"
    return {"combined_report": combined}

# Graph mit parallelen Nodes
workflow = StateGraph(ResearchState)
workflow.add_node("general", general_researcher)
workflow.add_node("technical", technical_researcher)
workflow.add_node("merge", merge_results)

workflow.set_entry_point("general")
workflow.add_edge("general", "technical")  # Beide parallel
workflow.add_edge("general", "merge")
workflow.add_edge("technical", "merge")
workflow.add_edge("merge", END)

Wann verwenden? Wenn unabhängige Teilaufgaben parallel ausgeführt werden können (z.B. 3 Recherchen, die dann zusammengeführt werden).

4. Event-Driven (Reaktiv)

Agenten reagieren auf Events statt auf direkte Aufrufe. Ein Event-Bus verteilt Events an alle interessierten Agenten.

[Event Bus]
  ├── [Agent A] (hört auf "research.done")
  ├── [Agent B] (hört auf "code.reviewed")
  └── [Agent C] (hört auf "test.failed")

Vorteile: Lose Kopplung, skalierbar, Agenten dynamisch hinzufügbar. Nachteile: Schwer nachvollziehbar, Race Conditions möglich, Debugging komplex.

from collections import defaultdict

class EventBus:
    def __init__(self):
        self.subscribers = defaultdict(list)
    
    def subscribe(self, event_type: str, handler):
        self.subscribers[event_type].append(handler)
    
    def publish(self, event_type: str, data: dict):
        results = []
        for handler in self.subscribers[event_type]:
            results.append(handler(data))
        return results

bus = EventBus()
bus.subscribe("research.done", writer_agent)
bus.subscribe("research.done", fact_checker_agent)
bus.subscribe("code.reviewed", refactor_agent)
bus.publish("research.done", {"topic": "KI 2026", "result": "..."})

Wann verwenden? Bei lose gekoppelten Systemen, wo Agenten dynamisch hinzugefügt werden und die Reihenfolge nicht feststeht.

Pattern-Vergleich

Pattern	Komplexität	Parallelität	Flexibilität	Debugging	Eignung
Pipeline	Niedrig	❌	Niedrig	Einfach	Lineare Workflows
Hub-and-Spoke	Mittel	❌	Mittel	Mittel	Dynamisches Routing
DAG	Hoch	✅	Hoch	Schwer	Parallele Tasks
Event-Driven	Hoch	✅	Sehr hoch	Sehr schwer	Lose gekoppelte Systeme

Best Practices — Detailliert

1. Klare Rollen definieren (Single Responsibility Principle)

Jeder Agent sollte genau eine Verantwortung haben. Ein Agent, der gleichzeitig recherchiert, schreibt und testet, ist schwer zu debuggen, teuer und produziert schlechtere Ergebnisse als drei spezialisierte Agenten.

Schlecht:

agent = Agent(
    role="Researcher, Writer and Tester",
    goal="Research, write and test everything"
)

Gut:

researcher = Agent(role="Researcher", goal="Find information")
writer = Agent(role="Writer", goal="Write content based on research")
tester = Agent(role="Tester", goal="Test functionality and report bugs")

2. Zustände explizit machen

Verwende typisierte States statt impliziter Kommunikation. Das macht den Workflow nachvollziehbar und fehlerresistent.

from typing import TypedDict, Optional, List, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    task_type: str
    research_result: Optional[str]
    draft: Optional[str]
    review_feedback: Optional[str]
    final_output: Optional[str]
    error: Optional[str]
    retry_count: int
    messages: Annotated[List[str], add]  # Wird appended, nicht überschrieben

3. Fehlerbehandlung und Retry-Logic

LLMs sind unzuverlässig — Rate-Limits, Timeouts, Halluzinationen. Jeder Agent braucht Fehlerbehandlung:

import logging

logger = logging.getLogger("agent_orchestrator")

def agent_with_retry(agent_func, state, max_retries=3, fallback=None):
    for attempt in range(max_retries):
        try:
            logger.info(f"Agent {agent_func.__name__} - Versuch {attempt + 1}")
            result = agent_func(state)
            if validate_result(result):
                return result
            else:
                logger.warning(f"Agent {agent_func.__name__} - ungültiges Ergebnis")
                state["error"] = "Invalid result"
        except Exception as e:
            logger.error(f"Agent {agent_func.__name__} - Fehler: {str(e)}")
            state["error"] = str(e)
            state["retry_count"] = attempt + 1
    
    if fallback:
        return fallback(state)
    return {**state, "error": f"Agent failed after {max_retries} retries"}

def validate_result(result):
    if not result:
        return False
    if "error" in result and result["error"]:
        return False
    return True

4. Kostenkontrolle

LLM-Calls sind teuer. Bei 5 Agenten mit je 3 Revisionen sind das 15+ LLM-Calls. Mit GPT-4o können das 5-20$ pro Durchlauf sein.

class CostTracker:
    def __init__(self, max_budget=10.0):
        self.costs = {}
        self.max_budget = max_budget
    
    def track(self, agent_name, tokens, model="gpt-4o"):
        cost_per_1k = {
            "gpt-4o": 0.005,
            "gpt-4o-mini": 0.0003,
            "claude-3.5-sonnet": 0.003,
        }
        rate = cost_per_1k.get(model, 0.005)
        cost = (tokens / 1000) * rate
        self.costs[agent_name] = self.costs.get(agent_name, 0) + cost
        
        if self.total() > self.max_budget:
            raise Exception(f"Budget überschritten: ${self.total():.2f}")
    
    def total(self):
        return sum(self.costs.values())
    
    def report(self):
        lines = [f"Cost Report (Total: ${self.total():.4f})"]
        for agent, cost in sorted(self.costs.items()):
            lines.append(f"  {agent}: ${cost:.4f}")
        return "\n".join(lines)

Kosten-Optimierungs-Strategien:

• Caching: Gleiche Prompts nicht zweimal ausführen
• Model-Auswahl: Einfache Tasks mit GPT-4o-mini ($0.30/M), komplexe mit GPT-4o ($5.00/M)
• Early Stopping: Abbrechen, wenn Ergebnis “gut genug”
• Token-Limits: max_tokens pro Agent festlegen
• Lokale Modelle: Für Entwicklung Ollama verwenden ($0)

5. Observability und Logging

Ohne Observability bist Du blind — Du weißt nicht, welcher Agent wie lange braucht und wo Fehler entstehen.

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_orchestrator")

def observed_agent(agent_func):
    @wraps(agent_func)
    def wrapper(state):
        name = agent_func.__name__
        start = time.time()
        logger.info(f"[{name}] START - Input keys: {list(state.keys())}")
        try:
            result = agent_func(state)
            duration = time.time() - start
            logger.info(f"[{name}] DONE - {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"[{name}] ERROR - {duration:.2f}s - {str(e)}")
            raise
    return wrapper

Zusätzlich: LangSmith für detailliertes Tracing — zeigt jeden LLM-Call, State-Übergang und Token-Berechnung visuell:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

6. Human-in-the-Loop

Für kritische Entscheidungen sollte ein Mensch bestätigen. LangGraph unterstützt das nativ mit interrupt_before:

app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["publisher"]  # Pausiert vor Publisher
)

# Erste Ausführung: läuft bis vor Publisher
result = app.invoke(initial_state, config={"configurable": {"thread_id": "1"}})

# Mensch prüft und genehmigt
approval = input("Genehmigen? (y/n): ")
result = app.invoke(
    {"approved": approval == "y"},
    config={"configurable": {"thread_id": "1"}}
)

Wann Human-in-the-Loop? Veröffentlichung von Inhalten, Code-Änderungen in Produktion, finanzielle Konsequenzen, niedrige Agenten-Confidence.

Anti-Patterns — Was Du vermeiden solltest

1. God Agent

Ein Agent, der alles macht. Der Prompt wird riesig → LLM verliert Fokus → schlechte Qualität. Schwer zu debuggen, teuer im Betrieb. Lösung: Aufteilen in 3-5 spezialisierte Agenten.

2. Unbegrenzte Schleifen

Agenten, die sich gegenseitig aufrufen ohne Abbruchbedingung. Lösung: Maximale Iterationen festlegen und Notbremse implementieren:

MAX_ITERATIONS = 5

def route_after_review(state):
    if state.get("approved"):
        return "publish"
    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        return "human_review"  # Notbremse
    return "writer"

3. Keine Validierung

Blindes Vertrauen in Agenten-Ausgaben. LLMs halluzinieren und erfinden Quellen. Lösung: Jede Ausgabe validieren:

def validate_research(result: str) -> bool:
    if not result or len(result) < 100:
        return False
    if "http" not in result:  # Mindestens eine Quelle
        return False
    return True

4. Zu viele Agenten

Mehr Agenten = mehr Koordinationsaufwand, höhere Kosten, mehr Fehlerquellen. Faustregel: 3-5 Agenten. Wenn Du mehr brauchst, teile in mehrere unabhängige Crews auf.

5. Keine Kostenkontrolle

Ohne Cost-Tracking können Multi-Agent-Systeme schnell Hunderte Dollar pro Durchlauf kosten. Lösung: CostTracker mit Budget-Limit (siehe oben).

Framework-Vergleich für Orchestrierung

Feature	LangGraph	CrewAI	Custom
Pipeline	✅	✅ (sequential)	✅
Hub-and-Spoke	✅ (conditional edges)	✅ (hierarchical)	✅
DAG/Parallel	✅ (native)	❌	✅ (asyncio)
Event-Driven	❌ (nicht nativ)	❌	✅ (custom bus)
Cycles	✅ (conditional edges)	Eingeschränkt	✅
Human-in-the-Loop	✅ (interrupt)	Manuell	✅
Checkpointing	✅ (native)	❌	✅ (custom)

Prüfungsrelevante Punkte

• Orchestrierung: Koordination mehrerer Agenten mit Aufgabenteilung, Kommunikation, Kontrolle, Fehlerbehandlung und Ressourcenmanagement
• 4 Patterns: Pipeline (sequenziell), Hub-and-Spoke (zentral), DAG (parallel), Event-Driven (reaktiv)
• Best Practices: Klare Rollen (SRP), explizite typisierte States, Retry-Logic mit max_retries, Kostenkontrolle mit CostTracker, Observability mit Logging + LangSmith, Human-in-the-Loop für kritische Entscheidungen
• Anti-Patterns: God Agent, unbegrenzte Schleifen, keine Validierung, zu viele Agenten, keine Kostenkontrolle
• Framework-Wahl: LangGraph für komplexe Graphen, CrewAI für schnelle Prototypen, Custom für Event-Driven

FAQ

Wann brauche ich Multi-Agent statt Single-Agent? Wenn die Aufgabe verschiedene Expertisen benötigt, Iteration nötig ist (Writer ↔ Reviewer), Parallelität möglich ist, oder die Aufgabe zu komplex für einen Prompt ist.

Wieviele Agenten sind optimal? 3-5 für die meisten Anwendungsfälle. Mehr als 8 werden schwer zu koordinieren und teuer. Für komplexe Anforderungen besser in mehrere unabhängige Crews/Graphen aufteilen.

Welches Orchestrierungs-Pattern ist am besten? Pipeline für einfache lineare Aufgaben, Hub-and-Spoke für dynamisches Routing, DAG für parallele Aufgaben, Event-Driven für lose gekoppelte Systeme. Die meisten Produktions-Systeme verwenden eine Kombination.

Wie verhindere ich Endlosschleifen? Setze immer eine maximale Iterations-Anzahl. Implementiere eine Notbremse, die bei zu vielen Iterationen auf Human-in-the-Loop oder END weiterleitet.

Wie kontrolliere ich Kosten in Multi-Agent-Systemen?

1. CostTracker mit Budget-Limit. 2) Günstige Modelle für einfache Tasks. 3) Caching für gleiche Prompts. 4) Early Stopping. 5) Lokale Modelle für Entwicklung.

Brauche ich LangSmith in Produktion? Stark empfohlen. Ohne Tracing bist Du blind — Du weißt nicht, welcher Agent wie lange braucht und wo Fehler entstehen. LangSmith hat eine kostenlose Stufe für kleine Projekte.

Empfohlene Literatur