ReAct Agent — Python (повна реалізація з LLM)

Production-style runnable приклад ReAct агента на Python з action schema, allowlist інструментів, бюджетами, loop detection і stop reasons.

На цій сторінці

Суть патерна (коротко)
Що демонструє цей приклад
Архітектура
Структура проєкту
Як запустити
Задача
Рішення
Код
tools.py — інструменти (джерело фактів)
gateway.py — policy boundary (найважливіший шар)
llm.py — decision step (Think)
main.py — повний ReAct loop
requirements.txt
Приклад виводу
Чому це саме ReAct, а не просто tool calling
Типові stop_reason
Що тут НЕ показано
Що спробувати далі
Повний код на GitHub

Суть патерна (коротко)

ReAct Agent — це патерн, у якому агент працює ітеративно: думає, обирає дію, виконує її та аналізує результат перед наступним кроком.

Модель приймає рішення на кожній ітерації, а виконання інструментів проходить через контрольований gateway з валідацією дій, allowlist і runtime-бюджетами.

Детальніше про ReAct Agent

Що демонструє цей приклад

повний цикл Think -> Act -> Observe
окремий policy boundary між decision (LLM) і tools (execution layer)
строгий формат дії: тільки tool або final
allowlist інструментів (deny by default)
бюджети run: max_steps, max_tool_calls, max_seconds
loop detection для повторного виклику того самого tool з тими самими аргументами
явні stop_reason для дебагу і продакшен-моніторингу

Архітектура

LLM отримує goal + історію кроків і повертає JSON-дію.
Система валідовує дію (validate_action).
Якщо це tool, ToolGateway перевіряє allowlist/бюджети/loop detection і виконує інструмент.
Observation додається в history і стає новим evidence для наступного decision step.
Якщо це final, run завершується з stop_reason="success".

LLM повертає intent (JSON action), який розглядається як недовірений input: policy boundary спершу валідовує його, а потім (якщо дозволено) викликає tools.

Так ReAct залишається керованим: модель приймає рішення, а policy-логіка контролює виконання.

Структура проєкту

TEXT

examples/
└── agent-patterns/
    └── react-agent/
        └── python/
            ├── main.py           # ReAct loop
            ├── llm.py            # LLM decision step (JSON action)
            ├── gateway.py        # policy boundary: validation, allowlist, budgets, loop detection
            ├── tools.py          # deterministic tools (Anna/Max, USD, policies)
            └── requirements.txt

Як запустити

BASH

git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd agentpatterns

cd examples/agent-patterns/react-agent/python
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

Потрібен Python 3.11+.

Варіант через export:

BASH

export OPENAI_API_KEY="sk-..."
# optional:
# export OPENAI_MODEL="gpt-4.1-mini"
# export OPENAI_TIMEOUT_SECONDS="60"

python main.py

Варіант через .env (опційно)

BASH

cat > .env <<'EOF'
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4.1-mini
OPENAI_TIMEOUT_SECONDS=60
EOF

set -a
source .env
set +a

python main.py

Це shell-варіант (macOS/Linux). На Windows простіше використовувати set змінних або, за бажанням, python-dotenv, щоб підвантажувати .env автоматично.

Задача

Уяви, що користувач пише в сапорт:

"Чи можу я зараз повернути гроші за підписку?"

Агент не має відповідати одразу. Він має:

зібрати факти через tools (профіль, білінг, правила)
після кожного кроку вирішити, що робити далі
дати фінальну відповідь лише коли фактів достатньо

Рішення

Тут агент працює крок за кроком (ReAct):

модель на кожному кроці обирає: викликати tool чи завершити
система перевіряє, що дія коректна і дозволена
tool повертає результат, який додається в історію
на основі цієї історії агент робить наступний крок
коли даних достатньо, агент повертає коротку фінальну відповідь

Код

`tools.py` — інструменти (джерело фактів)

PYTHON

from __future__ import annotations

from typing import Any

USERS = {
    42: {"id": 42, "name": "Anna", "country": "US", "tier": "pro"},
    7: {"id": 7, "name": "Max", "country": "US", "tier": "free"},
}

BILLING = {
    42: {
        "currency": "USD",
        "plan": "pro_monthly",
        "price_usd": 49.0,
        "days_since_first_payment": 10,
    },
    7: {
        "currency": "USD",
        "plan": "free",
        "price_usd": 0.0,
        "days_since_first_payment": 120,
    },
}

