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

Author: AgentPatterns

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

@@ -16,7 +16,7 @@ def __init__(self, reason: str):
 
 @dataclass(frozen=True)
 class Budget:
-    max_seconds: int = 30
+    max_seconds: int = 120
     max_draft_chars: int = 900
     max_risks: int = 5
     max_required_changes: int = 5
@@ -184,11 +184,45 @@ def _is_enforceable_required_change(item: str) -> bool:
 
 
 def _contains_normalized_phrase(*, text: str, phrase: str) -> bool:
-    normalized_text = _normalize_space(text).lower()
-    normalized_phrase = _normalize_space(phrase).lower()
+    # Compare using token-like normalization so punctuation differences
+    # (e.g. trailing dots/commas) do not cause false negatives.
+    normalized_text = re.sub(r"[^a-z0-9% ]+", " ", _normalize_space(text).lower())
+    normalized_phrase = re.sub(r"[^a-z0-9% ]+", " ", _normalize_space(phrase).lower())
+    normalized_text = " ".join(normalized_text.split())
+    normalized_phrase = " ".join(normalized_phrase.split())
     return normalized_phrase in normalized_text
 
 
+def _remove_phrase_occurrences(*, text: str, phrase: str) -> str:
+    cleaned = text
+    normalized_phrase = _normalize_space(phrase).strip()
+    if not normalized_phrase:
+        return cleaned
+
+    variants = {normalized_phrase, normalized_phrase.rstrip(".!?")}
+    for variant in variants:
+        if not variant:
+            continue
+        cleaned = re.sub(re.escape(variant), "", cleaned, flags=re.IGNORECASE)
+
+    cleaned = re.sub(r"\s+\.", ".", cleaned)
+    cleaned = re.sub(r"[ \t]{2,}", " ", cleaned)
+    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
+    return cleaned.strip()
+
+
+def _append_phrase_sentence(*, text: str, phrase: str) -> str:
+    sentence = _normalize_space(phrase).strip()
+    if not sentence:
+        return text
+
+    out = text.rstrip()
+    if out and out[-1] not in ".!?":
+        out += "."
+    separator = "\n\n" if "\n\n" in out else " "
+    return (out + separator + sentence).strip()
+
+
 
 def validate_draft(draft: Any, *, max_chars: int) -> str:
     if not isinstance(draft, str) or not draft.strip():
@@ -308,6 +342,28 @@ def enforce_execution_decision(self, decision: str) -> None:
         if decision not in self.allow_execution_decisions:
             raise StopRun(f"critique_decision_denied_execution:{decision}")
 
+    def apply_required_changes_fallback(self, *, text: str, required_changes: list[str]) -> str:
+        """
+        Deterministic fallback for enforceable required changes:
+        remove MUST_REMOVE/REMOVE phrases and append missing MUST_INCLUDE/ADD phrases.
+        """
+        candidate = (text or "").strip()
+        if not candidate:
+            return candidate
+
+        phrase_rules = _extract_required_change_rules(required_changes)
+        must_include = phrase_rules["must_include"]
+        must_remove = phrase_rules["must_remove"]
+
+        for phrase in must_remove:
+            candidate = _remove_phrase_occurrences(text=candidate, phrase=phrase)
+
+        for phrase in must_include:
+            if not _contains_normalized_phrase(text=candidate, phrase=phrase):
+                candidate = _append_phrase_sentence(text=candidate, phrase=phrase)
+
+        return candidate.strip()
+
     def validate_revision(
         self,
         *,

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

@@ -17,7 +17,7 @@
 INCIDENT_CONTEXT = build_incident_context(report_date="2026-03-06", region="US")
 
 BUDGET = Budget(
-    max_seconds=30,
+    max_seconds=120,
     max_draft_chars=900,
     max_risks=5,
     max_required_changes=5,
@@ -175,7 +175,8 @@ def stopped(stop_reason: str, *, phase: str, **extra: Any) -> dict[str, Any]:
         revise_attempts = 0
         revise_retried = False
         revised_payload: dict[str, Any] | None = None
-        for attempt in range(1, 3):
+        last_revised_candidate = draft
+        for attempt in range(1, 4):
             if (time.monotonic() - started) > BUDGET.max_seconds:
                 return stopped("max_seconds", phase="revise")
 
@@ -189,6 +190,7 @@ def stopped(stop_reason: str, *, phase: str, **extra: Any) -> dict[str, Any]:
                     required_changes=critique["required_changes"],
                     strict_mode=strict_mode,
                 )
+                last_revised_candidate = revised_raw
                 revised_payload = gateway.validate_revision(
                     original=draft,
                     revised=revised_raw,
@@ -203,9 +205,27 @@ def stopped(stop_reason: str, *, phase: str, **extra: Any) -> dict[str, Any]:
             except LLMEmpty:
                 return stopped("llm_empty", phase="revise")
             except StopRun as exc:
-                if exc.reason == "patch_violation:required_changes_not_applied" and attempt < 2:
+                if exc.reason == "patch_violation:required_changes_not_applied" and attempt < 3:
                     revise_retried = True
                     continue
+                if exc.reason == "patch_violation:required_changes_not_applied":
+                    # Final fallback: enforce required phrase edits deterministically.
+                    try:
+                        fallback_revised = gateway.apply_required_changes_fallback(
+                            text=last_revised_candidate,
+                            required_changes=critique["required_changes"],
+                        )
+                        revised_payload = gateway.validate_revision(
+                            original=draft,
+                            revised=fallback_revised,
+                            context=incident_context,
+                            required_changes=critique["required_changes"],
+                        )
+                        revise_attempts = attempt + 1
+                        revise_retried = True
+                        break
+                    except StopRun as fallback_exc:
+                        return stopped(fallback_exc.reason, phase="revise")
                 return stopped(exc.reason, phase="revise")
 
         if revised_payload is None: