Drei Muster für Agent-Orchestrierung, die die Produktion überstanden haben

Die Landschaft der Agent-Orchestrierung

Im vergangenen Jahr haben wir LLM-Agent-Systeme in drei Domänen eingesetzt:

1. Kundensupport-Automatisierung — 12.000 Tickets/Monat, 8 integrierte Tools

2. Business-Intelligence-Analyse — 200 Analysten, 15 Datenquellen

3. Rechtsdokumentenverarbeitung — 50k Dokumente/Monat, 6 Extraktionspipelines

Jedes Deployment lehrte uns, welche Orchestrierungsmuster in der Theorie funktionieren und welche unter Produktionsstress standhalten — Tool-Timeouts, API-Ratenlimits, mehrdeutige Abfragen und Benutzererwartungen an Antworten unter 3 Sekunden.

Dieser Artikel katalogisiert drei Muster, die überstanden haben: Router, Planner-Executor und Critic.

Muster 1: Router (Einfaches Dispatch)

Architektur

                Benutzerabfrage
                    ↓
            ┌───────────────┐
            │  Router LLM   │  "Welches Tool behandelt das?"
            └───────┬───────┘
                    │
        ┌───────────┼───────────┬───────────┐
        ↓           ↓           ↓           ↓
    [Tool A]    [Tool B]    [Tool C]    [Tool D]
    Suche       Taschenrechner Wetter    Kalender
        ↓           ↓           ↓           ↓
                Antwort (von einem Tool)

Wann verwenden

• Mehrere spezialisierte Tools mit klaren, nicht überlappenden Domänen.

• Einzelne Schritte (ein Tool-Aufruf → Ergebnis).

• Latenzsensitive Anwendungen (<1s Antwortzeit).

Implementierung

router_agent.py
from typing import Dict, Callable
class RouterAgent:
    """Einfacher Routing-Agent — dispatch an ein Tool."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
        self.tool_descriptions = self._generate_tool_docs()
    def _generate_tool_docs(self) -> str:
        """Tool-Dokumentation für Router-Prompt generieren."""
        docs = []
        for name, tool in self.tools.items():
            docs.append(f"- {name}: {tool.__doc__}")
        return "\n".join(docs)
    async def route(self, query: str) -> str:
        """Abfrage an passendes Tool leiten."""
        router_prompt = f"""
