Wann man dazu greift
- Die erforderlichen Tool-Aufrufe zu Beginn identifizierbar sind.
- LLM-Aufrufe drastisch reduziert werden müssen.
ReWOO.
Planen. Bündeln. Zusammenführen.
Der Agent plant alle notwendigen Tool-Aufrufe im Voraus, führt sie gebündelt aus und nutzt die aggregierten Ergebnisse für die finale Antwort.
Wann es nach hinten losgeht
- Tool-Ergebnisse stark verzweigen.
- Interaktive Fehlerbehandlung pro Zwischenschritt kritisch ist.
Die Abwägung
Deutlich niedrigere LLM-Kosten und Latenz, jedoch auf Kosten der Anpassungsfähigkeit während der Ausführung.
Eine Form, die du auf eine Serviette zeichnen kannst.
Im Voraus planen, alle Tools in einem Batch ausführen, Ergebnisse zusammenführen. Keine schrittweise Beobachtung; günstiger als ReAct, weniger adaptiv.
Ein realer Ablauf, Schritt für Schritt.
Aufgabe"Schätze den CO₂-Fußabdruck einer Europareise durch fünf Städte."
Planplanner(task) → [flight_lookup, train_lookup, hotel_lookup, carbon_coefficient]
BatchAlle vier Tools laufen parallel: flight_lookup(cities), train_lookup(cities), hotel_lookup(cities), carbon_coefficient(mode='air')
SammelnErgebnisse: { flights: [...], trains: [...], hotels: [...], kg_per_km: 0.255 }
Lösensolver(results) → { total_kg: 1847, breakdown: [...], recommendations: [...] }
Antwortenrespond(solver_output) → Markdown-Schätzung
AntwortMarkdown-Schätzung mit Gesamt-kg und Aufschlüsselung pro Strecke. Gesamt: 1 Plan + 4 parallele Tool-Aufrufe + 1 Lösung.
Die Schleife ist schon eingebaut.
LangGraphpython
from typing import TypedDict, Any
from langgraph.graph import StateGraph, END
class Plan(TypedDict):
tools: list[str]
args: list[dict]
class State(TypedDict):
task: str
plan: Plan | None
tool_outputs: dict[str, Any]
answer: str | None
def planner(state):
return {"plan": plan_llm(state["task"])}
# Parallel tool nodes via fan-out
def flight_lookup(state):
return {"tool_outputs": {"flights": search_flights(state["plan"]["args"][0])}}
def train_lookup(state): ...
def hotel_lookup(state): ...
def carbon_lookup(state): ...
def solver(state):
return {"answer": solve_llm(state["tool_outputs"])}
g = StateGraph(State)
g.add_node("planner", planner)
g.add_node("flight", flight_lookup)
g.add_node("train", train_lookup)
g.add_node("hotel", hotel_lookup)
g.add_node("carbon", carbon_lookup)
g.add_node("solver", solver)
# Fan-out from planner to all tools
g.add_edge("planner", "flight")
g.add_edge("planner", "train")
g.add_edge("planner", "hotel")
g.add_edge("planner", "carbon")
for node in ["flight", "train", "hotel", "carbon"]:
g.add_edge(node, "solver")
g.add_edge("solver", END)
g.set_entry_point("planner")
rewoo = g.compile()Zwei Wege, wie dieses Pattern Probleme verursachen kann.
Plan war falsch, aber kein Beobachtungsschritt fängt es ab
Der Planer ging von einer Direktverbindung aus. Das Flug-Tool liefert ein leeres Ergebnis zurück, aber es gibt keine Beobachtungskante für Neuplanung — der Solver übernimmt falsche Daten in die Antwort.
Lösung · Validierungsknoten nach jedem Tool hinzufügen oder den Batch in eine Evaluator-Optimizer-Schleife kapseln, die einen Neuplanungsaufruf anfordern kann.
Tool-Fehler hat keinen Fehlerbehandlungspfad
Ein Tool im Batch läuft in einen Timeout. Der Solver empfängt unvollständige Daten und produziert eine irreführende Schätzung.
Lösung · Einen Ergebnis-Aggregator verwenden, der fehlende Schlüssel markiert, oder eine Retry-Orchestrierungsschicht vor dem Solver einbauen.