Wann man dazu greift
- Teilaufgaben erst zur Laufzeit erkennbar werden.
- Worker hochspezialisierte Funktionen übernehmen.
- Aggregation zentral kontrolliert bleiben muss.
Orchestrator-Workers.
Zerlegen. Dispatchen. Zusammenführen.
Ein Orchestrator zerlegt eine Aufgabe dynamisch und weist Teilaufgaben spezialisierten Workern zu, die Aggregation bleibt zentral gesteuert.
Wann es nach hinten losgeht
- Ein statischer Workflow ausreicht.
- Workern klare Verantwortlichkeiten fehlen.
- Der Orchestrator zum schwerwiegenden Engpass wird.
Die Abwägung
Hochflexible Delegation wird gegen erheblichen Koordinations- und Integrationsaufwand erkauft.
Eine Form, die du auf eine Serviette zeichnen kannst.
Ein Orchestrator zerlegt den Auftrag, dispatcht parallel an Worker, dann führt er die Ergebnisse zusammen.
- Ein zentrales LLM plant; jeder Spezialist ist als typisierter Tool-Aufruf verfügbar.
- Der Orchestrator wählt ein Tool, ruft es auf und erhält einen Rückgabewert.
- Nach jedem Aufruf kehrt die Kontrolle zum Zentrum zurück — Spezialisten sprechen einander nie an.
Ein realer Ablauf, Schritt für Schritt.
Aufgabe"Recherchiere drei Konkurrenten und erstelle einen Vergleichsbericht."
Zerlegenorchestrator(task) → ['research_acme', 'research_globalcorp', 'research_stark']
DispatchSend(subtask_1) → Worker A, Send(subtask_2) → Worker B, Send(subtask_3) → Worker C
Parallele ArbeitWorker rufen ab und fassen parallel zusammen (kein gemeinsamer Zustand bis zur Zusammenführung).
Zusammenführenorchestrator([summary_A, summary_B, summary_C]) → comparison_table + narrative
Rendernrespond(folded) → Markdown-Bericht
AntwortVergleichsbericht mit Tabelle und Narrativ. Gesamt: 1 Zerlegen + 3 parallele Worker-Aufrufe + 1 Zusammenführen.
Die Schleife ist schon eingebaut.
LangGraphpython
from langgraph.graph import StateGraph, END
from langgraph.constants import Send
class State(TypedDict):
task: str
subtasks: list[str]
results: dict[str, str]
answer: str | None
def decompose(state):
return {"subtasks": planner_llm(state["task"])}
def worker(state):
# Each invocation receives one subtask
return {"results": {state["subtask"]: research_llm(state["subtask"])}}
def fold(state):
return {"answer": synthesizer_llm(state["results"])}
g = StateGraph(State)
g.add_node("decompose", decompose)
g.add_node("worker", worker)
g.add_node("fold", fold)
# Fan-out: one Send per subtask
g.add_conditional_edges("decompose", lambda s: [Send("worker", {"subtask": t}) for t in s["subtasks"]])
g.add_edge("worker", "fold")
g.add_edge("fold", END)
g.set_entry_point("decompose")
orchestrator = g.compile()Zwei Wege, wie dieses Pattern Probleme verursachen kann.
Orchestrator überzerlegt — zu viele kleine Worker
Der Orchestrator teilt die Aufgabe in 20 Mikro-Aufgaben auf. Der Overhead von Dispatch und Zusammenführen dominiert die eigentliche Arbeit.
Lösung · Eine Mindestchunkgröße festlegen; Teilaufgaben zusammenführen, die ein Kontextfenster oder einen Tool-Aufruf teilen.
Worker-Ausgaben widersprechen sich und der Zusammenführungsschritt wählt stillschweigend eine
Zwei Worker liefern widersprüchliche Daten. Der Zusammenführungsschritt mittelt oder wählt zufällig aus, ohne den Konflikt sichtbar zu machen.
Lösung · Den Zusammenführungsschritt dazu zwingen, Konfidenzwerte pro Quelle auszugeben, oder einen expliziten Konsistenzprüfknoten vor der finalen Antwort hinzufügen.
Wo Orchestrator-Workers nativ ist.
LangGraphNativ
CrewAINativ
Google ADKNativ
Microsoft Agent FrameworkNativ
AWS StrandsNativ
Anthropic Cookbookals Rezept dokumentiertAnpassbar