Sie sind ein Tool-Router. Wählen Sie das einzelne beste Tool zur Beantwortung einer Benutzerabfrage.
Verfügbare Tools:
{self.tool_descriptions}
Benutzerabfrage: {query}
Antworten Sie mit JSON: {{"tool": "tool_name", "reasoning": "warum dieses Tool"}}
"""
        routing_decision = await llm.generate(router_prompt)
        tool_name = json.loads(routing_decision)['tool']
        # Ausgewähltes Tool ausführen
        if tool_name not in self.tools:
            return f"Fehler: Unbekanntes Tool {tool_name}"
        return await self.toolstool_name
Verwendung
tools = {
    "search": search_knowledge_base,
    "calculator": calculate_expression,
    "weather": get_weather_forecast,
    "calendar": check_calendar_availability
}
agent = RouterAgent(tools)
response = await agent.route("Wie wird das Wetter morgen in Paris?")

Produktionsdaten (Kundensupport)

Metrik	Wert
Behandelte Abfragen	12.000/Monat
Korrektes Routing	94%
Durchschnittliche Latenz	820ms
p95-Latenz	1.2s
Mehrdeutiges Routing	6% (Eskalation an Mensch)

Stärken

✅ Niedrige Latenz — Einzelner LLM-Aufruf + eine Tool-Ausführung ✅ Vorhersagbar — Lineare Ausführung, einfach zu durchdenken ✅ Debugging-fähig — Einfach zu loggen: "Abfrage → Routing-Entscheidung → Tool → Ergebnis" ✅ Kosteneffektiv — Minimale LLM-Aufrufe

Schwächen

❌ Kein Tool-Chaining — Kann Tools nicht kombinieren ("Suche nach X, berechne dann Y") ❌ Routing-Fehler sind fatal — Falsche Tool-Auswahl = falsche Antwort ❌ Mehrdeutige Abfragen scheitern — "Termin buchen, wenn es nicht regnet" erfordert zwei Tools

Produktions-Lektionen

Lektion 1: Einen Fallback-Klassifikator aufbauen

Wenn die Routing-Konfidenz niedrig ist (<70%), an Mensch eskalieren:

routing_confidence = routing_decision['confidence']
if routing_confidence < 0.70:
    return escalate_to_human(query, reason="mehrdeutiges Routing")

Lektion 2: Routing-Entscheidungen zwischenspeichern

Häufige Abfragen ("Bestellstatus prüfen") routen jedes Mal gleich:

@cache(ttl=3600)
def route_query(query: str):
    # Routing 1 Stunde zwischenspeichern
    return router.route(query)

Lektion 3: Routing-Genauigkeit überwachen

Verfolgen Sie, welche Tools ausgewählt wurden vs. was Benutzer tatsächlich brauchten:

Routing-Entscheidungen loggen
log_routing_decision(
    query=query,
    selected_tool=tool_name,
    user_satisfaction=feedback  # Interaktion nachverfolgen
)
Wöchentliche Analyse
routing_errors = query_logs.filter(user_satisfaction < 3)
print(f"Top falsch geroutete Abfragen: {routing_errors.most_common(10)}")

Ergebnis: Wir verbesserten die Routing-Genauigkeit von 87% → 94%, indem wir falsch geroutete Abfragen nachtrainierten.

Muster 2: Planner-Executor (Mehrstufiges Reasoning)

Architektur

            Benutzerabfrage: "Vergleiche Umsatz Q1 vs Q2"
                    ↓
            ┌───────────────┐
            │  Planner LLM  │  Ausführungsplan generieren
            └───────┬───────┘
                    ↓
            Plan: [Schritt 1, Schritt 2, Schritt 3]
            1. Q1-Umsatz aus DB abrufen
            2. Q2-Umsatz aus DB abrufen
            3. Differenz berechnen
                    ↓
            ┌───────────────┐
            │   Executor    │  Plan nacheinander ausführen
            └───────┬───────┘
                    ↓
        ┌───────────┼───────────┐
        ↓           ↓           ↓
    [DB abfragen] [DB abfragen] [Berechnen]
        ↓           ↓           ↓
    $120K       $145K        +$25K (+21%)
                    ↓
            Finale Antwort

Wann verwenden

• Mehrstufige Workflows, die Tool-Zusammensetzung erfordern.

• Dynamische Tool-Auswahl (Tool-Sequenz kann nicht im Voraus vorhergesagt werden).

• Strukturierte Aufgaben (Datenanalyse, Berichtgenerierung).

Implementierung

planner_executor_agent.py
from typing import List, Dict
import json
class PlannerExecutorAgent:
    """Agent, der vor der Ausführung plant."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
    async def plan(self, query: str) -> List[Dict]:
        """Ausführungsplan generieren."""
        planner_prompt = f"""
Sie sind ein Aufgabenplaner. Zerlegen Sie diese Abfrage in ausführbare Schritte mit verfügbaren Tools.
Verfügbare Tools:
{self._tool_docs()}
Benutzerabfrage: {query}
Generieren Sie einen Plan als JSON-Array:
[
  {{"step": 1, "tool": "tool_name", "input": "...", "output_var": "var1"}},
  {{"step": 2, "tool": "tool_name", "input": "verwende {{var1}}", "output_var": "var2"}},
  ...
]
"""
        plan_json = await llm.generate(planner_prompt)
        return json.loads(plan_json)
    async def execute(self, plan: List[Dict]) -> Dict:
        """Plan Schritt für Schritt ausführen."""
        context = {}  # Zwischenergebnisse speichern
        for step in plan:
            tool_name = step['tool']
            tool_input = step['input']
            # Variablen aus Kontext ersetzen
            for var, value in context.items():
                tool_input = tool_input.replace(f"{{{var}}}", str(value))
            # Tool ausführen
            result = await self.toolstool_name
            # Ergebnis im Kontext speichern
            output_var = step.get('output_var', f"step_{step['step']}")
            context[output_var] = result
            print(f"Schritt {step['step']}: {tool_name}({tool_input}) → {result}")
        return context
    async def run(self, query: str) -> str:
        """Planen und ausführen."""
        plan = await self.plan(query)
        context = await self.execute(plan)
        # Finale Antwort mit Kontext generieren
        final_prompt = f"""
Benutzerabfrage: {query}
Ausführungsergebnisse:
{json.dumps(context, indent=2)}
Geben Sie eine natürlichsprachliche Antwort an den Benutzer.
"""
        return await llm.generate(final_prompt)
