Як агент вирішує, що робити далі (Planning vs Reactive) — Python (повна реалізація)

examples/
└── foundations/
    └── planning-vs-reactive/
        └── python/
            ├── main.py             # запускає обидві стратегії і порівнює результат
            ├── llm.py              # виклики моделі: plan / replan / next action
            ├── planning_agent.py   # агент із попереднім планом
            ├── reactive_agent.py   # агент, що діє по ситуації
            ├── tools.py            # нестабільні зовнішні джерела (імітація)
            └── requirements.txt

git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd examples/foundations/planning-vs-reactive/python

pip install -r requirements.txt

export OPENAI_API_KEY="sk-..."

python main.py

import random
from typing import Any


def fetch_profile(user_id: int) -> dict[str, Any]:
    # Стабільне джерело
    return {
        "profile": {
            "user_id": user_id,
            "name": "Anna",
            "tier": "pro",
        }
    }


def fetch_orders(user_id: int) -> dict[str, Any]:
    # Імітуємо флейк: іноді API "падає"
    if random.random() < 0.35:
        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(user_id: int) -> dict[str, Any]:
    if random.random() < 0.2:
        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}

import json
import os
from typing import Any

from openai import OpenAI

MODEL = "gpt-4.1-mini"
ALLOWED_ACTIONS = {"fetch_profile", "fetch_orders", "fetch_balance", "build_summary", "finish"}
api_key = os.environ.get("OPENAI_API_KEY")

if not api_key:
    raise EnvironmentError(
        "OPENAI_API_KEY is not set.\n"
        "Run: export OPENAI_API_KEY='sk-...'"
    )

client = OpenAI(api_key=api_key)


def _ask_json(prompt: str, fallback: dict[str, Any]) -> dict[str, Any]:
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
    )
    text = (resp.choices[0].message.content or "").strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return fallback


def create_plan(task: str) -> list[str]:
    prompt = f"""
Return strict JSON: {{"plan": ["step_name", "..."]}}
Available steps: fetch_profile, fetch_orders, fetch_balance, build_summary.
Task: {task}
""".strip()
    data = _ask_json(prompt, {"plan": ["fetch_profile", "fetch_orders", "fetch_balance", "build_summary"]})
    plan = data.get("plan") or []
    return [str(step) for step in plan]


def replan(task: str, state: dict[str, Any], failed_step: str, error: str) -> list[str]:
    prompt = f"""
Return strict JSON: {{"plan": ["step_name", "..."]}}
Available steps: fetch_profile, fetch_orders, fetch_balance, build_summary.
Task: {task}
Failed step: {failed_step}
Error: {error}
Current state keys: {list(state.keys())}
Create only remaining steps.
""".strip()
    data = _ask_json(prompt, {"plan": ["fetch_orders", "fetch_balance", "build_summary"]})
    plan = data.get("plan") or []
    return [str(step) for step in plan]


def choose_next_action(task: str, state: dict[str, Any]) -> str:
    prompt = f"""
Return strict JSON: {{"action":"step_name"}}
Available actions: fetch_profile, fetch_orders, fetch_balance, build_summary, finish.
Task: {task}
Known state keys: {list(state.keys())}
If summary exists, choose finish.
If data missing, choose one action to progress.
If previous action failed, pick a different action.
""".strip()
    data = _ask_json(prompt, {"action": "fetch_profile"})
    action = str(data.get("action") or "fetch_profile")
    return action if action in ALLOWED_ACTIONS else "fetch_profile"

from typing import Any

from llm import create_plan, replan
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile

TOOLS = {
    "fetch_profile": lambda state: fetch_profile(state["user_id"]),
    "fetch_orders": lambda state: fetch_orders(state["user_id"]),
    "fetch_balance": lambda state: fetch_balance(state["user_id"]),
    "build_summary": build_summary,
}


def run_planning_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
    state: dict[str, Any] = {"user_id": 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}")
            continue

        result = tool(state)
        trace.append(f"result={result}")

        if "error" in result:
            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)
        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}

from typing import Any

from llm import choose_next_action
from tools import build_summary, fetch_balance, fetch_orders, fetch_profile

TOOLS = {
    "fetch_profile": lambda state: fetch_profile(state["user_id"]),
    "fetch_orders": lambda state: fetch_orders(state["user_id"]),
    "fetch_balance": lambda state: fetch_balance(state["user_id"]),
    "build_summary": build_summary,
}


def run_reactive_agent(task: str, user_id: int, max_steps: int = 8) -> dict[str, Any]:
    state: dict[str, Any] = {"user_id": user_id}
    trace: list[str] = []

    for step in range(1, max_steps + 1):
        action = choose_next_action(task, state)
        trace.append(f"[{step}] action={action}")

        if action == "finish":
            if "summary" in state:
                return {"mode": "reactive", "done": True, "steps": step, "state": state, "trace": trace}
            trace.append("finish_rejected: summary is missing")
            continue

        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}

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

openai>=1.0.0

=== PLANNING ===
done=True | steps=6
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']
  ...

=== 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={'orders': [{'id': 'ord-1001', 'total': 49.9, 'status': 'paid'}, {'id': 'ord-1002', 'total': 19.0, 'status': 'shipped'}]}
  [3] action=fetch_balance
  result={'balance': {'currency': 'USD', 'value': 128.4}}
  [4] action=build_summary
  result={'summary': 'User Anna (pro) has 2 recent orders and balance 128.4 USD.'}
  [5] action=finish