POLICY_DOCS = [
    {
        "id": "refund-v3",
        "title": "Refund Policy",
        "snippet": "Pro monthly subscriptions are refundable within 14 days from the first payment.",
    },
    {
        "id": "free-v1",
        "title": "Free Plan Policy",
        "snippet": "Free plan has no billable payments and cannot be refunded.",
    },
    {
        "id": "billing-v2",
        "title": "Billing Rules",
        "snippet": "All refunds are returned to the original payment method in USD.",
    },
]


def get_user_profile(user_id: int) -> dict[str, Any]:
    user = USERS.get(user_id)
    if not user:
        return {"error": f"user {user_id} not found"}
    return {"user": user}


def get_user_billing(user_id: int) -> dict[str, Any]:
    billing = BILLING.get(user_id)
    if not billing:
        return {"error": f"billing record for user {user_id} not found"}
    return {"billing": billing}


def search_policy(query: str) -> dict[str, Any]:
    words = [w for w in query.lower().split() if w]

    def score(doc: dict[str, str]) -> int:
        haystack = f"{doc['title']} {doc['snippet']}".lower()
        return sum(1 for word in words if word in haystack)

    ranked = sorted(POLICY_DOCS, key=score, reverse=True)
    top = [doc for doc in ranked if score(doc) > 0][:2]
    if not top:
        top = POLICY_DOCS[:1]

    return {"matches": top}

Що тут найважливіше (простими словами)

Інструменти детерміновані й не містять LLM-логіки. Агент лише вирішує, який tool викликати — але не виконує бізнес-логіку самостійно.

`gateway.py` — policy boundary (найважливіший шар)

PYTHON

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

Що тут найважливіше (простими словами)

validate_action(...) — це governance/control layer: система приймає тільки дозволений контракт дії й відхиляє все зайве (invalid_action:extra_keys).
Budget + StopRun(...) — це production-патерн керованого завершення: run не “пливе” безмежно, а зупиняється з чіткою причиною.
ToolGateway.call(...) — це межа agent ≠ executor: агент лише пропонує дію, а фактичний виклик tool виконує контрольований шар системи.
loop_detected ловить exact-repeat (той самий tool + ті самі args). Semantic-loop — окрема опція (див. “Що спробувати далі”).

`llm.py` — decision step (Think)

LLM бачить лише каталог доступних tools; якщо tool не в allowlist, gateway зупинить run.

PYTHON

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}

Що тут найважливіше (простими словами)

timeout=LLM_TIMEOUT_SECONDS + LLMTimeout — це production-патерн: якщо модель зависла, run повертає явний llm_timeout.
state_summary + recent_history — це production-патерн масштабу: контекст росте керовано, а не “без ліміту” на кожному кроці.
SYSTEM_PROMPT описує тільки формат intent (tool/final) — тобто LLM вирішує що зробити, але не виконує інструменти сама.

`main.py` — повний ReAct loop

PYTHON

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

Що тут найважливіше (простими словами)

run_react(...) — керує циклом і stop conditions; бізнес-дії виконуються тільки через ToolGateway.
validate_action(...) і gateway.call(...) у середині loop — це governance/control layer в дії на кожному кроці.
Розділення decide_next_action(...) і gateway.call(...) — головний принцип agent ≠ executor: агент повертає intent, а tools викликаються лише через policy boundary.

`requirements.txt`

TEXT

openai==2.21.0

Приклад виводу

Порядок tool calls може трохи відрізнятись між запусками, але stop_reason і policy-гейти (allowlist, budget, validation) залишаються стабільними.

JSON

{
  "status": "ok",
  "stop_reason": "success",
  "answer": "Yes, you can get a refund of USD 49.00 ...",
  "trace": [
    {"step": 1, "tool": "get_user_profile", "args_hash": "...", "ok": true},
    {"step": 2, "tool": "get_user_billing", "args_hash": "...", "ok": true},
    {"step": 3, "tool": "search_policy", "args_hash": "...", "ok": true}
  ],
  "history": [{...}]
}

history — це журнал виконання кроків: для кожного step там зберігаються action (що агент вирішив зробити) і observation (що повернув tool).

