Wann man dazu greift
- Die Aufgabe sich natürlich in Phasen aufteilt, die jeweils ein überprüfbares Zwischenprodukt liefern.
- Kontrolle wichtiger ist als Autonomie.
Sequential Pipeline.
Feste Schritte. Keine Rückkanten.
Mehrere Schritte werden in fester Reihenfolge ausgeführt, wobei jeder Schritt die Ausgabe des vorherigen nutzt.
Wann es nach hinten losgeht
- Der Workflow stark verzweigt oder Ergebnisse dynamisch neue Ziele erzeugen.
- Schritte ohne Abhängigkeiten parallelisiert werden können.
Die Abwägung
Hohe Vorhersagbarkeit des Kontrollflusses wird auf Kosten geringer Flexibilität erkauft; der Ausgabeinhalt bleibt an jedem LLM-Knoten nicht-deterministisch.
Eine Form, die du auf eine Serviette zeichnen kannst.
Jeder Auftrag folgt derselben vierstufigen Kette. Vorhersagbarkeit ist der Zweck — Anpassungsfähigkeit gehört in ein anderes Pattern.
- Entwickler verdrahten jede Stufe und Kante im Code, bevor der Lauf startet.
- Jede Stufe transformiert den Zustand und reicht ihn genau einmal weiter.
- Der Graph ist bei jedem Lauf identisch; nur die Modellausgabe in den Knoten variiert.
Ein realer Ablauf, Schritt für Schritt.
Aufgabe"Fasse diesen Vorfallbericht zusammen und erstelle eine Behebungs-Checkliste."
Stufe 1ingest_report(text) → { paragraphs[], entities[] }
Stufe 2analyze(paragraphs, entities) → { rootCause, severity, affectedSystems }
Stufe 3synthesize(analysis) → { summary, checklist:[fix1, fix2, fix3] }
Stufe 4respond(payload) → MD-Ausgabe
AntwortMarkdown-Zusammenfassung + Checkliste zurückgegeben. Gesamt: 4 LLM-Aufrufe in fester Reihenfolge; keine Verzweigung, keine Wiederholungen innerhalb der Pipeline.
Die Schleife ist schon eingebaut.
LangGraphpython
from typing import TypedDict
from langgraph.graph import StateGraph, END
class State(TypedDict):
raw: str
parsed: dict | None
analysis: dict | None
payload: dict | None
answer: str | None
def ingest(state): return {"parsed": parse(state["raw"])}
def analyze(state): return {"analysis": analyze_llm(state["parsed"])}
def synthesize(state): return {"payload": synth_llm(state["analysis"])}
def respond(state): return {"answer": render_md(state["payload"])}
g = StateGraph(State)
g.add_node("ingest", ingest)
g.add_node("analyze", analyze)
g.add_node("synthesize", synthesize)
g.add_node("respond", respond)
g.add_edge("ingest", "analyze")
g.add_edge("analyze", "synthesize")
g.add_edge("synthesize", "respond")
g.add_edge("respond", END)
g.set_entry_point("ingest")
pipeline = g.compile()Drei Wege, wie dieses Pattern Probleme verursachen kann.
Jeden Auftrag als Pipeline behandeln
Aufträge, die eine andere Form benötigen (nach Inhalt routen, bei Fehler wiederholen), werden in dieselben vier Stufen gezwängt und verarbeiten entweder zu viel oder schlagen stillschweigend fehl.
Lösung · Zu Routing oder Orchestrator-Workers greifen, wenn die Arbeit nicht einheitlich ist; Sequential Pipeline nur für echte homogene Abläufe behalten.
Kein Ort, um ein schlechtes Zwischenergebnis abzufangen
Stufe 2 produziert eine Ausgabe mit geringer Konfidenz, aber die Pipeline hat keine bedingte Kante zum Stoppen oder Eskalieren — die schlechte Analyse fließt weiter und korrumpiert die Antwort.
Lösung · Ein HITL-Gate nach der riskanten Stufe hinzufügen oder die Pipeline in eine Evaluator-Optimizer-Schleife kapseln, die einen Neustart anfordern kann.
Stufengrenzen, die Funktionsnamen statt Zustand widerspiegeln
Autoren bilden Stufen nach 'was die nächste Funktion tut' statt nach 'welcher neue Zustand beobachtbar ist'. Die Pipeline wird zu einer Kette von Mikro-LLM-Aufrufen mit überlappender Verantwortung.
Lösung · Jede Stufe durch das Zustandsfeld definieren, das sie erzeugt (geparst → Analyse → Payload → Antwort); Stufen mit gemeinsamer Ausgabe zusammenfassen.
Wo Sequential Pipeline nativ ist.
CrewAINativ
Google ADKNativ
Microsoft Agent FrameworkNativ
AWS StrandsNativ
Anthropic Cookbookals Rezept dokumentiertAnpassbar