Verwendung
tools = {
    "sql_query": execute_sql,
    "calculator": calculate,
    "search_docs": search_documentation,
    "send_email": send_email
}
agent = PlannerExecutorAgent(tools)
response = await agent.run("Vergleiche Umsatz Q1 vs Q2 und sende Zusammenfassung per E-Mail an CFO")

Produktionsdaten (Business Intelligence)

Metrik	Wert
Behandelte Abfragen	1.200/Monat
Erfolgreiche Abschlüsse	89%
Durchschnittliche Latenz	3.2s
p95-Latenz	8.4s
Planfehler	11% (ungültiges Tool, falsche Reihenfolge)

Stärken

✅ Behandelt komplexe Workflows — Multi-Tool-Zusammensetzung ✅ Flexibel — Passt sich dynamisch an die Abfragenkomplexität an ✅ Transparent — Plan ist menschenlesbar, debugging-fähig ✅ Wiederherstellbar — Einzelne Schritte können bei Fehlern wiederholt werden

Schwächen

❌ Höhere Latenz — N+1 LLM-Aufrufe (1 für Planung, N für Ausführung) ❌ Pläne können falsch sein — Ungültige Tool-Auswahl, falsche Reihenfolge, fehlende Schritte ❌ Fehlerfortpflanzung — Früher Schrittfehler bricht den gesamten Plan ❌ Kosten skalieren mit Schritten — 5-Schritt-Plan = 6 LLM-Aufrufe

Produktions-Lektionen

Lektion 1: Pläne vor der Ausführung validieren

Vertrauen Sie LLM-generierten Plänen nicht blind:

def validate_plan(plan: List[Dict]) -> bool:
    """Auf häufige Fehler prüfen."""
    for step in plan:
        # Prüfen ob Tool existiert
        if step['tool'] not in self.tools:
            raise PlanError(f"Unbekanntes Tool: {step['tool']}")
        # Variablenabhängigkeiten prüfen
        required_vars = extract_variables(step['input'])
        available_vars = [s['output_var'] for s in plan[:step['step']-1]]
        for var in required_vars:
            if var not in available_vars:
                raise PlanError(f"Variable {var} nicht verfügbar bei Schritt {step['step']}")
    return True

Lektion 2: Schritt-Level-Wiederholungen hinzufügen

Netzwerkfehler und Ratenlimits passieren. Wiederholen Sie einzelne Schritte:

from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def execute_step(tool_name: str, tool_input: str):
    return await self.toolstool_name

Lektion 3: Plan-Caching für ähnliche Abfragen implementieren

Abfragen wie "Vergleiche Umsatz Q1 vs Q2" haben ähnliche Pläne:

Plan-Vorlagen zwischenspeichern
plan_template = cached_plans.get(query_category)
if plan_template:
    plan = instantiate_template(plan_template, query_params)
else:
    plan = await self.plan(query)

Ergebnis: Reduzierte Planungslatenz um 40% für wiederkehrende Abfragemuster.

Muster 3: Critic (Iterative Verfeinerung)

Architektur

            Benutzerabfrage: "Entwerfe eine professionelle Entschuldigungs-E-Mail"
                    ↓
            ┌───────────────┐
            │ Generator LLM │  Initiale Antwort generieren
            └───────┬───────┘
                    ↓
            Entwurf v1: "Entschuldigen Sie sich für das Problem..."
                    ↓
            ┌───────────────┐
            │  Critic LLM   │  Qualität bewerten
            └───────┬───────┘
                    ↓
        [Bestanden: Score ≥ 8/10] ────→ Antwort zurückgeben
                    │
        [Nicht bestanden: Score < 8/10]
                    ↓
            Feedback: "Zu locker. Fügen Sie spezifische Details hinzu."
                    ↓
            ┌───────────────┐
            │  Generator    │  Mit Feedback neu generieren
            └───────┬───────┘
                    ↓
            Entwurf v2: "Wir entschuldigen uns aufrichtig für [spezifisches Problem]..."
                    ↓
            [Bis zu max_iterations=3 wiederholen]

Wann verwenden

• Qualitätskritische Ausgaben (Rechtsdokumente, Kundenkommunikation).

• Iterative Verfeinerung erforderlich.

• Latenztoleranz (Benutzer erwarten 3-10s für komplexe Aufgaben).

Implementierung

