ReAct Agent — Python (implémentation complète avec LLM)

Exemple exécutable d’agent ReAct en Python, style production, avec schéma d’action, allowlist d’outils, budgets, détection de boucle et stop reasons.

Sur cette page

Essence du pattern (bref)
Ce que cet exemple démontre
Architecture
Structure du projet
Lancer le projet
Tâche
Solution
Code
tools.py — outils (source de faits)
gateway.py — policy boundary (la couche la plus importante)
llm.py — decision step (Think)
main.py — boucle ReAct complète
requirements.txt
Exemple de sortie
Pourquoi c’est du ReAct et pas juste du tool calling
stop_reason typiques
Ce qui N’est PAS montré ici
Que tester ensuite
Code complet sur GitHub

Essence du pattern (bref)

ReAct Agent est un pattern dans lequel l'agent travaille de façon itérative : il réfléchit, choisit une action, l'exécute et analyse le résultat avant l'étape suivante.

Le modèle prend des décisions à chaque itération, et l'exécution des outils passe par un gateway contrôlé avec validation des actions, allowlist et budgets runtime.

En savoir plus sur ReAct Agent

Ce que cet exemple démontre

boucle complète Think -> Act -> Observe
frontière de policy séparée entre decision (LLM) et tools (execution layer)
format d’action strict : uniquement tool ou final
allowlist d’outils (deny by default)
budgets de run : max_steps, max_tool_calls, max_seconds
loop detection pour les appels répétés du même tool avec les mêmes args
stop_reason explicites pour le débogage et le monitoring production

Architecture

Le LLM reçoit le goal + l’historique des étapes et renvoie une action JSON.
Le système valide l’action (validate_action).
Si c’est un tool, ToolGateway vérifie allowlist/budgets/loop detection et exécute l’outil.
Observation est ajoutée à history et devient une nouvelle preuve pour la decision step suivante.
Si c’est final, le run se termine avec stop_reason="success".

Le LLM renvoie un intent (action JSON), traité comme un input non fiable : la policy boundary le valide d’abord puis (si autorisé) appelle les tools.

Ainsi ReAct reste contrôlable : le modèle décide, et la logique de policy contrôle l’exécution.

Structure du projet

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

Lancer le projet

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+ est requis.

Option via export :

BASH

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

python main.py

Option via .env (optionnel)

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

C’est la variante shell (macOS/Linux). Sur Windows, il est plus simple d’utiliser des variables set ou, si souhaité, python-dotenv pour charger .env automatiquement.

Tâche

Imagine qu’un utilisateur écrit au support :

"Est-ce que je peux être remboursé de mon abonnement maintenant ?"

L’agent ne doit pas répondre immédiatement. Il doit :

collecter les faits via les tools (profil, facturation, politiques)
décider quoi faire après chaque étape
donner la réponse finale seulement quand les faits sont suffisants

Solution

Ici, l’agent travaille étape par étape (ReAct) :

à chaque étape, le modèle choisit : appeler un tool ou terminer
le système vérifie que l’action est correcte et autorisée
le tool renvoie un résultat qui est ajouté à l’historique
sur la base de cet historique, l’agent fait l’étape suivante
quand les données sont suffisantes, l’agent renvoie une réponse finale courte

Code

`tools.py` — outils (source de faits)

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}

Ce qui est le plus important ici (en mots simples)

Les outils sont déterministes et ne contiennent pas de logique LLM. L’agent décide seulement quel tool appeler, mais n’exécute pas lui-même la logique métier.

`gateway.py` — policy boundary (la couche la plus importante)

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

Ce qui est le plus important ici (en mots simples)

validate_action(...) est la governance/control layer : le système accepte seulement le contrat d’action autorisé et rejette tout surplus (invalid_action:extra_keys).
Budget + StopRun(...) est un pattern de production de terminaison contrôlée : un run ne dérive pas indéfiniment, il s’arrête avec une raison claire.
ToolGateway.call(...) est la frontière agent ≠ executor : l’agent propose seulement une action, et l’appel réel du tool est exécuté par une couche système contrôlée.
loop_detected attrape les exact-repeat (même tool + mêmes args). Semantic-loop est une option séparée (voir « Que tester ensuite »).

`llm.py` — decision step (Think)

Le LLM voit seulement le catalogue des tools disponibles ; si un tool n’est pas dans l’allowlist, le gateway arrête le 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}

Ce qui est le plus important ici (en mots simples)

timeout=LLM_TIMEOUT_SECONDS + LLMTimeout est un pattern de production : si le modèle se bloque, le run renvoie llm_timeout explicitement.
state_summary + recent_history est un pattern de passage à l’échelle en production : le contexte grandit de façon contrôlée, pas sans limite à chaque étape.
SYSTEM_PROMPT décrit uniquement le format d’intent (tool/final) - le LLM décide quoi faire, mais n’exécute pas les outils lui-même.

`main.py` — boucle ReAct complète

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

Ce qui est le plus important ici (en mots simples)