args_hash — це hash аргументів, тому для одного user_id він може збігатися між різними tools; loop guard перевіряє саме комбінацію tool + args_hash.

Чому це саме ReAct, а не просто tool calling

	Разовий виклик	ReAct loop
Рішення після кожного observation	❌	✅
Явні stop reasons	❌	✅
Контроль повторних однакових дій	❌	✅
Бюджет run (steps/tools/time)	частково	✅

Типові `stop_reason`

success — агент повернув фінальну відповідь
max_steps — вичерпано кроки
max_tool_calls — вичерпано ліміт викликів інструментів
max_seconds — перевищено time budget
llm_timeout — LLM не відповів у межах OPENAI_TIMEOUT_SECONDS
loop_detected — повторився той самий tool call з тими самими args
tool_denied:<name> — інструмент не в allowlist
invalid_action:* — модель повернула некоректну структуру дії

Що тут НЕ показано

Немає auth/PII та продакшен-контролів доступу до персональних даних.
Немає retry/backoff політик для LLM і tool-layer.
Немає бюджетів по токенах/вартості (cost guardrails).
Інструменти тут детерміновані mocks для навчання, а не реальні зовнішні API.

Що спробувати далі

Прибери search_policy з ALLOWED_TOOLS і подивись, як зміниться stop_reason.
Постав max_tool_calls=1 і перевір, що агента зупиняє policy, а не модель.
Зміни GOAL на user_id=7 (Max) і перевір фінальну відповідь.
Спробуй викликати неіснуючий tool (модель інколи так робить) — побачиш tool_missing:*.
Додай soft-loop режим: нормалізуй string-аргументи (trim + collapse spaces) перед hash, щоб ловити семантично однакові повтори.
Додай JSONL лог кроків (trace) для observability у продакшені.

Повний код на GitHub

У репозиторії лежить повна runnable-версія цього прикладу: ReAct loop, policy boundary, allowlist, budgets, loop detection і stop reasons.

Переглянути повний код на GitHub ↗

Task Decomposition Agent — Python (повна реалізація з LLM)

Повернутися

Коли агент має зупинитись (і хто це вирішує) - Python

Task Decomposition Agent — Python (повна реалізація з LLM)

⏱️ 12 хв читання • Оновлено Бер, 2026Складність: ★★☆

Використовується в патернах

Пов’язані відмови

Потрібні контролі

Інтегровано: продакшен-контрольOnceOnly

Додай guardrails до агентів з tool-calling

Зашип цей патерн з governance:

Бюджетами (кроки / ліміти витрат)
Дозволами на інструменти (allowlist / blocklist)
Kill switch та аварійна зупинка
Ідемпотентність і dedupe
Audit logs та трасування

Спробувати OnceOnly Документація та приклади

Інтегрована згадка: OnceOnly — контрольний шар для продакшен агент-систем.

Автор

Цю документацію курують і підтримують інженери, які запускають AI-агентів у продакшені.

Контент створено з допомогою AI, із людською редакторською відповідальністю за точність, ясність і продакшн-релевантність.

Патерни та рекомендації базуються на постмортемах, режимах відмов і операційних інцидентах у розгорнутих системах, зокрема під час розробки та експлуатації governance-інфраструктури для агентів у OnceOnly.

ReAct Agent — Python (повна реалізація з LLM)

Суть патерна (коротко)

Що демонструє цей приклад

Архітектура

Структура проєкту

Як запустити

Задача

Рішення

Код

tools.py — інструменти (джерело фактів)

Що тут найважливіше (простими словами)

gateway.py — policy boundary (найважливіший шар)

Що тут найважливіше (простими словами)

llm.py — decision step (Think)

Що тут найважливіше (простими словами)

main.py — повний ReAct loop

Що тут найважливіше (простими словами)

requirements.txt

Приклад виводу

Чому це саме ReAct, а не просто tool calling

Типові stop_reason

Що тут НЕ показано

Що спробувати далі

Повний код на GitHub

Використовується в патернах

Пов’язані відмови

Потрібні контролі

`tools.py` — інструменти (джерело фактів)

`gateway.py` — policy boundary (найважливіший шар)

`llm.py` — decision step (Think)

`main.py` — повний ReAct loop

`requirements.txt`

Типові `stop_reason`