Dies ist die vollstandige Lernimplementierung des Beispiels aus dem Artikel Wann ein Agent stoppen muss (und wer das entscheidet).
Wenn du den Artikel noch nicht gelesen hast, beginne dort. Hier liegt der Fokus nur auf Code: wie das System nach jedem Schritt stop conditions pruft und kontrolliert beendet.
Was Dieses Beispiel Zeigt
- Wie der Agent in Schritten arbeitet, wahrend die Stop-Entscheidung nicht das Modell, sondern die runtime-policy trifft
- Wie die grundlegenden stop conditions funktionieren:
goal_reached,step_limit,too_many_errors,no_progress - Welche Bedingungen in dieser Demo auslosen:
goal_reached,too_many_errors,step_limit - Warum der Agent
stop_reasonzuruckgeben muss, auch wenn die Aufgabe nicht abgeschlossen ist - Wie drei unterschiedliche Situationen zu unterschiedlichen Enden fuhren: Erfolg vs Not-Stopp
Projektstruktur
TEXT
foundations/
└── stop-conditions/
└── python/
├── main.py # startet Szenarien und druckt Zusammenfassung
├── agent.py # Agentenzyklus + Stop-Condition-Prufung
├── llm.py # einfache Entscheidungsschicht: next action
├── tools.py # Lern-Tools und kontrollierte Fehler
└── requirements.txt
Ausfuhren
1. Repository klonen und in den Ordner wechseln:
BASH
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/stop-conditions/python
2. Abhangigkeiten installieren (dieses Beispiel hat keine externen Pakete):
BASH
pip install -r requirements.txt
3. Demo starten:
BASH
python main.py
Was Wir Im Code Bauen
Wir bauen einen einfachen Agentenzyklus mit einem separaten Stop-Kontur.
- das Modell wahlt die nachste Aktion
- ein Tool wird ausgefuhrt und andert den Zustand
- runtime zahlt Metriken (
steps,errors,no_progress) - nach jedem Schritt pruft policy die stop conditions
- wenn eine Bedingung greift, endet die Schleife mit explizitem
stop_reason
Kernidee: das Modell entscheidet nicht, wann "genug" ist. Das macht das System.
Code
tools.py - Lern-Tools mit kontrolliertem Fehler
PYTHON
from typing import Any
def make_initial_state(user_id: int, fail_fetch_times: int) -> dict[str, Any]:
return {
"user_id": user_id,
"fetch_calls": 0,
"fail_fetch_times": fail_fetch_times,
}
def fetch_orders(state: dict[str, Any]) -> dict[str, Any]:
state["fetch_calls"] += 1
if state["fetch_calls"] <= state["fail_fetch_times"]:
return {"ok": False, "error": "orders_api_timeout"}
orders = [
{"id": "ord-2001", "total": 49.9, "status": "paid"},
{"id": "ord-2002", "total": 19.0, "status": "shipped"},
]
state["orders"] = orders
return {"ok": True, "orders": orders}
def build_summary(state: dict[str, Any]) -> dict[str, Any]:
orders = state.get("orders")
if not orders:
return {"ok": False, "error": "missing_orders"}
summary = f"Prepared report for {len(orders)} recent orders."
state["summary"] = summary
return {"ok": True, "summary": summary}
llm.py - einfache lernorientierte Wahl der nachsten Aktion
PYTHON
from typing import Any
def choose_next_action(task: str, state: dict[str, Any]) -> dict[str, Any]:
# Learning version: fixed policy keeps behavior easy to reason about.
_ = task
if "orders" not in state:
return {"action": "fetch_orders", "parameters": {}}
return {"action": "build_summary", "parameters": {}}
agent.py - Agentenzyklus mit policy-driven stop conditions
PYTHON
from dataclasses import dataclass
from typing import Any
from llm import choose_next_action
from tools import build_summary, fetch_orders, make_initial_state
TOOLS = {
"fetch_orders": fetch_orders,
"build_summary": build_summary,
}
@dataclass
class StopPolicy:
max_steps: int
max_errors: int
max_no_progress: int
def evaluate_stop_conditions(
state: dict[str, Any],
steps: int,
errors: int,
no_progress: int,
policy: StopPolicy,
) -> str | None:
if "summary" in state:
return "goal_reached"
if steps >= policy.max_steps:
return "step_limit"
if errors >= policy.max_errors:
return "too_many_errors"
if no_progress >= policy.max_no_progress:
return "no_progress"
return None
def run_agent(task: str, user_id: int, fail_fetch_times: int, policy: StopPolicy) -> dict[str, Any]:
state = make_initial_state(user_id=user_id, fail_fetch_times=fail_fetch_times)
history: list[dict[str, Any]] = []
steps = 0
errors = 0
no_progress = 0
stop_reason: str | None = None
while True:
stop_reason = evaluate_stop_conditions(
state=state,
steps=steps,
errors=errors,
no_progress=no_progress,
policy=policy,
)
if stop_reason is not None:
break
steps += 1
call = choose_next_action(task, state)
action = call["action"]
history.append({"step": steps, "action": action, "status": "requested"})
tool = TOOLS.get(action)
if not tool:
errors += 1
no_progress += 1
state["last_error"] = f"unknown_action:{action}"
history.append({"step": steps, "action": action, "status": "error"})
else:
before_keys = set(state.keys())
result = tool(state)
if result.get("ok"):
after_keys = set(state.keys())
progress = len(after_keys - before_keys) > 0
no_progress = 0 if progress else no_progress + 1
state.pop("last_error", None)
history.append({"step": steps, "action": action, "status": "ok"})
else:
errors += 1
no_progress += 1
state["last_error"] = result.get("error", "unknown_error")
history.append({"step": steps, "action": action, "status": "error"})
return {
"done": stop_reason == "goal_reached",
"stop_reason": stop_reason,
"steps": steps,
"errors": errors,
"no_progress": no_progress,
"summary": state.get("summary"),
"history": history,
}
main.py - drei Szenarien mit unterschiedlichem Ende
PYTHON
import json
from agent import StopPolicy, run_agent
TASK = "Build weekly orders summary"
POLICY = StopPolicy(max_steps=6, max_errors=2, max_no_progress=3)
STEP_LIMIT_POLICY = StopPolicy(max_steps=1, max_errors=2, max_no_progress=3)
def compact_result(result: dict) -> str:
return (
"{"
f"\"done\": {str(bool(result.get('done'))).lower()}, "
f"\"stop_reason\": {json.dumps(result.get('stop_reason'), ensure_ascii=False)}, "
f"\"steps\": {int(result.get('steps', 0))}, "
f"\"errors\": {int(result.get('errors', 0))}, "
f"\"no_progress\": {int(result.get('no_progress', 0))}, "
f"\"summary\": {json.dumps(result.get('summary'), ensure_ascii=False)}, "
"\"history\": [{...}]"
"}"
)
def print_policy(policy: StopPolicy) -> None:
print(
"Policy:",
json.dumps(
{
"max_steps": policy.max_steps,
"max_errors": policy.max_errors,
"max_no_progress": policy.max_no_progress,
},
ensure_ascii=False,
),
)
def main() -> None:
print("=== SCENARIO 1: GOAL REACHED ===")
print_policy(POLICY)
result_ok = run_agent(
task=TASK,
user_id=42,
fail_fetch_times=1,
policy=POLICY,
)
print("Run result:", compact_result(result_ok))
print("\n=== SCENARIO 2: STOPPED BY ERROR LIMIT ===")
print_policy(POLICY)
result_stopped = run_agent(
task=TASK,
user_id=42,
fail_fetch_times=10,
policy=POLICY,
)
print("Run result:", compact_result(result_stopped))
print("\n=== SCENARIO 3: STOPPED BY STEP LIMIT ===")
print_policy(STEP_LIMIT_POLICY)
result_step_limit = run_agent(
task=TASK,
user_id=42,
fail_fetch_times=0,
policy=STEP_LIMIT_POLICY,
)
print("Run result:", compact_result(result_step_limit))
if __name__ == "__main__":
main()
requirements.txt
TEXT
# No external dependencies for this learning example.
Beispielausgabe
TEXT
python main.py
=== SCENARIO 1: GOAL REACHED ===
Policy: {"max_steps": 6, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": true, "stop_reason": "goal_reached", "steps": 3, "errors": 1, "no_progress": 0, "summary": "Prepared report for 2 recent orders.", "history": [{...}]}
=== SCENARIO 2: STOPPED BY ERROR LIMIT ===
Policy: {"max_steps": 6, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": false, "stop_reason": "too_many_errors", "steps": 2, "errors": 2, "no_progress": 2, "summary": null, "history": [{...}]}
=== SCENARIO 3: STOPPED BY STEP LIMIT ===
Policy: {"max_steps": 1, "max_errors": 2, "max_no_progress": 3}
Run result: {"done": false, "stop_reason": "step_limit", "steps": 1, "errors": 0, "no_progress": 0, "summary": null, "history": [{...}]}
Hinweis: in der Ausgabe wird
historyabsichtlich in Kurzform gezeigt -"history": [{...}].
Korrektheitskriterium des Beispiels: der Agent beendet die Schleife immer mit explizitemstop_reasonund lauft nicht endlos.
Was Man In Der Praxis Sieht
| SCENARIO 1 | SCENARIO 2 | SCENARIO 3 | |
|---|---|---|---|
| Ende | goal_reached | too_many_errors | step_limit |
| Aufgabe erfolgreich abgeschlossen | ✅ | ❌ | ❌ |
| Schleife ist durch policy begrenzt | ✅ | ✅ | ✅ |
| Es gibt eine explizite Begrundung fur den Stopp | ✅ | ✅ | ✅ |
Was Du In Diesem Beispiel Andern Kannst
- Fuge
max_duration_secund Stop per wall-clock timeout hinzu - Fuge ein separates Budget fur Tool-Calls hinzu (
max_tool_calls) - Fuge
stop_reason_detailshinzu, um den Grund genauer zu loggen - Fuge ein viertes Szenario hinzu, in dem
no_progressauslost
Vollstandiger Code auf GitHub
Im Repository liegt die vollstandige Version dieser Demo: Agentenzyklus, policy stop conditions und kontrollierter Abschluss.
Vollstandigen Code auf GitHub ansehen ↗