feat(examples/python): add runnable example

Author: AgentPatterns

examples/task-decomposition-agent/python/README.md

@@ -0,0 +1,16 @@
+📖 Based on article:
+
+EN:
+https://agentpatterns.tech/en/agent-patterns/task-decomposition-agent
+
+DE:
+https://agentpatterns.tech/de/agent-patterns/task-decomposition-agent
+
+FR:
+https://agentpatterns.tech/fr/agent-patterns/task-decomposition-agent
+
+ES:
+https://agentpatterns.tech/es/agent-patterns/task-decomposition-agent
+
+UK:
+https://agentpatterns.tech/uk/agent-patterns/task-decomposition-agent

examples/task-decomposition-agent/python/gateway.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any, Callable
+
+
+class StopRun(Exception):
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
+@dataclass(frozen=True)
+class Budget:
+    max_plan_steps: int = 6
+    max_execute_steps: int = 8
+    max_tool_calls: int = 8
+    max_seconds: int = 60
+
+
+def _stable_json(value: Any) -> str:
+    if value is None or isinstance(value, (bool, int, float, str)):
+        return json.dumps(value, ensure_ascii=True, sort_keys=True)
+    if isinstance(value, list):
+        return "[" + ",".join(_stable_json(item) for item in value) + "]"
+    if isinstance(value, dict):
+        parts = []
+        for key in sorted(value):
+            parts.append(
+                json.dumps(str(key), ensure_ascii=True) + ":" + _stable_json(value[key])
+            )
+        return "{" + ",".join(parts) + "}"
+    return json.dumps(str(value), ensure_ascii=True)
+
+
+def args_hash(args: dict[str, Any]) -> str:
+    raw = _stable_json(args or {})
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]
+
+
+def validate_plan_action(
+    action: Any, *, max_plan_steps: int, allowed_tools: set[str]
+) -> list[dict[str, Any]]:
+    if not isinstance(action, dict):
+        raise StopRun("invalid_plan:not_object")
+
+    kind = action.get("kind")
+    if kind == "invalid":
+        raise StopRun("invalid_plan:non_json")
+    if kind != "plan":
+        raise StopRun("invalid_plan:bad_kind")
+
+    allowed_top_keys = {"kind", "steps"}
+    if set(action.keys()) - allowed_top_keys:
+        raise StopRun("invalid_plan:extra_keys")
+
+    steps = action.get("steps")
+    if not isinstance(steps, list) or not steps:
+        raise StopRun("invalid_plan:missing_steps")
+    if len(steps) < 3:
+        raise StopRun("invalid_plan:min_steps")
+    if len(steps) > max_plan_steps:
+        raise StopRun("invalid_plan:max_steps")
+
+    normalized: list[dict[str, Any]] = []
+    seen_ids: set[str] = set()
+
+    for index, step in enumerate(steps, start=1):
+        if not isinstance(step, dict):
+            raise StopRun(f"invalid_plan:step_{index}_not_object")
+
+        allowed_step_keys = {"id", "title", "tool", "args"}
+        if set(step.keys()) - allowed_step_keys:
+            raise StopRun(f"invalid_plan:step_{index}_extra_keys")
+
+        step_id = step.get("id")
+        if not isinstance(step_id, str) or not step_id.strip():
+            raise StopRun(f"invalid_plan:step_{index}_missing_id")
+        if step_id in seen_ids:
+            raise StopRun("invalid_plan:duplicate_step_id")
+        seen_ids.add(step_id)
+
+        title = step.get("title")
+        if not isinstance(title, str) or not title.strip():
+            raise StopRun(f"invalid_plan:step_{index}_missing_title")
+
+        tool = step.get("tool")
+        if not isinstance(tool, str) or not tool.strip():
+            raise StopRun(f"invalid_plan:step_{index}_missing_tool")
+        tool = tool.strip()
+        if tool not in allowed_tools:
+            raise StopRun(f"invalid_plan:tool_not_allowed:{tool}")
+
+        args = step.get("args", {})
+        if args is None:
+            args = {}
+        if not isinstance(args, dict):
+            raise StopRun(f"invalid_plan:step_{index}_bad_args")
+
+        normalized.append(
+            {
+                "id": step_id.strip(),
+                "title": title.strip(),
+                "tool": tool,
+                "args": args,
+            }
+        )
+
+    return normalized
+
+
+class ToolGateway:
+    def __init__(
+        self,
+        *,
+        allow: set[str],
+        registry: dict[str, Callable[..., dict[str, Any]]],
+        budget: Budget,
+    ):
+        self.allow = set(allow)
+        self.registry = registry
+        self.budget = budget
+        self.tool_calls = 0
+        self.seen_calls: set[str] = set()
+
+    def call(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
+        self.tool_calls += 1
+        if self.tool_calls > self.budget.max_tool_calls:
+            raise StopRun("max_tool_calls")
+
+        if name not in self.allow:
+            raise StopRun(f"tool_denied:{name}")
+
+        tool = self.registry.get(name)
+        if tool is None:
+            raise StopRun(f"tool_missing:{name}")
+
+        signature = f"{name}:{args_hash(args)}"
+        if signature in self.seen_calls:
+            raise StopRun("loop_detected")
+        self.seen_calls.add(signature)
+
+        try:
+            return tool(**args)
+        except TypeError as exc:
+            raise StopRun(f"tool_bad_args:{name}") from exc
+        except Exception as exc:
+            raise StopRun(f"tool_error:{name}") from exc

examples/task-decomposition-agent/python/llm.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from openai import APIConnectionError, APITimeoutError, OpenAI
+
+MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")
+LLM_TIMEOUT_SECONDS = float(os.getenv("OPENAI_TIMEOUT_SECONDS", "60"))
+
+
+class LLMTimeout(Exception):
+    pass
+
+
+class LLMEmpty(Exception):
+    pass
+
+
+PLAN_SYSTEM_PROMPT = """
+You are a task decomposition planner.
+Return only one JSON object in this exact shape:
+{
+  "kind": "plan",
+  "steps": [
+    {"id": "step_1", "title": "...", "tool": "...", "args": {...}}
+  ]
+}
+
+Rules:
+- Create 3 to 6 steps.
+- Use only tools from available_tools.
+- Keep args minimal and valid.
+- Do not add extra keys.
+- Do not output markdown.
+""".strip()
+
+FINAL_SYSTEM_PROMPT = """
+You are a reporting assistant.
+Write a short final summary in English for a US business audience.
+Include: manager name, month, gross sales (USD), refunds (USD), net sales (USD), refund rate (%), and key risk note.
+""".strip()
+
+TOOL_CATALOG = [
+    {
+        "name": "get_manager_profile",
+        "description": "Get manager profile by manager_id",
+        "args": {"manager_id": "integer"},
+    },
+    {
+        "name": "fetch_sales_data",
+        "description": "Get daily gross sales for a month",
+        "args": {"month": "string in YYYY-MM"},
+    },
+    {
+        "name": "fetch_refund_data",
+        "description": "Get daily refund values for a month",
+        "args": {"month": "string in YYYY-MM"},
+    },
+    {
+        "name": "calculate_monthly_kpis",
+        "description": "Calculate gross/refunds/net/order KPIs for a month",
+        "args": {"month": "string in YYYY-MM"},
+    },
+    {
+        "name": "detect_risk_signals",
+        "description": "Detect risk warnings for a month",
+        "args": {"month": "string in YYYY-MM"},
+    },
+]
+
+
+def _get_client() -> OpenAI:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise EnvironmentError(
+            "OPENAI_API_KEY is not set. Run: export OPENAI_API_KEY='sk-...'"
+        )
+    return OpenAI(api_key=api_key)
+
+
+def create_plan(goal: str, max_plan_steps: int) -> dict[str, Any]:
+    payload = {
+        "goal": goal,
+        "max_plan_steps": max_plan_steps,
+        "available_tools": TOOL_CATALOG,
+    }
+
+    client = _get_client()
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL,
+            temperature=0,
+            timeout=LLM_TIMEOUT_SECONDS,
+            response_format={"type": "json_object"},
+            messages=[
+                {"role": "system", "content": PLAN_SYSTEM_PROMPT},
+                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
+            ],
+        )
+    except (APITimeoutError, APIConnectionError) as exc:
+        raise LLMTimeout("llm_timeout") from exc
+
+    text = completion.choices[0].message.content or "{}"
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return {"kind": "invalid", "raw": text}
+
+
+def compose_final_answer(goal: str, history: list[dict[str, Any]]) -> str:
+    payload = {
+        "goal": goal,
+        "history": history,
+    }
+
+    client = _get_client()
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL,
+            temperature=0,
+            timeout=LLM_TIMEOUT_SECONDS,
+            messages=[
+                {"role": "system", "content": FINAL_SYSTEM_PROMPT},
+                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
+            ],
+        )
+    except (APITimeoutError, APIConnectionError) as exc:
+        raise LLMTimeout("llm_timeout") from exc
+
+    text = completion.choices[0].message.content or ""
+    text = text.strip()
+    if not text:
+        raise LLMEmpty("llm_empty")
+    return text