ReAct Agent — Python (vollständige Implementierung mit LLM)

Ausführbares ReAct-Agent-Beispiel im Production-Stil für Python mit Action-Schema, Tool-Allowlist, Budgets, Loop Detection und Stop-Reasons.

Auf dieser Seite

Kern des Musters (Kurz)
Was dieses Beispiel zeigt
Architektur
Projektstruktur
Ausführen
Aufgabe
Lösung
Code
tools.py — Tools (Quelle der Fakten)
gateway.py — Policy Boundary (wichtigste Schicht)
llm.py — decision step (Think)
main.py — vollständiger ReAct-Loop
requirements.txt
Beispielausgabe
Warum das ReAct ist und nicht nur Tool Calling
Typische stop_reason-Werte
Was hier NICHT gezeigt wird
Was als Nächstes probieren
Vollständiger Code auf GitHub

Kern des Musters (Kurz)

ReAct Agent ist ein Muster, bei dem der Agent iterativ arbeitet: denkt, eine Aktion auswählt, sie ausführt und das Ergebnis vor dem nächsten Schritt analysiert.

Das Modell trifft bei jeder Iteration Entscheidungen, und die Tool-Ausführung läuft über ein kontrolliertes Gateway mit Action-Validierung, Allowlist und Runtime-Budgets.

Mehr über ReAct Agent

Was dieses Beispiel zeigt

vollständiger Think -> Act -> Observe-Zyklus
klare Policy-Grenze zwischen Decision (LLM) und Tools (Execution Layer)
striktes Action-Format: nur tool oder final
Tool-Allowlist (deny by default)
Run-Budgets: max_steps, max_tool_calls, max_seconds
Loop Detection für wiederholte Aufrufe desselben Tools mit denselben Args
explizite stop_reason-Werte für Debugging und Production-Monitoring

Architektur

Das LLM erhält Goal + Step-History und gibt eine JSON-Action zurück.
Das System validiert die Action (validate_action).
Wenn es ein tool ist, prüft ToolGateway Allowlist/Budgets/Loop Detection und führt das Tool aus.
Observation wird zu history hinzugefügt und wird neue Evidenz für den nächsten Decision-Step.
Wenn es final ist, endet der Run mit stop_reason="success".

Das LLM gibt Intent (JSON-Action) zurück, der als untrusted Input behandelt wird: Die Policy-Grenze validiert zuerst und ruft erst dann (wenn erlaubt) Tools auf.

So bleibt ReAct kontrollierbar: Das Modell trifft Entscheidungen, und die Policy-Logik kontrolliert die Ausführung.

Projektstruktur

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

Ausführen

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+ ist erforderlich.

Variante über export:

BASH

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

python main.py

Variante über .env (optional)

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

Das ist die Shell-Variante (macOS/Linux). Unter Windows ist es einfacher, set-Umgebungsvariablen zu verwenden oder optional python-dotenv, um .env automatisch zu laden.

Aufgabe

Stell dir vor, ein Nutzer schreibt an den Support:

"Kann ich jetzt eine Rückerstattung für mein Abo bekommen?"

Der Agent sollte nicht sofort antworten. Er muss:

Fakten über Tools sammeln (Profil, Billing, Richtlinien)
nach jedem Schritt entscheiden, was als Nächstes zu tun ist
die finale Antwort erst geben, wenn genug Fakten vorliegen

Lösung

Hier arbeitet der Agent Schritt für Schritt (ReAct):

das Modell wählt bei jedem Schritt: Tool aufrufen oder beenden
das System prüft, dass die Action korrekt und erlaubt ist
das Tool gibt ein Ergebnis zurück, das zur History hinzugefügt wird
auf Basis dieser History macht der Agent den nächsten Schritt
wenn die Daten ausreichen, gibt der Agent eine kurze finale Antwort zurück

Code

`tools.py` — Tools (Quelle der Fakten)

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}

Was hier am wichtigsten ist (einfach erklärt)