critic_agent.py
from typing import Tuple
class CriticAgent:
    """Agent mit Selbstkritik-Schleife."""
    def __init__(self, max_iterations: int = 3):
        self.max_iterations = max_iterations
    async def generate(self, query: str, feedback: str = None) -> str:
        """Antwort generieren (mit optionalem Feedback)."""
        if feedback:
            prompt = f"""
Benutzeranfrage: {query}
Der vorherige Versuch erhielt dieses Feedback:
{feedback}
Generieren Sie eine verbesserte Antwort, die das Feedback berücksichtigt.
"""
        else:
            prompt = f"Benutzeranfrage: {query}\n\nGenerieren Sie eine Antwort."
        return await llm.generate(prompt)
    async def critique(self, query: str, response: str) -> Tuple[float, str]:
        """Antwortqualität bewerten (Score 0-10, Feedback)."""
        critic_prompt = f"""
Bewerten Sie diese Antwort auf Qualität, Genauigkeit und Professionalität.
Benutzeranfrage: {query}
Antwort: {response}
Geben Sie an:
1. Score (0-10)
2. Spezifisches Feedback zur Verbesserung
Format: {{"score": X, "feedback": "..."}}
"""
        critique = await llm.generate(critic_prompt)
        result = json.loads(critique)
        return result['score'], result['feedback']
    async def run(self, query: str, min_score: float = 8.0) -> Dict:
        """Mit iterativer Verfeinerung generieren."""
        history = []
        for iteration in range(self.max_iterations):
            # Antwort generieren (mit Feedback der vorherigen Iteration)
            feedback = history[-1]['feedback'] if history else None
            response = await self.generate(query, feedback)
            # Antwort kritisieren
            score, feedback = await self.critique(query, response)
            history.append({
                "iteration": iteration + 1,
                "response": response,
                "score": score,
                "feedback": feedback
            })
            # Prüfen ob Qualitätsschwelle erreicht
            if score >= min_score:
                return {
                    "response": response,
                    "iterations": iteration + 1,
                    "final_score": score,
                    "history": history
                }
        # Maximale Iterationen erreicht, besten Versuch zurückgeben
        best = max(history, key=lambda x: x['score'])
        return {
            "response": best['response'],
            "iterations": self.max_iterations,
            "final_score": best['score'],
            "history": history,
            "warning": "Maximale Iterationen erreicht ohne Qualitätsschwelle zu treffen"
        }
Verwendung
agent = CriticAgent(max_iterations=3)
result = await agent.run("Entwerfe eine professionelle Entschuldigung für verspätete Lieferung")
print(f"Finale Antwort (Score: {result['final_score']}):\n{result['response']}")

Produktionsdaten (Rechtsdokumentgenerierung)

Metrik	Wert
Generierte Dokumente	800/Monat
Erfolg im ersten Versuch	62% (Score ≥ 8/10)
Erfolg im zweiten Versuch	89%
Erfolg im dritten Versuch	96%
Durchschnittliche Latenz	4.2s
p95-Latenz	11.8s

Stärken

✅ Höhere Ausgabequalität — Selbstkorrektur fängt Fehler ab ✅ Anpassungsfähig — Lernt innerhalb einer Sitzung aus eigenen Fehlern ✅ Transparent — Kritik-Feedback erklärt Qualitätsprobleme ✅ Graceful Degradation — Gibt besten Versuch zurück, wenn Schwelle nicht erreicht

Schwächen

❌ Hohe Latenz — 2-6 LLM-Aufrufe (2x pro Iteration) ❌ Teuer — Kosten skalieren mit Iterationen ❌ Kann endlos schleifen — max_iterations muss gesetzt werden ❌ Critic kann falsch liegen — Falsch-negative (gute Antwort niedrig bewertet)

Produktions-Lektionen

Lektion 1: Aggressive max_iterations-Grenze setzen

Unsere Anfangsgrenze war 5. 12% der Abfragen erreichten diese Grenze (verschwendeten 10 LLM-Aufrufe). Reduziert auf 3:

Kostenanalyse
avg_cost_per_llm_call = $0.02
max_iterations = 5 → avg_cost = $0.20 (10 Aufrufe)
max_iterations = 3 → avg_cost = $0.12 (6 Aufrufe)
40% Kostenreduktion mit minimalem Qualitätseinfluss

Lektion 2: Schnelle Modelle für Kritik verwenden

Critic braucht keine Frontier-Modell-Intelligenz. Wir verwenden GPT-4 für Generierung, GPT-3.5-turbo für Kritik:

