ReAct Agent — Python (implementación completa con LLM)

Ejemplo ejecutable de agente ReAct en Python con estilo de producción, con esquema de acciones, allowlist de herramientas, presupuestos, detección de bucles y stop reasons.

En esta página

Esencia del patrón (breve)
Qué demuestra este ejemplo
Arquitectura
Estructura del proyecto
Cómo ejecutar
Tarea
Solución
Código
tools.py — herramientas (fuente de hechos)
gateway.py — policy boundary (la capa más importante)
llm.py — decision step (Think)
main.py — loop ReAct completo
requirements.txt
Ejemplo de salida
Por qué esto es ReAct y no solo tool calling
stop_reason típicos
Qué NO se muestra aquí
Qué probar después
Código completo en GitHub

Esencia del patrón (breve)

ReAct Agent es un patrón en el que el agente trabaja de forma iterativa: piensa, elige una acción, la ejecuta y analiza el resultado antes del siguiente paso.

El modelo toma decisiones en cada iteración, y la ejecución de herramientas pasa por un gateway controlado con validación de acciones, allowlist y presupuestos de runtime.

Más sobre ReAct Agent

Qué demuestra este ejemplo

ciclo completo Think -> Act -> Observe
policy boundary separado entre decision (LLM) y tools (execution layer)
formato de acción estricto: solo tool o final
allowlist de herramientas (deny by default)
presupuestos de run: max_steps, max_tool_calls, max_seconds
loop detection para llamadas repetidas del mismo tool con los mismos args
stop_reason explícitos para depuración y monitoreo en producción

Arquitectura

El LLM recibe el goal + historial de pasos y devuelve una acción JSON.
El sistema valida la acción (validate_action).
Si es tool, ToolGateway revisa allowlist/presupuestos/loop detection y ejecuta la herramienta.
Observation se agrega a history y se convierte en nueva evidencia para el siguiente decision step.
Si es final, el run termina con stop_reason="success".

El LLM devuelve intent (acción JSON), que se trata como input no confiable: el policy boundary primero lo valida y luego (si está permitido) llama tools.

Así ReAct se mantiene controlable: el modelo decide, y la lógica de policy controla la ejecución.

Estructura del proyecto

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

Cómo ejecutar

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

Se requiere Python 3.11+.

Opción con export:

BASH

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

python main.py

Opción con .env (opcional)

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

Esta es la variante de shell (macOS/Linux). En Windows es más fácil usar variables con set o, si lo prefieres, python-dotenv para cargar .env automáticamente.

Tarea

Imagina que un usuario escribe al soporte:

"¿Puedo obtener un reembolso de mi suscripción ahora mismo?"

El agente no debe responder de inmediato. Debe:

recopilar hechos mediante tools (perfil, facturación, políticas)
decidir qué hacer después de cada paso
dar la respuesta final solo cuando haya suficientes hechos

Solución

Aquí el agente trabaja paso a paso (ReAct):

en cada paso, el modelo elige: llamar una tool o finalizar
el sistema verifica que la acción sea válida y permitida
la tool devuelve un resultado que se agrega al historial
con base en ese historial, el agente realiza el siguiente paso
cuando los datos son suficientes, el agente devuelve una respuesta final breve

Código

`tools.py` — herramientas (fuente de hechos)

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}

Qué es lo más importante aquí (en palabras simples)

Las herramientas son determinísticas y no contienen lógica de LLM. El agente solo decide qué tool llamar, pero no ejecuta la lógica de negocio por sí solo.

`gateway.py` — policy boundary (la capa más 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

Qué es lo más importante aquí (en palabras simples)

validate_action(...) es la governance/control layer: el sistema acepta solo el contrato de acción permitido y rechaza todo lo extra (invalid_action:extra_keys).
Budget + StopRun(...) es un patrón de producción para finalización controlada: el run no se alarga indefinidamente, se detiene con una razón clara.
ToolGateway.call(...) es el límite de agent ≠ executor: el agente solo propone una acción; la llamada real de tool la ejecuta una capa controlada del sistema.
loop_detected detecta exact-repeat (misma tool + mismos args). Semantic-loop es una opción aparte (ver “Qué probar después”).

`llm.py` — decision step (Think)

El LLM ve solo el catálogo de tools disponibles; si una tool no está en la allowlist, el gateway detiene el 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}

Qué es lo más importante aquí (en palabras simples)

timeout=LLM_TIMEOUT_SECONDS + LLMTimeout es un patrón de producción: si el modelo se cuelga, el run devuelve llm_timeout explícito.
state_summary + recent_history es un patrón de escalado en producción: el contexto crece de forma controlada, no sin límite en cada paso.
SYSTEM_PROMPT define solo el formato de intent (tool/final) - el LLM decide qué hacer, pero no ejecuta herramientas por sí mismo.

`main.py` — loop ReAct completo

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

Qué es lo más importante aquí (en palabras simples)

