Ceci est l implementation pedagogique complete de l exemple de l article Quand un agent doit s arreter (et qui le decide).
Si vous n avez pas encore lu l article, commencez par la. Ici, le focus est uniquement sur le code : comment le systeme verifie les stop conditions apres chaque etape et termine de maniere controlee.
Ce Que Cet Exemple Demontre
- Comment l agent avance par etapes, alors que la decision d arret n est pas prise par le modele mais par la policy au niveau runtime
- Comment fonctionnent les stop conditions de base :
goal_reached,step_limit,too_many_errors,no_progress - Quelles conditions se declenchent dans cette demo :
goal_reached,too_many_errors,step_limit - Pourquoi l agent doit renvoyer
stop_reason, meme si la tache n est pas terminee - Comment trois situations differentes donnent une fin differente : succes vs arret d urgence
Structure du Projet
foundations/
└── stop-conditions/
└── python/
├── main.py # lance les scenarios et affiche le resume
├── agent.py # boucle agent + verification des stop conditions
├── llm.py # couche simple de decision : next action
├── tools.py # tools pedagogiques et echecs controles
└── requirements.txt
Comment Executer
1. Clone le depot et va dans le dossier :
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/stop-conditions/python
2. Installe les dependances (cet exemple n a pas de packages externes) :
pip install -r requirements.txt
3. Lance la demo :
python main.py
Ce Que Nous Construisons Dans Le Code
Nous construisons une boucle d agent simple avec un circuit d arret separe.
- le modele choisit l action suivante
- un tool s execute et modifie l etat
- runtime compte les metriques (
steps,errors,no_progress) - apres chaque etape, policy verifie les stop conditions
- si une condition se declenche, la boucle se termine avec
stop_reasonexplicite
Idee cle : le modele ne decide pas quand "c est suffisant". C est le systeme qui decide.
Code
tools.py - tools pedagogiques avec echec controle
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 - choix pedagogique simple de l action suivante
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 - boucle agent avec stop conditions policy-driven
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 - trois scenarios avec des fins differentes
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
# No external dependencies for this learning example.
Exemple de Sortie
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": [{...}]}
Note : dans la sortie,
historyest volontairement affiche en forme courte -"history": [{...}].
Critere de correction de l exemple : l agent termine toujours la boucle avec unstop_reasonexplicite et ne tourne pas a l infini.
Ce Que L on Voit En Pratique
| SCENARIO 1 | SCENARIO 2 | SCENARIO 3 | |
|---|---|---|---|
| Fin | goal_reached | too_many_errors | step_limit |
| Tache terminee avec succes | ✅ | ❌ | ❌ |
| Boucle limitee par policy | ✅ | ✅ | ✅ |
| Il y a une explication explicite de l arret | ✅ | ✅ | ✅ |
Ce Qu Il Faut Changer Dans Cet Exemple
- Ajoute
max_duration_secet un arret par wall-clock timeout - Ajoute un budget separe pour les tool-calls (
max_tool_calls) - Ajoute
stop_reason_detailspour journaliser la cause plus precisement - Ajoute un quatrieme scenario ou
no_progressse declenche
Code Complet sur GitHub
Le depot contient la version complete de cette demo : boucle agent, policy stop conditions et finalisation controlee.
Voir le code complet sur GitHub ↗