feat(examples/python): add runnable example

AgentPatterns · AgentPatterns · commit 8f344968279c · 2026-02-26T05:40:56.000+02:00
diff --git a/examples/react-agent/python/README.md b/examples/react-agent/python/README.md
@@ -0,0 +1,16 @@
+📖 Based on article:
+
+EN:
+https://agentpatterns.tech/en/agent-patterns/react-agent
+
+DE:
+https://agentpatterns.tech/de/agent-patterns/react-agent
+
+FR:
+https://agentpatterns.tech/fr/agent-patterns/react-agent
+
+ES:
+https://agentpatterns.tech/es/agent-patterns/react-agent
+
+UK:
+https://agentpatterns.tech/uk/agent-patterns/react-agent
diff --git a/examples/react-agent/python/gateway.py b/examples/react-agent/python/gateway.py
@@ -0,0 +1,116 @@
+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_steps: int = 8
+    max_tool_calls: int = 6
+    max_seconds: int = 20
+
+
+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_action(action: Any) -> dict[str, Any]:
+    if not isinstance(action, dict):
+        raise StopRun("invalid_action:not_object")
+
+    kind = action.get("kind")
+    if kind == "invalid":
+        raise StopRun("invalid_action:bad_json")
+    if kind not in {"tool", "final"}:
+        raise StopRun("invalid_action:bad_kind")
+
+    if kind == "final":
+        allowed = {"kind", "answer"}
+        extra = set(action.keys()) - allowed
+        if extra:
+            raise StopRun("invalid_action:extra_keys")
+        answer = action.get("answer")
+        if not isinstance(answer, str) or not answer.strip():
+            raise StopRun("invalid_action:missing_answer")
+        return {"kind": "final", "answer": answer.strip()}
+
+    allowed = {"kind", "name", "args"}
+    extra = set(action.keys()) - allowed
+    if extra:
+        raise StopRun("invalid_action:extra_keys")
+
+    name = action.get("name")
+    if not isinstance(name, str) or not name:
+        raise StopRun("invalid_action:missing_tool_name")
+
+    args = action.get("args", {})
+    if args is None:
+        args = {}
+    if not isinstance(args, dict):
+        raise StopRun("invalid_action:bad_args")
+
+    return {"kind": "tool", "name": name, "args": args}
+
+
+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
diff --git a/examples/react-agent/python/llm.py b/examples/react-agent/python/llm.py
@@ -0,0 +1,104 @@
+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
+
+SYSTEM_PROMPT = """
+You are a ReAct decision engine.
+Return only one JSON object with one of these shapes:
+1) {"kind":"tool","name":"<tool_name>","args":{...}}
+2) {"kind":"final","answer":"<short final answer>"}
+
+Rules:
+- Use tools when you do not have enough facts.
+- Do not invent tool outputs.
+- Prefer the smallest next step.
+- When evidence is sufficient, return "final".
+- Never output markdown or extra keys.
+""".strip()
+
+TOOL_CATALOG = [
+    {
+        "name": "get_user_profile",
+        "description": "Get user profile by user_id",
+        "args": {"user_id": "integer"},
+    },
+    {
+        "name": "get_user_billing",
+        "description": "Get billing info by user_id",
+        "args": {"user_id": "integer"},
+    },
+    {
+        "name": "search_policy",
+        "description": "Search refund and billing policy snippets",
+        "args": {"query": "string"},
+    },
+]
+
+
+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 _build_state_summary(history: list[dict[str, Any]]) -> dict[str, Any]:
+    tools_used = [
+        step.get("action", {}).get("name")
+        for step in history
+        if isinstance(step, dict)
+        and isinstance(step.get("action"), dict)
+        and step.get("action", {}).get("kind") == "tool"
+    ]
+    last_observation = history[-1].get("observation") if history else None
+    return {
+        "steps_completed": len(history),
+        "tools_used": tools_used,
+        "last_observation": last_observation,
+    }
+
+
+def decide_next_action(goal: str, history: list[dict[str, Any]]) -> dict[str, Any]:
+    # Keep full history in memory, but send summary + last N steps
+    # so the prompt remains stable as runs get longer.
+    recent_history = history[-3:]
+    payload = {
+        "goal": goal,
+        "state_summary": _build_state_summary(history),
+        "recent_history": recent_history,
+        "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": 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}
diff --git a/examples/react-agent/python/main.py b/examples/react-agent/python/main.py
@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+import json
+import time
+from typing import Any
+
+from gateway import Budget, StopRun, ToolGateway, args_hash, validate_action
+from llm import LLMTimeout, decide_next_action
+from tools import get_user_billing, get_user_profile, search_policy
+
+GOAL = (
+    "User 42 asked: Can I get a refund now? "
+    "Use tools to verify profile, billing state, and policy. "
+    "Return a short final answer in English with USD amount and reason."
+)
+
+BUDGET = Budget(max_steps=8, max_tool_calls=5, max_seconds=20)
+
+TOOL_REGISTRY = {
+    "get_user_profile": get_user_profile,
+    "get_user_billing": get_user_billing,
+    "search_policy": search_policy,
+}
+
+ALLOWED_TOOLS = {"get_user_profile", "get_user_billing", "search_policy"}
+
+
+def run_react(goal: str) -> dict[str, Any]:
+    started = time.monotonic()
+    history: list[dict[str, Any]] = []
+    trace: list[dict[str, Any]] = []
+
+    gateway = ToolGateway(allow=ALLOWED_TOOLS, registry=TOOL_REGISTRY, budget=BUDGET)
+
+    for step in range(1, BUDGET.max_steps + 1):
+        elapsed = time.monotonic() - started
+        if elapsed > BUDGET.max_seconds:
+            return {
+                "status": "stopped",
+                "stop_reason": "max_seconds",
+                "trace": trace,
+                "history": history,
+            }
+
+        try:
+            raw_action = decide_next_action(goal=goal, history=history)
+        except LLMTimeout:
+            return {
+                "status": "stopped",
+                "stop_reason": "llm_timeout",
+                "trace": trace,
+                "history": history,
+            }
+
+        try:
+            action = validate_action(raw_action)
+        except StopRun as exc:
+            return {
+                "status": "stopped",
+                "stop_reason": exc.reason,
+                "raw_action": raw_action,
+                "trace": trace,
+                "history": history,
+            }
+
+        if action["kind"] == "final":
+            return {
+                "status": "ok",
+                "stop_reason": "success",
+                "answer": action["answer"],
+                "trace": trace,
+                "history": history,
+            }
+
+        tool_name = action["name"]
+        tool_args = action["args"]
+
+        try:
+            observation = gateway.call(tool_name, tool_args)
+            trace.append(
+                {
+                    "step": step,
+                    "tool": tool_name,
+                    "args_hash": args_hash(tool_args),
+                    "ok": True,
+                }
+            )
+        except StopRun as exc:
+            trace.append(
+                {
+                    "step": step,
+                    "tool": tool_name,
+                    "args_hash": args_hash(tool_args),
+                    "ok": False,
+                    "stop_reason": exc.reason,
+                }
+            )
+            return {
+                "status": "stopped",
+                "stop_reason": exc.reason,
+                "trace": trace,
+                "history": history,
+            }
+
+        history.append(
+            {
+                "step": step,
+                "action": action,
+                "observation": observation,
+            }
+        )
+
+    return {
+        "status": "stopped",
+        "stop_reason": "max_steps",
+        "trace": trace,
+        "history": history,
+    }
+
+
+def main() -> None:
+    result = run_react(GOAL)
+    print(json.dumps(result, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/react-agent/python/requirements.txt b/examples/react-agent/python/requirements.txt
@@ -0,0 +1 @@
+openai==2.21.0
diff --git a/examples/react-agent/python/tools.py b/examples/react-agent/python/tools.py