run_react(...) controla el ciclo y las stop conditions; las acciones de negocio se ejecutan solo a través de ToolGateway.
validate_action(...) y gateway.call(...) dentro del loop son la governance/control layer en acción en cada paso.
Separar decide_next_action(...) y gateway.call(...) es el principio clave de agent ≠ executor: el agente devuelve intent, y las tools se llaman solo a través del policy boundary.

`requirements.txt`

TEXT

openai==2.21.0

Ejemplo de salida

El orden de llamadas a tools puede variar un poco entre ejecuciones, pero stop_reason y los policy gates (allowlist, budget, validation) se mantienen estables.

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 es el registro de ejecución de pasos: para cada step guarda action (qué decidió hacer el agente) y observation (qué devolvió la tool).

args_hash es un hash de argumentos, por eso para el mismo user_id puede coincidir entre distintas tools; el loop guard verifica la combinación tool + args_hash.

Por qué esto es ReAct y no solo tool calling

	Llamada única	ReAct loop
Decisión después de cada observation	❌	✅
Stop reasons explícitos	❌	✅
Control de acciones repetidas idénticas	❌	✅
Presupuesto de run (steps/tools/time)	parcial	✅

`stop_reason` típicos

success — el agente devolvió una respuesta final
max_steps — se agotó el presupuesto de pasos
max_tool_calls — se agotó el límite de llamadas a herramientas
max_seconds — se superó el presupuesto de tiempo
llm_timeout — el LLM no respondió dentro de OPENAI_TIMEOUT_SECONDS
loop_detected — se repitió la misma llamada de tool con los mismos args
tool_denied:<name> — la herramienta no está en la allowlist
invalid_action:* — el modelo devolvió una estructura de acción inválida

Qué NO se muestra aquí

No hay auth/PII ni controles de acceso de producción para datos personales.
No hay políticas de retry/backoff para LLM y la capa de herramientas.
No hay presupuestos por tokens/costo (cost guardrails).
Las herramientas aquí son mocks determinísticos para aprendizaje, no APIs externas reales.

Qué probar después

Quita search_policy de ALLOWED_TOOLS y mira cómo cambia stop_reason.
Pon max_tool_calls=1 y verifica que quien detiene al agente es la policy, no el modelo.
Cambia GOAL a user_id=7 (Max) y revisa la respuesta final.
Intenta llamar una tool inexistente (el modelo a veces lo hace) - verás tool_missing:*.
Agrega modo soft-loop: normaliza argumentos string (trim + collapse spaces) antes del hash para capturar repeticiones semánticamente iguales.
Agrega logs JSONL de pasos (trace) para observabilidad en producción.

Código completo en GitHub

En el repositorio está la versión runnable completa de este ejemplo: ReAct loop, policy boundary, allowlist, budgets, loop detection y stop reasons.

Ver código completo en GitHub ↗

Leer siguiente

Task Decomposition Agent — Python (implementación completa con LLM)

Volver

Condiciones de parada - Python

Leer siguiente

Task Decomposition Agent — Python (implementación completa con LLM)

⏱️ 13 min de lectura • Actualizado Mar, 2026Dificultad: ★★☆

Usado por patrones

Fallos relacionados

Gobernanza requerida

Integrado: control en producciónOnceOnly

Guardrails para agentes con tool-calling

Lleva este patrón a producción con gobernanza:

Presupuestos (pasos / topes de gasto)
Permisos de herramientas (allowlist / blocklist)
Kill switch y parada por incidente
Idempotencia y dedupe
Audit logs y trazabilidad

Probar OnceOnly Docs y ejemplos

Mención integrada: OnceOnly es una capa de control para sistemas de agentes en producción.

Autor

Esta documentación está curada y mantenida por ingenieros que despliegan agentes de IA en producción.

El contenido es asistido por IA, con responsabilidad editorial humana sobre la exactitud, la claridad y la relevancia en producción.

Los patrones y las recomendaciones se basan en post-mortems, modos de fallo e incidentes operativos en sistemas desplegados, incluido durante el desarrollo y la operación de infraestructura de gobernanza para agentes en OnceOnly.

ReAct Agent — Python (implementación completa con LLM)

Esencia del patrón (breve)

Qué demuestra este ejemplo

Arquitectura

Estructura del proyecto

Cómo ejecutar

Tarea

Solución

Código

tools.py — herramientas (fuente de hechos)

Qué es lo más importante aquí (en palabras simples)

gateway.py — policy boundary (la capa más importante)

Qué es lo más importante aquí (en palabras simples)

llm.py — decision step (Think)

Qué es lo más importante aquí (en palabras simples)

main.py — loop ReAct completo

Qué es lo más importante aquí (en palabras simples)

requirements.txt

Ejemplo de salida

Por qué esto es ReAct y no solo tool calling

stop_reason típicos

Qué NO se muestra aquí

Qué probar después

Código completo en GitHub

Usado por patrones

Fallos relacionados

Gobernanza requerida

`tools.py` — herramientas (fuente de hechos)

`gateway.py` — policy boundary (la capa más importante)

`llm.py` — decision step (Think)

`main.py` — loop ReAct completo

`requirements.txt`

`stop_reason` típicos