feat(examples/python): add stop-conditions runnable example

Author: AgentPatterns

examples/foundations/stop-conditions/python/agent.py

@@ -0,0 +1,94 @@
+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,
+    }

examples/foundations/stop-conditions/python/llm.py

@@ -0,0 +1,10 @@
+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": {}}

examples/foundations/stop-conditions/python/main.py

@@ -0,0 +1,71 @@
+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()

examples/foundations/stop-conditions/python/requirements.txt

@@ -0,0 +1 @@
+# No external dependencies for this learning example.

examples/foundations/stop-conditions/python/tools.py

@@ -0,0 +1,33 @@
+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}