Tools sind deterministisch und enthalten keine LLM-Logik. Der Agent entscheidet nur, welches Tool aufgerufen wird, führt aber die Business-Logik nicht selbst aus.

`gateway.py` — Policy Boundary (wichtigste Schicht)

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

Was hier am wichtigsten ist (einfach erklärt)

validate_action(...) ist die Governance/Control Layer: Das System akzeptiert nur den erlaubten Action-Vertrag und lehnt alles Zusätzliche ab (invalid_action:extra_keys).
Budget + StopRun(...) ist ein Production-Muster für kontrolliertes Beenden: Ein Run läuft nicht unbegrenzt weiter, sondern stoppt mit klarem Grund.
ToolGateway.call(...) ist die Grenze von agent ≠ executor: Der Agent schlägt nur eine Action vor, der echte Tool-Aufruf erfolgt in einer kontrollierten Systemschicht.
loop_detected erkennt exact-repeat (gleiches Tool + gleiche Args). Semantic-Loop ist eine eigene Option (siehe „Was als Nächstes probieren“).

`llm.py` — decision step (Think)

Das LLM sieht nur den Katalog verfügbarer Tools; ist ein Tool nicht in der Allowlist, stoppt das Gateway den 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}

Was hier am wichtigsten ist (einfach erklärt)

timeout=LLM_TIMEOUT_SECONDS + LLMTimeout ist ein Production-Muster: Wenn das Modell hängt, gibt der Run explizit llm_timeout zurück.
state_summary + recent_history ist ein Production-Skalierungsmuster: Kontext wächst kontrolliert, nicht unbegrenzt in jedem Schritt.
SYSTEM_PROMPT beschreibt nur das Intent-Format (tool/final) - das LLM entscheidet was zu tun ist, führt Tools aber nicht selbst aus.

`main.py` — vollständiger 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()

Was hier am wichtigsten ist (einfach erklärt)

run_react(...) steuert den Loop und Stop-Bedingungen; Business-Aktionen werden nur über ToolGateway ausgeführt.
validate_action(...) und gateway.call(...) innerhalb des Loops sind die Governance/Control Layer in jedem Schritt.
Die Trennung von decide_next_action(...) und gateway.call(...) ist das Kernprinzip agent ≠ executor: Der Agent liefert Intent, Tools werden nur über die Policy Boundary aufgerufen.

`requirements.txt`

TEXT

openai==2.21.0

Beispielausgabe

Die Reihenfolge der Tool-Calls kann zwischen Runs leicht variieren, aber stop_reason und Policy-Gates (Allowlist, Budget, Validation) bleiben stabil.

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 ist das Ausführungsprotokoll der Schritte: Für jeden step werden action (was der Agent entschieden hat) und observation (was das Tool zurückgegeben hat) gespeichert.

args_hash ist ein Hash der Argumente. Daher kann er sich für dieselbe user_id über verschiedene Tools hinweg wiederholen; der Loop-Guard prüft die Kombination tool + args_hash.

Warum das ReAct ist und nicht nur Tool Calling

	Einmaliger Aufruf	ReAct loop
Entscheidung nach jeder Observation	❌	✅
Explizite Stop-Reasons	❌	✅
Kontrolle wiederholter identischer Aktionen	❌	✅
Run-Budget (steps/tools/time)	teilweise	✅

Typische `stop_reason`-Werte

success — der Agent hat eine finale Antwort geliefert
max_steps — Step-Budget ausgeschöpft
max_tool_calls — Tool-Call-Limit ausgeschöpft
max_seconds — Zeitbudget überschritten
llm_timeout — LLM hat nicht innerhalb von OPENAI_TIMEOUT_SECONDS geantwortet
loop_detected — derselbe Tool-Call mit denselben Args wurde wiederholt
tool_denied:<name> — Tool ist nicht in der Allowlist
invalid_action:* — Modell hat eine ungültige Action-Struktur zurückgegeben

