Це повна навчальна реалізація прикладу зі статті Як агент вирішує, що робити далі (Planning vs Reactive).
Якщо ти ще не читав статтю, почни з неї. Тут фокус лише на коді і поведінці двох стратегій.
Що цей приклад демонструє
- Як одна й та сама задача виконується двома підходами: Planning і Reactive
- Як planning-агент спочатку будує план, а при збої перебудовує його
- Як reactive-агент обирає наступну дію після кожного результату
- Чому reactive зазвичай стійкіший до флейків, а planning простіший для контролю
Структура проєкту
foundations/
└── planning-vs-reactive/
└── python/
├── main.py # запускає обидві стратегії і порівнює результат
├── llm.py # простий шар рішень: plan / replan / next action
├── planning_agent.py # агент із попереднім планом
├── reactive_agent.py # агент, що діє по ситуації
├── tools.py # інструменти + детермінований флейк для навчання
└── requirements.txt
tools.py навмисно містить контрольований (детермінований) збій, щоб різниця між підходами була відтворюваною на кожному запуску.
Як запустити
1. Клонуй репозиторій і перейди в папку:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/planning-vs-reactive/python
2. Встанови залежності (для цього прикладу зовнішніх пакетів немає):
pip install -r requirements.txt
3. Запусти порівняння:
python main.py
Що ми будуємо в коді
Ми робимо двох роботів і даємо їм одну й ту саму задачу.
- перший робот спочатку складає план, а потім іде по плану
- другий робот вирішує наступний крок прямо під час роботи
- якщо щось ламається, ми дивимось, хто швидше підлаштовується
Це як дві поїздки: один їде за готовою картою, інший орієнтується по дорозі.
Код
tools.py — інструменти з детермінованим флейком
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 — простий навчальний шар рішень
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 тут не викликає зовнішнє API. Це свідоме спрощення для навчання: легше бачити різницю між planning і reactive.
planning_agent.py — спочатку план, потім виконання
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}
Planning-агент приймає стратегічне рішення наперед, і лише при збоях переходить до перебудови плану.
reactive_agent.py — рішення на кожному кроці
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}
Reactive-агент не тримається за початковий план. Він оцінює стан після кожної дії і обирає наступний крок за поточним state.
main.py — порівняння двох підходів
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.
Приклад виводу
=== 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.'}
Примітка: приклад зроблено детермінованим для навчання.
На кожному запускуfetch_ordersпадає один раз, тому trace стабільно відтворюється.
Що видно на практиці
| Planning агент | Reactive агент | |
|---|---|---|
| Коли обирає кроки | На старті (план) | Після кожної дії |
| Реакція на збій | Перебудовує план | Одразу обирає новий крок |
| Передбачуваність | Вища | Нижча |
| Стійкість до флейків | Середня | Зазвичай вища |
Що змінити в цьому прикладі
- Зміни
orders_failures_leftуmake_initial_stateз1на2і подивись, як зміниться trace - Додай окремий ліміт на кількість
replanу planning-агента - Додай правило "не повторювати одну й ту саму дію 3 рази підряд" для reactive-агента
- Зроби
balance_failures_left = 1і подивись, хто швидше відновлюється після двох різних збоїв
Повний код на GitHub
У репозиторії лежить повна версія цього демо: дві стратегії агента, спільні інструменти і трасування кроків.
Переглянути повний код на GitHub ↗