Agent-Planung vs Reaktive Ausführung in Python: Vollständiges Beispiel

Lehrreiches runnable Beispiel, in dem eine Aufgabe mit zwei Agentenstrategien gelöst wird: planning und reactive.
Auf dieser Seite
  1. Was dieses Beispiel zeigt
  2. Projektstruktur
  3. Ausführen
  4. Was wir im Code bauen
  5. Code
  6. tools.py — Tools mit deterministischem Flake
  7. llm.py — einfache lehrreiche Entscheidungsschicht
  8. planning_agent.py — erst planen, dann ausführen
  9. reactive_agent.py — Entscheidung bei jedem Schritt
  10. main.py — Vergleich zweier Ansätze
  11. requirements.txt
  12. Beispielausgabe
  13. Was man in der Praxis sieht
  14. Was du in diesem Beispiel ändern kannst
  15. Vollständiger Code auf GitHub

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

TEXT
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:

BASH
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):

BASH
pip install -r requirements.txt

3. Vergleich starten:

BASH
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

PYTHON
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

PYTHON
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

PYTHON
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

PYTHON
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

PYTHON
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

TEXT
# No external dependencies for this learning example.

Beispielausgabe

TEXT
=== 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ällt fetch_orders einmal aus, daher ist der Trace stabil reproduzierbar.


Was man in der Praxis sieht

Planning-AgentReactive-Agent
Wann Schritte gewählt werdenAm Start (Plan)Nach jeder Aktion
Reaktion auf FehlerBaut den Plan neu aufWählt sofort einen neuen Schritt
VorhersagbarkeitHöherNiedriger
Robustheit gegen FlakesMittelMeist höher

Was du in diesem Beispiel ändern kannst

  • Ändere orders_failures_left in make_initial_state von 1 auf 2 und sieh, wie sich der Trace verändert
  • Füge ein separates Limit für die Anzahl von replan beim Planning-Agenten hinzu
  • Füge die Regel "nicht dieselbe Aktion 3-mal hintereinander wiederholen" für den Reactive-Agenten hinzu
  • Setze balance_failures_left = 1 und 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 ↗
⏱️ 7 Min. LesezeitAktualisiert 3. März 2026Schwierigkeit: ★☆☆
Integriert: Production ControlOnceOnly
Guardrails für Tool-Calling-Agents
Shippe dieses Pattern mit Governance:
  • Budgets (Steps / Spend Caps)
  • Tool-Permissions (Allowlist / Blocklist)
  • Kill switch & Incident Stop
  • Idempotenz & Dedupe
  • Audit logs & Nachvollziehbarkeit
Integrierter Hinweis: OnceOnly ist eine Control-Layer für Production-Agent-Systeme.

Autor

Nick — Engineer, der Infrastruktur für KI-Agenten in Produktion aufbaut.

Fokus: Agent-Patterns, Failure-Modes, Runtime-Steuerung und Systemzuverlässigkeit.

🔗 GitHub: https://github.com/mykolademyanov


Redaktioneller Hinweis

Diese Dokumentation ist KI-gestützt, mit menschlicher redaktioneller Verantwortung für Genauigkeit, Klarheit und Produktionsrelevanz.

Der Inhalt basiert auf realen Ausfällen, Post-Mortems und operativen Vorfällen in produktiv eingesetzten KI-Agenten-Systemen.