Was hier NICHT gezeigt wird

Keine Auth/PII- und Production-Zugriffskontrollen für personenbezogene Daten.
Keine Retry/Backoff-Strategien für LLM und Tool-Layer.
Keine Token-/Kostenbudgets (cost guardrails).
Tools sind hier deterministische Lern-Mocks, keine echten externen APIs.

Was als Nächstes probieren

Entferne search_policy aus ALLOWED_TOOLS und beobachte, wie sich stop_reason ändert.
Setze max_tool_calls=1 und prüfe, dass die Policy den Agenten stoppt, nicht das Modell.
Ändere GOAL auf user_id=7 (Max) und prüfe die finale Antwort.
Probiere einen Aufruf eines nicht existierenden Tools (das Modell macht das manchmal) - dann siehst du tool_missing:*.
Füge einen Soft-Loop-Modus hinzu: normalisiere String-Args (trim + collapse spaces) vor dem Hashing, um semantisch gleiche Wiederholungen zu erkennen.
Füge JSONL-Step-Logs (trace) für Production-Observability hinzu.

Vollständiger Code auf GitHub

Im Repository liegt die vollständige runnable-Version dieses Beispiels: ReAct loop, policy boundary, allowlist, budgets, loop detection und stop reasons.

Vollständigen Code auf GitHub ansehen ↗

Task Decomposition Agent — Python (vollständige Implementierung mit LLM)

Zurück

Stoppbedingungen - Python

Task Decomposition Agent — Python (vollständige Implementierung mit LLM)

⏱️ 12 Min. Lesezeit • Aktualisiert Mär, 2026Schwierigkeit: ★★☆

Von Patterns genutzt

Erforderliche Governance

Integriert: Production ControlOnceOnly

Guardrails für Tool-Calling-Agents

Shippe dieses Pattern mit Governance:

Budgets (Steps / Spend Caps)
Tool-Permissions (Allowlist / Blocklist)
Kill switch & Incident Stop
Idempotenz & Dedupe
Audit logs & Nachvollziehbarkeit

OnceOnly testen Doku & Beispiele

Integrierter Hinweis: OnceOnly ist eine Control-Layer für Production-Agent-Systeme.

Autor

Diese Dokumentation wird von Engineers kuratiert und gepflegt, die AI-Agenten in der Produktion betreiben.

Die Inhalte sind KI-gestützt, mit menschlicher redaktioneller Verantwortung für Genauigkeit, Klarheit und Produktionsrelevanz.

Patterns und Empfehlungen basieren auf Post-Mortems, Failure-Modes und operativen Incidents in produktiven Systemen, auch bei der Entwicklung und dem Betrieb von Governance-Infrastruktur für Agenten bei OnceOnly.

ReAct Agent — Python (vollständige Implementierung mit LLM)

Kern des Musters (Kurz)

Was dieses Beispiel zeigt

Architektur

Projektstruktur

Ausführen

Aufgabe

Lösung

Code

tools.py — Tools (Quelle der Fakten)

Was hier am wichtigsten ist (einfach erklärt)

gateway.py — Policy Boundary (wichtigste Schicht)

Was hier am wichtigsten ist (einfach erklärt)

llm.py — decision step (Think)

Was hier am wichtigsten ist (einfach erklärt)

main.py — vollständiger ReAct-Loop

Was hier am wichtigsten ist (einfach erklärt)

requirements.txt

Beispielausgabe

Warum das ReAct ist und nicht nur Tool Calling

Typische stop_reason-Werte

Was hier NICHT gezeigt wird

Was als Nächstes probieren

Vollständiger Code auf GitHub

Von Patterns genutzt

Verwandte Failures

Erforderliche Governance

`tools.py` — Tools (Quelle der Fakten)

`gateway.py` — Policy Boundary (wichtigste Schicht)

`llm.py` — decision step (Think)

`main.py` — vollständiger ReAct-Loop

`requirements.txt`

Typische `stop_reason`-Werte