Skip to content

Commit 0d8271d

Browse files
author
AgentPatterns
committed
feat(examples/python): add planning-vs-reactive runnable example
1 parent 12b32f1 commit 0d8271d

7 files changed

Lines changed: 277 additions & 0 deletions

File tree

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
# Planning vs Reactive - Python Implementation
2+
3+
Runnable learning example that compares two agent strategies on the same task:
4+
planning-first execution vs reactive step-by-step decisions.
5+
6+
---
7+
8+
## Quick start
9+
10+
```bash
11+
# (optional) create venv
12+
python -m venv .venv && source .venv/bin/activate
13+
14+
# install dependencies (none required, command kept for consistency)
15+
pip install -r requirements.txt
16+
17+
# run the comparison
18+
python main.py
19+
```
20+
21+
## Full walkthrough
22+
23+
Read the concept article:
24+
https://agentpatterns.tech/en/foundations/planning-vs-reactive
25+
26+
## What's inside
27+
28+
- Deterministic flaky tools (orders fails once, then succeeds)
29+
- Planning agent with explicit `create_plan -> execute -> replan`
30+
- Reactive agent that chooses the next action from current state
31+
- Side-by-side trace output for easy comparison
32+
33+
## Project layout
34+
35+
```text
36+
examples/
37+
foundations/
38+
planning-vs-reactive/
39+
python/
40+
README.md
41+
main.py
42+
llm.py
43+
planning_agent.py
44+
reactive_agent.py
45+
tools.py
46+
requirements.txt
47+
```
48+
49+
## Notes
50+
51+
- This code is intentionally simple for learning.
52+
- The website provides multilingual explanation and context.
53+
54+
## License
55+
56+
MIT
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
from typing import Any
2+
3+
4+
DEFAULT_PLAN = ["fetch_profile", "fetch_orders", "fetch_balance", "build_summary"]
5+
6+
7+
def create_plan(task: str) -> list[str]:
8+
# Learning version: fixed starter plan keeps behavior easy to reason about.
9+
_ = task
10+
return DEFAULT_PLAN.copy()
11+
12+
13+
def replan(task: str, state: dict[str, Any], failed_step: str, error: str) -> list[str]:
14+
# Learning version: rebuild plan from missing data in state.
15+
_ = task, failed_step, error
16+
remaining: list[str] = []
17+
18+
if "profile" not in state:
19+
remaining.append("fetch_profile")
20+
if "orders" not in state:
21+
remaining.append("fetch_orders")
22+
if "balance" not in state:
23+
remaining.append("fetch_balance")
24+
if "summary" not in state:
25+
remaining.append("build_summary")
26+
27+
return remaining
28+
29+
30+
def choose_next_action(task: str, state: dict[str, Any]) -> str:
31+
# Learning version: one-step-at-a-time policy driven by current state.
32+
_ = task
33+
34+
if "profile" not in state:
35+
return "fetch_profile"
36+
37+
# If orders just failed, fetch other missing data first.
38+
if state.get("last_error") == "orders_api_timeout" and "balance" not in state:
39+
return "fetch_balance"
40+
41+
if "orders" not in state:
42+
return "fetch_orders"
43+
if "balance" not in state:
44+
return "fetch_balance"
45+
return "build_summary"
Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
from planning_agent import run_planning_agent
2+
from reactive_agent import run_reactive_agent
3+
4+
TASK = "Prepare a short account summary for user_id=42 with profile, orders, and balance."
5+
USER_ID = 42
6+
7+
8+
def print_result(result: dict) -> None:
9+
print(f"\n=== {result['mode'].upper()} ===")
10+
print(f"done={result['done']} | steps={result['steps']}")
11+
print("summary:", result["state"].get("summary"))
12+
print("\ntrace:")
13+
for line in result["trace"]:
14+
print(" ", line)
15+
16+
17+
def main() -> None:
18+
planning = run_planning_agent(task=TASK, user_id=USER_ID)
19+
reactive = run_reactive_agent(task=TASK, user_id=USER_ID)
20+
21+
print_result(planning)
22+
print_result(reactive)
23+
24+
25+
if __name__ == "__main__":
26+
main()
Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
from typing import Any
2+
3+
from llm import create_plan, replan
4+
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile, make_initial_state
5+
6+
TOOLS = {
7+
"fetch_profile": fetch_profile,
8+
"fetch_orders": fetch_orders,
9+
"fetch_balance": fetch_balance,
10+
"build_summary": build_summary,
11+
}
12+
13+
14+
def run_planning_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
15+
state = make_initial_state(user_id)
16+
plan = create_plan(task)
17+
trace: list[str] = [f"Initial plan: {plan}"]
18+
19+
step = 0
20+
while plan and step < max_steps:
21+
action = plan.pop(0)
22+
step += 1
23+
trace.append(f"[{step}] action={action}")
24+
25+
tool = TOOLS.get(action)
26+
if not tool:
27+
trace.append(f"unknown_action={action}")
28+
state["last_error"] = f"unknown_action:{action}"
29+
continue
30+
31+
result = tool(state)
32+
trace.append(f"result={result}")
33+
34+
if "error" in result:
35+
state["last_error"] = result["error"]
36+
trace.append("planning: replan after failure")
37+
plan = replan(task, state, failed_step=action, error=result["error"])
38+
trace.append(f"new_plan={plan}")
39+
continue
40+
41+
state.update(result)
42+
state.pop("last_error", None)
43+
44+
if "summary" in state:
45+
return {"mode": "planning", "done": True, "steps": step, "state": state, "trace": trace}
46+
47+
return {"mode": "planning", "done": False, "steps": step, "state": state, "trace": trace}
Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
from typing import Any
2+
3+
from llm import choose_next_action
4+
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile, make_initial_state
5+
6+
TOOLS = {
7+
"fetch_profile": fetch_profile,
8+
"fetch_orders": fetch_orders,
9+
"fetch_balance": fetch_balance,
10+
"build_summary": build_summary,
11+
}
12+
13+
14+
def run_reactive_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
15+
state = make_initial_state(user_id)
16+
trace: list[str] = []
17+
18+
for step in range(1, max_steps + 1):
19+
if "summary" in state:
20+
return {"mode": "reactive", "done": True, "steps": step - 1, "state": state, "trace": trace}
21+
22+
action = choose_next_action(task, state)
23+
trace.append(f"[{step}] action={action}")
24+
25+
tool = TOOLS.get(action)
26+
if not tool:
27+
trace.append(f"unknown_action={action}")
28+
state["last_error"] = f"unknown_action:{action}"
29+
continue
30+
31+
result = tool(state)
32+
trace.append(f"result={result}")
33+
34+
if "error" in result:
35+
state["last_error"] = result["error"]
36+
continue
37+
38+
state.update(result)
39+
state.pop("last_error", None)
40+
41+
return {"mode": "reactive", "done": False, "steps": max_steps, "state": state, "trace": trace}
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
# No external dependencies for this learning example.
Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
from typing import Any
2+
3+
4+
def make_initial_state(user_id: int) -> dict[str, Any]:
5+
# Deterministic flake config for teaching: orders fails once, then succeeds.
6+
return {
7+
"user_id": user_id,
8+
"_flaky": {
9+
"orders_failures_left": 1,
10+
"balance_failures_left": 0,
11+
},
12+
}
13+
14+
15+
def fetch_profile(state: dict[str, Any]) -> dict[str, Any]:
16+
user_id = state["user_id"]
17+
return {
18+
"profile": {
19+
"user_id": user_id,
20+
"name": "Anna",
21+
"tier": "pro",
22+
}
23+
}
24+
25+
26+
def fetch_orders(state: dict[str, Any]) -> dict[str, Any]:
27+
flaky = state["_flaky"]
28+
if flaky["orders_failures_left"] > 0:
29+
flaky["orders_failures_left"] -= 1
30+
return {"error": "orders_api_timeout"}
31+
32+
return {
33+
"orders": [
34+
{"id": "ord-1001", "total": 49.9, "status": "paid"},
35+
{"id": "ord-1002", "total": 19.0, "status": "shipped"},
36+
]
37+
}
38+
39+
40+
def fetch_balance(state: dict[str, Any]) -> dict[str, Any]:
41+
flaky = state["_flaky"]
42+
if flaky["balance_failures_left"] > 0:
43+
flaky["balance_failures_left"] -= 1
44+
return {"error": "billing_api_unavailable"}
45+
46+
return {"balance": {"currency": "USD", "value": 128.4}}
47+
48+
49+
def build_summary(state: dict[str, Any]) -> dict[str, Any]:
50+
profile = state.get("profile")
51+
orders = state.get("orders")
52+
balance = state.get("balance")
53+
54+
if not profile or not orders or not balance:
55+
return {"error": "not_enough_data_for_summary"}
56+
57+
text = (
58+
f"User {profile['name']} ({profile['tier']}) has "
59+
f"{len(orders)} recent orders and balance {balance['value']} {balance['currency']}."
60+
)
61+
return {"summary": text}

0 commit comments

Comments
 (0)