feat(examples/python): add self-critique-agent runnable example

Author: AgentPatterns

examples/agent-patterns/self-critique-agent/python/llm.py

@@ -98,6 +98,24 @@ class LLMInvalid(Exception):
 - Do not output markdown or extra keys.
 """.strip()
 
+REVISE_SYSTEM_PROMPT_STRICT = """
+You are an editor applying a constrained rewrite.
+Return exactly one JSON object:
+{
+  "revised_answer": "updated answer"
+}
+
+Rules:
+- Apply required_changes only.
+- Keep original scope and customer intent.
+- Do not add new facts or numbers.
+- Keep the answer concise and actionable.
+- You MUST satisfy each required_changes item exactly.
+- For ADD/MUST_INCLUDE: include the quoted phrase verbatim.
+- For REMOVE/MUST_REMOVE: ensure the quoted phrase does not appear.
+- Do not output markdown or extra keys.
+""".strip()
+
 
 
 def _get_client() -> OpenAI:
@@ -196,14 +214,16 @@ def revise_once(
     incident_context: dict[str, Any],
     draft: str,
     required_changes: list[str],
+    strict_mode: bool = False,
 ) -> str:
     payload = {
         "goal": goal,
         "incident_context": incident_context,
         "draft": draft,
         "required_changes": required_changes,
     }
-    data = _chat_json(system_prompt=REVISE_SYSTEM_PROMPT, payload=payload)
+    system_prompt = REVISE_SYSTEM_PROMPT_STRICT if strict_mode else REVISE_SYSTEM_PROMPT
+    data = _chat_json(system_prompt=system_prompt, payload=payload)
 
     revised = data.get("revised_answer")
     if not isinstance(revised, str):

examples/agent-patterns/self-critique-agent/python/main.py

@@ -172,30 +172,44 @@ def stopped(stop_reason: str, *, phase: str, **extra: Any) -> dict[str, Any]:
     revised = False
 
     if critique["decision"] == "revise":
-        if (time.monotonic() - started) > BUDGET.max_seconds:
-            return stopped("max_seconds", phase="revise")
-
-        try:
-            revised_raw = revise_once(
-                goal=goal,
-                incident_context=incident_context,
-                draft=draft,
-                required_changes=critique["required_changes"],
-            )
-            revised_payload = gateway.validate_revision(
-                original=draft,
-                revised=revised_raw,
-                context=incident_context,
-                required_changes=critique["required_changes"],
-            )
-        except LLMTimeout:
-            return stopped("llm_timeout", phase="revise")
-        except LLMInvalid as exc:
-            return stopped(exc.args[0], phase="revise")
-        except LLMEmpty:
-            return stopped("llm_empty", phase="revise")
-        except StopRun as exc:
-            return stopped(exc.reason, phase="revise")
+        revise_attempts = 0
+        revise_retried = False
+        revised_payload: dict[str, Any] | None = None
+        for attempt in range(1, 3):
+            if (time.monotonic() - started) > BUDGET.max_seconds:
+                return stopped("max_seconds", phase="revise")
+
+            revise_attempts = attempt
+            strict_mode = attempt > 1
+            try:
+                revised_raw = revise_once(
+                    goal=goal,
+                    incident_context=incident_context,
+                    draft=draft,
+                    required_changes=critique["required_changes"],
+                    strict_mode=strict_mode,
+                )
+                revised_payload = gateway.validate_revision(
+                    original=draft,
+                    revised=revised_raw,
+                    context=incident_context,
+                    required_changes=critique["required_changes"],
+                )
+                break
+            except LLMTimeout:
+                return stopped("llm_timeout", phase="revise")
+            except LLMInvalid as exc:
+                return stopped(exc.args[0], phase="revise")
+            except LLMEmpty:
+                return stopped("llm_empty", phase="revise")
+            except StopRun as exc:
+                if exc.reason == "patch_violation:required_changes_not_applied" and attempt < 2:
+                    revise_retried = True
+                    continue
+                return stopped(exc.reason, phase="revise")
+
+        if revised_payload is None:
+            return stopped("patch_violation:required_changes_not_applied", phase="revise")
 
         final_answer = revised_payload["answer"]
         revised = True
@@ -209,6 +223,8 @@ def stopped(stop_reason: str, *, phase: str, **extra: Any) -> dict[str, Any]:
                 "required_changes_total": revised_payload["required_changes_total"],
                 "required_changes_enforced": revised_payload["required_changes_enforced"],
                 "required_changes_unenforced": revised_payload["required_changes_unenforced"],
+                "attempts_used": revise_attempts,
+                "retried": revise_retried,
                 "revised_hash": text_hash(final_answer),
                 "ok": True,
             }