Dies ist die vollständige Lehr-Implementierung des Beispiels aus dem Artikel Wie ein Agent entscheidet, was als Nächstes zu tun ist (Planning vs Reactive).
Wenn du den Artikel noch nicht gelesen hast, starte dort. Hier liegt der Fokus nur auf dem Code und dem Verhalten der beiden Strategien.
Was dieses Beispiel zeigt
- Wie dieselbe Aufgabe mit zwei Ansätzen ausgeführt wird: Planning und Reactive
- Wie ein Planning-Agent zuerst einen Plan erstellt und ihn bei Fehlern neu aufbaut
- Wie ein Reactive-Agent nach jedem Ergebnis die nächste Aktion auswählt
- Warum reactive meist robuster gegen Flakes ist, während planning einfacher zu kontrollieren ist
Projektstruktur
foundations/
└── planning-vs-reactive/
└── python/
├── main.py # startet beide Strategien und vergleicht Ergebnisse
├── llm.py # einfache Entscheidungsschicht: plan / replan / next action
├── planning_agent.py # Agent mit vorab erstelltem Plan
├── reactive_agent.py # Agent, der situativ handelt
├── tools.py # Tools + deterministischer Flake zum Lernen
└── requirements.txt
tools.py enthält absichtlich einen kontrollierten (deterministischen) Fehler, damit der Unterschied zwischen den Ansätzen bei jedem Run reproduzierbar ist.
Ausführen
1. Repository klonen und in den Ordner wechseln:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/planning-vs-reactive/python
2. Abhängigkeiten installieren (für dieses Beispiel gibt es keine externen Pakete):
pip install -r requirements.txt
3. Vergleich starten:
python main.py
Was wir im Code bauen
Wir bauen zwei Roboter und geben ihnen dieselbe Aufgabe.
- der erste Roboter erstellt zuerst einen Plan und folgt dann dem Plan
- der zweite Roboter entscheidet den nächsten Schritt direkt während der Ausführung
- wenn etwas kaputtgeht, schauen wir, wer sich schneller anpasst
Das ist wie zwei Fahrten: einer fährt nach fertiger Karte, der andere orientiert sich unterwegs.
Code
tools.py — Tools mit deterministischem Flake
from typing import Any
def make_initial_state(user_id: int) -> dict[str, Any]:
# Deterministic flake config for teaching: orders fails once, then succeeds.
return {
"user_id": user_id,
"_flaky": {
"orders_failures_left": 1,
"balance_failures_left": 0,
},
}
def fetch_profile(state: dict[str, Any]) -> dict[str, Any]:
user_id = state["user_id"]
return {
"profile": {
"user_id": user_id,
"name": "Anna",
"tier": "pro",
}
}
def fetch_orders(state: dict[str, Any]) -> dict[str, Any]:
flaky = state["_flaky"]
if flaky["orders_failures_left"] > 0:
flaky["orders_failures_left"] -= 1
return {"error": "orders_api_timeout"}
return {
"orders": [
{"id": "ord-1001", "total": 49.9, "status": "paid"},
{"id": "ord-1002", "total": 19.0, "status": "shipped"},
]
}
def fetch_balance(state: dict[str, Any]) -> dict[str, Any]:
flaky = state["_flaky"]
if flaky["balance_failures_left"] > 0:
flaky["balance_failures_left"] -= 1
return {"error": "billing_api_unavailable"}
return {"balance": {"currency": "USD", "value": 128.4}}
def build_summary(state: dict[str, Any]) -> dict[str, Any]:
profile = state.get("profile")
orders = state.get("orders")
balance = state.get("balance")
if not profile or not orders or not balance:
return {"error": "not_enough_data_for_summary"}
text = (
f"User {profile['name']} ({profile['tier']}) has "
f"{len(orders)} recent orders and balance {balance['value']} {balance['currency']}."
)
return {"summary": text}
llm.py — einfache lehrreiche Entscheidungsschicht
from typing import Any
DEFAULT_PLAN = ["fetch_profile", "fetch_orders", "fetch_balance", "build_summary"]
def create_plan(task: str) -> list[str]:
# Learning version: fixed starter plan keeps behavior easy to reason about.
_ = task
return DEFAULT_PLAN.copy()
def replan(task: str, state: dict[str, Any], failed_step: str, error: str) -> list[str]:
# Learning version: rebuild plan from missing data in state.
_ = task, failed_step, error
remaining: list[str] = []
if "profile" not in state:
remaining.append("fetch_profile")
if "orders" not in state:
remaining.append("fetch_orders")
if "balance" not in state:
remaining.append("fetch_balance")
if "summary" not in state:
remaining.append("build_summary")
return remaining
def choose_next_action(task: str, state: dict[str, Any]) -> str:
# Learning version: one-step-at-a-time policy driven by current state.
_ = task
if "profile" not in state:
return "fetch_profile"
# If orders just failed, fetch other missing data first.
if state.get("last_error") == "orders_api_timeout" and "balance" not in state:
return "fetch_balance"
if "orders" not in state:
return "fetch_orders"
if "balance" not in state:
return "fetch_balance"
return "build_summary"
llm.py ruft hier kein externes API auf. Das ist eine bewusste Vereinfachung fürs Lernen: so ist der Unterschied zwischen planning und reactive besser sichtbar.
planning_agent.py — erst planen, dann ausführen
from typing import Any
from llm import create_plan, replan
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile, make_initial_state
TOOLS = {
"fetch_profile": fetch_profile,
"fetch_orders": fetch_orders,
"fetch_balance": fetch_balance,
"build_summary": build_summary,
}
def run_planning_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
state = make_initial_state(user_id)
plan = create_plan(task)
trace: list[str] = [f"Initial plan: {plan}"]
step = 0
while plan and step < max_steps:
action = plan.pop(0)
step += 1
trace.append(f"[{step}] action={action}")
tool = TOOLS.get(action)
if not tool:
trace.append(f"unknown_action={action}")
state["last_error"] = f"unknown_action:{action}"
continue
result = tool(state)
trace.append(f"result={result}")
if "error" in result:
state["last_error"] = result["error"]
trace.append("planning: replan after failure")
plan = replan(task, state, failed_step=action, error=result["error"])
trace.append(f"new_plan={plan}")
continue
state.update(result)
state.pop("last_error", None)
if "summary" in state:
return {"mode": "planning", "done": True, "steps": step, "state": state, "trace": trace}
return {"mode": "planning", "done": False, "steps": step, "state": state, "trace": trace}
Der Planning-Agent trifft eine strategische Entscheidung im Voraus und wechselt nur bei Fehlern zum Replanning.
reactive_agent.py — Entscheidung bei jedem Schritt
from typing import Any
from llm import choose_next_action
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile, make_initial_state
TOOLS = {
"fetch_profile": fetch_profile,
"fetch_orders": fetch_orders,
"fetch_balance": fetch_balance,
"build_summary": build_summary,
}
def run_reactive_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
state = make_initial_state(user_id)
trace: list[str] = []
for step in range(1, max_steps + 1):
if "summary" in state:
return {"mode": "reactive", "done": True, "steps": step - 1, "state": state, "trace": trace}
action = choose_next_action(task, state)
trace.append(f"[{step}] action={action}")
tool = TOOLS.get(action)
if not tool:
trace.append(f"unknown_action={action}")
state["last_error"] = f"unknown_action:{action}"
continue
result = tool(state)
trace.append(f"result={result}")
if "error" in result:
state["last_error"] = result["error"]
continue
state.update(result)
state.pop("last_error", None)
return {"mode": "reactive", "done": False, "steps": max_steps, "state": state, "trace": trace}
Der Reactive-Agent hält nicht am initialen Plan fest. Er bewertet den Zustand nach jeder Aktion und wählt den nächsten Schritt aus dem aktuellen state.
main.py — Vergleich zweier Ansätze
from planning_agent import run_planning_agent
from reactive_agent import run_reactive_agent
TASK = "Prepare a short account summary for user_id=42 with profile, orders, and balance."
USER_ID = 42
def print_result(result: dict) -> None:
print(f"\n=== {result['mode'].upper()} ===")
print(f"done={result['done']} | steps={result['steps']}")
print("summary:", result["state"].get("summary"))
print("\ntrace:")
for line in result["trace"]:
print(" ", line)
def main() -> None:
planning = run_planning_agent(task=TASK, user_id=USER_ID)
reactive = run_reactive_agent(task=TASK, user_id=USER_ID)
print_result(planning)
print_result(reactive)
if __name__ == "__main__":
main()
requirements.txt
# No external dependencies for this learning example.
Beispielausgabe
=== PLANNING ===
done=True | steps=5
summary: User Anna (pro) has 2 recent orders and balance 128.4 USD.
trace:
Initial plan: ['fetch_profile', 'fetch_orders', 'fetch_balance', 'build_summary']
[1] action=fetch_profile
result={'profile': {'user_id': 42, 'name': 'Anna', 'tier': 'pro'}}
[2] action=fetch_orders
result={'error': 'orders_api_timeout'}
planning: replan after failure
new_plan=['fetch_orders', 'fetch_balance', 'build_summary']
[3] action=fetch_orders
result={'orders': [{'id': 'ord-1001', 'total': 49.9, 'status': 'paid'}, {'id': 'ord-1002', 'total': 19.0, 'status': 'shipped'}]}
[4] action=fetch_balance
result={'balance': {'currency': 'USD', 'value': 128.4}}
[5] action=build_summary
result={'summary': 'User Anna (pro) has 2 recent orders and balance 128.4 USD.'}
=== REACTIVE ===
done=True | steps=5
summary: User Anna (pro) has 2 recent orders and balance 128.4 USD.
trace:
[1] action=fetch_profile
result={'profile': {'user_id': 42, 'name': 'Anna', 'tier': 'pro'}}
[2] action=fetch_orders
result={'error': 'orders_api_timeout'}
[3] action=fetch_balance
result={'balance': {'currency': 'USD', 'value': 128.4}}
[4] action=fetch_orders
result={'orders': [{'id': 'ord-1001', 'total': 49.9, 'status': 'paid'}, {'id': 'ord-1002', 'total': 19.0, 'status': 'shipped'}]}
[5] action=build_summary
result={'summary': 'User Anna (pro) has 2 recent orders and balance 128.4 USD.'}
Hinweis: das Beispiel ist zum Lernen deterministisch gemacht.
Bei jedem Run fälltfetch_orderseinmal aus, daher ist der Trace stabil reproduzierbar.
Was man in der Praxis sieht
| Planning-Agent | Reactive-Agent | |
|---|---|---|
| Wann Schritte gewählt werden | Am Start (Plan) | Nach jeder Aktion |
| Reaktion auf Fehler | Baut den Plan neu auf | Wählt sofort einen neuen Schritt |
| Vorhersagbarkeit | Höher | Niedriger |
| Robustheit gegen Flakes | Mittel | Meist höher |
Was du in diesem Beispiel ändern kannst
- Ändere
orders_failures_leftinmake_initial_statevon1auf2und sieh, wie sich der Trace verändert - Füge ein separates Limit für die Anzahl von
replanbeim Planning-Agenten hinzu - Füge die Regel "nicht dieselbe Aktion 3-mal hintereinander wiederholen" für den Reactive-Agenten hinzu
- Setze
balance_failures_left = 1und schau, wer sich nach zwei unterschiedlichen Fehlern schneller erholt
Vollständiger Code auf GitHub
Im Repository liegt die vollständige Version dieser Demo: zwei Agentenstrategien, gemeinsame Tools und Schritt-Tracing.
Vollständigen Code auf GitHub ansehen ↗