async def critique(self, query: str, response: str):
    # Billigeres, schnelleres Modell für Kritik verwenden
    critique = await llm.generate(critic_prompt, model="gpt-3.5-turbo")
    # ...

Ergebnis: Reduzierte Kritikslatenz um 60% (600ms → 240ms) bei gleicher Genauigkeit.

Lektion 3: Frühen Stopp bei "perfekten" Scores hinzufügen

Wenn der erste Versuch 9.5/10 punktet, überspringen Sie weitere Iterationen:

if score >= 9.5:  # "Perfekt"-Schwelle
    return early_with_success(response, score)

Latenzvergleich: Reale Produktionsdaten

Muster	Durchschn. Latenz	p95-Latenz	p99-Latenz	LLM-Aufrufe
Router	820ms	1.2s	1.8s	1
Planner-Executor (3 Schritte)	3.2s	8.4s	14.1s	4
Critic (durchschn. 1.8 Iterationen)	4.2s	11.8s	18.5s	3.6

Kostenvergleich

Annahmen:

• GPT-4 Input: $0.01/1K Tokens.

• GPT-4 Output: $0.03/1K Tokens.

• Durchschnittliche Abfrage: 200 Input-Tokens.

• Durchschnittliche Antwort: 500 Output-Tokens.

Muster	LLM-Aufrufe	Durchschn. Kosten
Router	1	$0.017
Planner-Executor	4	$0.068
Critic	3.6	$0.061

Entscheidungsmatrix: Welches Muster verwenden?

Wählen Sie Router, wenn:

✅ Einzeltool-Dispatch reicht ✅ Latenz <1s erforderlich ✅ Abfragenrouting ist eindeutig ✅ Kosten pro Abfrage sind wichtig

Wählen Sie Planner-Executor, wenn:

✅ Mehrstufige Workflows benötigt ✅ Tool-Zusammensetzung erforderlich ✅ Latenz <5s akzeptabel ✅ Transparenz (sichtbarer Plan) ist wertvoll

Wählen Sie Critic, wenn:

✅ Ausgabequalität ist mission-kritisch ✅ Latenz <10s akzeptabel ✅ Selbstkorrektur schafft Mehrwert ✅ Erstentwurfsqualität ist unzureichend

Hybride Muster, die wir getestet haben

Muster 4: Router + Planner-Executor

Einfache Abfragen an einzelne Tools leiten, komplexe an Planner:

if query_complexity(query) < 0.5:
    return router.route(query)  # Schneller Pfad
else:
    return planner_executor.run(query)  # Langsamer Pfad

Ergebnis: 70% der Abfragen nehmen den schnellen Pfad (durchschn. 850ms), 30% den langsamen Pfad (durchschn. 3.5s). Gesamtdurchschnitt: 1.6s.

Muster 5: Planner-Executor + Critic

Planen, ausführen, dann finale Antwort kritisieren:

context = await planner_executor.execute(query)
final_response = await generate_response(context)
score, feedback = await critic.critique(query, final_response)
if score < 8.0:
    final_response = await regenerate_with_feedback(context, feedback)

Ergebnis: Für hochrangige Berichte verwendet. Latenz: 8-12s. Qualität: 98% Benutzerzufriedenheit.

Fazit

Nach 18+ Monaten in der Produktion:

1. Router behandelt 80% der Abfragen mit hervorragender Latenz

2. Planner-Executor glänzt bei mehrstufigen Workflows, erfordert aber Planvalidierung

3. Critic verbessert die Qualität um 15-20%, verdoppelt aber Kosten und Latenz

Unsere Standard-Empfehlung:

• Beginnen Sie mit Router für MVP.

• Fügen Sie Planner-Executor hinzu, wenn Benutzer mehrstufige Aufgaben anfordern.

• Reservieren Sie Critic für qualitätskritische Ausgaben (Recht, Finanzen, Medizin).

Das beste Muster hängt von Ihrem Latenzbudget, Ihren Qualitätsanforderungen und Ihren Kostenbeschränkungen ab. Überengineering Sie nicht — deployen Sie einfach, steigern Sie die Komplexität nach Bedarf.

---

Bauen Sie Agent-Systeme? Kontaktieren Sie uns um Ihre Architektur zu besprechen. Wir bieten Agent-Design-Beratung, Implementierungsunterstützung und Produktionsoptimierungsdienste an.