run_react(...) pilote la boucle et les stop conditions ; les actions métier passent uniquement par ToolGateway.
validate_action(...) et gateway.call(...) dans la boucle sont la governance/control layer en action à chaque étape.
La séparation de decide_next_action(...) et gateway.call(...) est le principe clé agent ≠ executor : l’agent renvoie l’intent, et les tools sont appelés uniquement via la policy boundary.

`requirements.txt`

TEXT

openai==2.21.0

Exemple de sortie

L’ordre des appels de tools peut varier légèrement entre les runs, mais stop_reason et les policy-gates (allowlist, budget, validation) restent stables.

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 est le journal d’exécution des étapes : pour chaque step, il stocke action (ce que l’agent a décidé de faire) et observation (ce que le tool a renvoyé).

args_hash est un hash des arguments ; pour un même user_id, il peut donc coïncider entre différents tools ; le loop guard vérifie la combinaison tool + args_hash.

Pourquoi c’est du ReAct et pas juste du tool calling

	Appel unique	ReAct loop
Décision après chaque observation	❌	✅
Stop reasons explicites	❌	✅
Contrôle des actions répétées identiques	❌	✅
Budget de run (steps/tools/time)	partiel	✅

`stop_reason` typiques

success — l’agent a renvoyé une réponse finale
max_steps — budget d’étapes épuisé
max_tool_calls — limite d’appels d’outils épuisée
max_seconds — budget temps dépassé
llm_timeout — le LLM n’a pas répondu dans OPENAI_TIMEOUT_SECONDS
loop_detected — le même appel de tool avec les mêmes args s’est répété
tool_denied:<name> — l’outil n’est pas dans l’allowlist
invalid_action:* — le modèle a renvoyé une structure d’action invalide

Ce qui N’est PAS montré ici

Pas d’auth/PII ni de contrôles d’accès production aux données personnelles.
Pas de politiques retry/backoff pour le LLM et la couche tools.
Pas de budgets token/coût (cost guardrails).
Les outils ici sont des mocks déterministes pour l’apprentissage, pas de vraies API externes.

Que tester ensuite

Retire search_policy de ALLOWED_TOOLS et observe comment stop_reason change.
Mets max_tool_calls=1 et vérifie que c’est la policy qui arrête l’agent, pas le modèle.
Change GOAL en user_id=7 (Max) et vérifie la réponse finale.
Essaie d’appeler un tool inexistant (le modèle le fait parfois) - tu verras tool_missing:*.
Ajoute un mode soft-loop : normalise les args string (trim + collapse spaces) avant le hash pour attraper des répétitions sémantiquement identiques.
Ajoute des logs d’étapes JSONL (trace) pour l’observability en production.

Code complet sur GitHub

Le dépôt contient la version runnable complète de cet exemple : ReAct loop, policy boundary, allowlist, budgets, loop detection et stop reasons.

Voir le code complet sur GitHub ↗

Task Decomposition Agent — Python (implémentation complète avec LLM)

Retour

Conditions d’arrêt - Python

Task Decomposition Agent — Python (implémentation complète avec LLM)

⏱️ 13 min de lecture • Mis à jour Mars, 2026Difficulté: ★★☆

Utilisé par les patterns

Pannes associées

Gouvernance requise

Intégré : contrôle en productionOnceOnly

Ajoutez des garde-fous aux agents tool-calling

Livrez ce pattern avec de la gouvernance :

Budgets (steps / plafonds de coût)
Permissions outils (allowlist / blocklist)
Kill switch & arrêt incident
Idempotence & déduplication
Audit logs & traçabilité

Essayer OnceOnly Docs & exemples

Mention intégrée : OnceOnly est une couche de contrôle pour des systèmes d’agents en prod.

Auteur

Cette documentation est organisée et maintenue par des ingénieurs qui déploient des agents IA en production.

Le contenu est assisté par l’IA, avec une responsabilité éditoriale humaine quant à l’exactitude, la clarté et la pertinence en production.

Les patterns et recommandations s’appuient sur des post-mortems, des modes de défaillance et des incidents opérationnels dans des systèmes déployés, notamment lors du développement et de l’exploitation d’une infrastructure de gouvernance pour les agents chez OnceOnly.

ReAct Agent — Python (implémentation complète avec LLM)

Essence du pattern (bref)

Ce que cet exemple démontre

Architecture

Structure du projet

Lancer le projet

Tâche

Solution

Code

tools.py — outils (source de faits)

Ce qui est le plus important ici (en mots simples)

gateway.py — policy boundary (la couche la plus importante)

Ce qui est le plus important ici (en mots simples)

llm.py — decision step (Think)

Ce qui est le plus important ici (en mots simples)

main.py — boucle ReAct complète

Ce qui est le plus important ici (en mots simples)

requirements.txt

Exemple de sortie

Pourquoi c’est du ReAct et pas juste du tool calling

stop_reason typiques

Ce qui N’est PAS montré ici

Que tester ensuite

Code complet sur GitHub

Utilisé par les patterns

Pannes associées

Gouvernance requise

`tools.py` — outils (source de faits)

`gateway.py` — policy boundary (la couche la plus importante)

`llm.py` — decision step (Think)

`main.py` — boucle ReAct complète

`requirements.txt`

`stop_reason` typiques