Коли AI-агент має зупинитися у Python: runtime-ліміти (повний приклад)

foundations/
└── stop-conditions/
    └── python/
        ├── main.py            # запускає сценарії і друкує підсумок
        ├── agent.py           # цикл агента + перевірка stop conditions
        ├── llm.py             # простий шар рішень: next action
        ├── tools.py           # навчальні tools і контрольовані збої
        └── requirements.txt

git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/stop-conditions/python

pip install -r requirements.txt

python main.py

from typing import Any


def make_initial_state(user_id: int, fail_fetch_times: int) -> dict[str, Any]:
    return {
        "user_id": user_id,
        "fetch_calls": 0,
        "fail_fetch_times": fail_fetch_times,
    }


def fetch_orders(state: dict[str, Any]) -> dict[str, Any]:
    state["fetch_calls"] += 1

    if state["fetch_calls"] <= state["fail_fetch_times"]:
        return {"ok": False, "error": "orders_api_timeout"}

    orders = [
        {"id": "ord-2001", "total": 49.9, "status": "paid"},
        {"id": "ord-2002", "total": 19.0, "status": "shipped"},
    ]
    state["orders"] = orders
    return {"ok": True, "orders": orders}


def build_summary(state: dict[str, Any]) -> dict[str, Any]:
    orders = state.get("orders")
    if not orders:
        return {"ok": False, "error": "missing_orders"}

    summary = f"Prepared report for {len(orders)} recent orders."
    state["summary"] = summary
    return {"ok": True, "summary": summary}

from typing import Any


def choose_next_action(task: str, state: dict[str, Any]) -> dict[str, Any]:
    # Learning version: fixed policy keeps behavior easy to reason about.
    _ = task

    if "orders" not in state:
        return {"action": "fetch_orders", "parameters": {}}
    return {"action": "build_summary", "parameters": {}}

from dataclasses import dataclass
from typing import Any

from llm import choose_next_action
from tools import build_summary, fetch_orders, make_initial_state

TOOLS = {
    "fetch_orders": fetch_orders,
    "build_summary": build_summary,
}


@dataclass
class StopPolicy:
    max_steps: int
    max_errors: int
    max_no_progress: int


def evaluate_stop_conditions(
    state: dict[str, Any],
    steps: int,
    errors: int,
    no_progress: int,
    policy: StopPolicy,
) -> str | None:
    if "summary" in state:
        return "goal_reached"
    if steps >= policy.max_steps:
        return "step_limit"
    if errors >= policy.max_errors:
        return "too_many_errors"
    if no_progress >= policy.max_no_progress:
        return "no_progress"
    return None


def run_agent(task: str, user_id: int, fail_fetch_times: int, policy: StopPolicy) -> dict[str, Any]:
    state = make_initial_state(user_id=user_id, fail_fetch_times=fail_fetch_times)
    history: list[dict[str, Any]] = []

    steps = 0
    errors = 0
    no_progress = 0
    stop_reason: str | None = None

    while True:
        stop_reason = evaluate_stop_conditions(
            state=state,
            steps=steps,
            errors=errors,
            no_progress=no_progress,
            policy=policy,
        )
        if stop_reason is not None:
            break

        steps += 1

        call = choose_next_action(task, state)
        action = call["action"]
        history.append({"step": steps, "action": action, "status": "requested"})

        tool = TOOLS.get(action)
        if not tool:
            errors += 1
            no_progress += 1
            state["last_error"] = f"unknown_action:{action}"
            history.append({"step": steps, "action": action, "status": "error"})
        else:
            before_keys = set(state.keys())
            result = tool(state)

            if result.get("ok"):
                after_keys = set(state.keys())
                progress = len(after_keys - before_keys) > 0
                no_progress = 0 if progress else no_progress + 1
                state.pop("last_error", None)
                history.append({"step": steps, "action": action, "status": "ok"})
            else:
                errors += 1
                no_progress += 1
                state["last_error"] = result.get("error", "unknown_error")
                history.append({"step": steps, "action": action, "status": "error"})

    return {
        "done": stop_reason == "goal_reached",
        "stop_reason": stop_reason,
        "steps": steps,
        "errors": errors,
        "no_progress": no_progress,
        "summary": state.get("summary"),
        "history": history,
    }

import json

from agent import StopPolicy, run_agent

TASK = "Build weekly orders summary"
POLICY = StopPolicy(max_steps=6, max_errors=2, max_no_progress=3)
STEP_LIMIT_POLICY = StopPolicy(max_steps=1, max_errors=2, max_no_progress=3)


def compact_result(result: dict) -> str:
    return (
        "{"
        f"\"done\": {str(bool(result.get('done'))).lower()}, "
        f"\"stop_reason\": {json.dumps(result.get('stop_reason'), ensure_ascii=False)}, "
        f"\"steps\": {int(result.get('steps', 0))}, "
        f"\"errors\": {int(result.get('errors', 0))}, "
        f"\"no_progress\": {int(result.get('no_progress', 0))}, "
        f"\"summary\": {json.dumps(result.get('summary'), ensure_ascii=False)}, "
        "\"history\": [{...}]"
        "}"
    )


def print_policy(policy: StopPolicy) -> None:
    print(
        "Policy:",
        json.dumps(
            {
                "max_steps": policy.max_steps,
                "max_errors": policy.max_errors,
                "max_no_progress": policy.max_no_progress,
            },
            ensure_ascii=False,
        ),
    )


def main() -> None:
    print("=== SCENARIO 1: GOAL REACHED ===")
    print_policy(POLICY)
    result_ok = run_agent(
        task=TASK,
        user_id=42,
        fail_fetch_times=1,
        policy=POLICY,
    )
    print("Run result:", compact_result(result_ok))

    print("\n=== SCENARIO 2: STOPPED BY ERROR LIMIT ===")
    print_policy(POLICY)
    result_stopped = run_agent(
        task=TASK,
        user_id=42,
        fail_fetch_times=10,
        policy=POLICY,
    )
    print("Run result:", compact_result(result_stopped))

    print("\n=== SCENARIO 3: STOPPED BY STEP LIMIT ===")
    print_policy(STEP_LIMIT_POLICY)
    result_step_limit = run_agent(
        task=TASK,
        user_id=42,
        fail_fetch_times=0,
        policy=STEP_LIMIT_POLICY,
    )
    print("Run result:", compact_result(result_step_limit))


if __name__ == "__main__":
    main()

# No external dependencies for this learning example.

python main.py

=== SCENARIO 1: GOAL REACHED ===
Policy: {"max_steps": 6, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": true, "stop_reason": "goal_reached", "steps": 3, "errors": 1, "no_progress": 0, "summary": "Prepared report for 2 recent orders.", "history": [{...}]}

=== SCENARIO 2: STOPPED BY ERROR LIMIT ===
Policy: {"max_steps": 6, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": false, "stop_reason": "too_many_errors", "steps": 2, "errors": 2, "no_progress": 2, "summary": null, "history": [{...}]}

=== SCENARIO 3: STOPPED BY STEP LIMIT ===
Policy: {"max_steps": 1, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": false, "stop_reason": "step_limit", "steps": 1, "errors": 0, "no_progress": 0, "summary": null, "history": [{...}]}