Voici l implémentation pédagogique complète de l exemple de l article Comment un agent décide quoi faire ensuite (Planning vs Reactive).
Si tu n as pas encore lu l article, commence par lui. Ici le focus est uniquement sur le code et le comportement des deux stratégies.
Ce que cet exemple montre
- Comment une même tâche est exécutée avec deux approches: Planning et Reactive
- Comment l agent planning construit d abord un plan, puis le reconstruit en cas d échec
- Comment l agent reactive choisit l action suivante après chaque résultat
- Pourquoi reactive est généralement plus robuste aux flakes, tandis que planning est plus simple à contrôler
Structure du projet
foundations/
└── planning-vs-reactive/
└── python/
├── main.py # lance les deux stratégies et compare les résultats
├── llm.py # couche simple de décision: plan / replan / next action
├── planning_agent.py # agent avec plan préétabli
├── reactive_agent.py # agent qui agit selon la situation
├── tools.py # outils + flake déterministe pour l apprentissage
└── requirements.txt
tools.py contient volontairement une panne contrôlée (déterministe) afin que la différence entre les approches soit reproductible à chaque run.
Lancer le projet
1. Clone le repository et va dans le dossier:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/planning-vs-reactive/python
2. Installe les dépendances (cet exemple n a pas de packages externes):
pip install -r requirements.txt
3. Lance la comparaison:
python main.py
Ce que nous construisons dans le code
Nous créons deux robots et leur donnons la même tâche.
- le premier robot fait d abord un plan, puis suit ce plan
- le deuxième robot décide l étape suivante pendant l exécution
- si quelque chose casse, on observe qui s adapte le plus vite
C est comme deux trajets: l un suit une carte prête, l autre s oriente en route.
Code
tools.py — outils avec flake déterministe
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 — couche simple de décision pédagogique
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 n appelle pas d API externe ici. C est une simplification volontaire pour l apprentissage: on voit plus facilement la différence entre planning et reactive.
planning_agent.py — plan d abord, exécution ensuite
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}
L agent planning prend une décision stratégique à l avance, et ne passe au replanning qu en cas d échec.
reactive_agent.py — décision à chaque étape
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}
L agent reactive ne s accroche pas au plan initial. Il évalue l état après chaque action et choisit l étape suivante selon le state courant.
main.py — comparaison de deux approches
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.
Exemple de sortie
=== 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.'}
Note: l exemple est rendu déterministe pour l apprentissage.
À chaque run,fetch_orderséchoue une fois, donc le trace est reproduit de façon stable.
Ce que l on voit en pratique
| Agent Planning | Agent Reactive | |
|---|---|---|
| Quand il choisit les étapes | Au démarrage (plan) | Après chaque action |
| Réaction à l échec | Reconstruit le plan | Choisit immédiatement une nouvelle étape |
| Prévisibilité | Plus élevée | Plus faible |
| Robustesse aux flakes | Moyenne | Généralement plus élevée |
Ce qu il faut changer dans cet exemple
- Change
orders_failures_leftdansmake_initial_statede1à2et regarde comment le trace change - Ajoute une limite séparée sur le nombre de
replanpour l agent planning - Ajoute la règle "ne pas répéter la même action 3 fois d affilée" pour l agent reactive
- Mets
balance_failures_left = 1et observe qui récupère plus vite après deux échecs différents
Code complet sur GitHub
Le repository contient la version complète de cette démo: deux stratégies d agent, outils partagés et traçage des étapes.
Voir le code complet sur GitHub ↗