feat(guardrails): label-anchored PHI redaction, custom patterns, strict mode, phi preset

Author: ctrl-gaurav

effgen/core/agent.py

@@ -164,6 +164,9 @@ class AgentConfig:
     system_prompt: str = "You are a helpful AI assistant."
     max_iterations: int = 10
     temperature: float = 0.7
+    # Default output-token budget for every run(). None lets the model pick a
+    # size-aware default; run(max_tokens=...) overrides it for a single call.
+    max_tokens: int | None = None
     enable_sub_agents: bool = True
     enable_memory: bool = True
     enable_streaming: bool = False
@@ -915,6 +918,13 @@ def run(self,
             output_model: Pydantic BaseModel class — when provided, output is
                 validated and the parsed instance is stored in
                 ``response.metadata["parsed"]``.
+
+                Reasoning models and deeply nested schemas need a generous output
+                budget: the model spends tokens on internal reasoning and on the
+                JSON structure before it fills any values. Set ``max_tokens``
+                accordingly (``run(..., max_tokens=8192)`` for a reasoning model).
+                When an extraction validates but every field is empty,
+                ``response.metadata["structured_output_empty"]`` is set to True.
             inputs: Optional list of multimodal content parts created by
                 ``image_from``, ``audio_from``, or ``video_from``. When present,
                 the agent sends a structured Message directly to the model.
@@ -963,6 +973,10 @@ def run(self,
         # every run reports its own cost_usd + token counts (see _finalize_cost).
         self._run_cost_accum = {}
         debug = kwargs.pop("debug", False)
+        # A max_tokens set on the config is the default output budget for every
+        # run(); an explicit run(max_tokens=...) still overrides it per call.
+        if "max_tokens" not in kwargs and self.config.max_tokens is not None:
+            kwargs["max_tokens"] = self.config.max_tokens
         run_id = generate_run_id()
         # Capture checkpoint args here so the outer run() can use
         # them for the final-checkpoint write even after _run_single_agent
@@ -975,6 +989,7 @@ def run(self,
         prom_metrics.active_agents.inc(labels=labels)
 
         # Pre-run input guardrail check
+        input_redaction: dict[str, Any] | None = None
         if self._guardrail_chain is not None:
             from ..guardrails.base import GuardrailPosition
             gr = self._guardrail_chain.check(task, position=GuardrailPosition.INPUT)
@@ -988,6 +1003,13 @@ def run(self,
                 )
             if gr.modified_content is not None:
                 task = gr.modified_content
+                # Record what the input redaction removed so a run is auditable
+                # from its response (e.g. a note de-identified before a cloud call).
+                pii_types = gr.metadata.get("pii_types")
+                if pii_types:
+                    input_redaction = {"types": pii_types}
+                    if gr.metadata.get("pii_counts"):
+                        input_redaction["counts"] = gr.metadata["pii_counts"]
 
         # Resolve structured output schema. Accept either a JSON-Schema dict or
         # a Pydantic model class for `output_schema` (and the config default),
@@ -1080,6 +1102,8 @@ def run(self,
                 response.execution_trace = self.execution_tracker.get_trace()
                 response.execution_tree = self.execution_tracker.generate_execution_tree()
                 response.metadata["run_id"] = run_id
+                if input_redaction is not None:
+                    response.metadata["input_redaction"] = input_redaction
                 # Surface this run's cost + token usage on the result so callers
                 # can budget per call without a side channel.
                 self._finalize_cost_metadata(response)

effgen/core/agent_generation.py

@@ -97,6 +97,8 @@ def _apply_structured_output(
                         )
                     else:
                         response.metadata["parsed"] = parsed
+                    if self._structured_all_empty(parsed):
+                        response.metadata["structured_output_empty"] = True
                     return response
             except (json.JSONDecodeError, TypeError):
                 pass
@@ -137,6 +139,8 @@ def _apply_structured_output(
                 )
             else:
                 response.metadata["parsed"] = outcome.parsed
+            if self._structured_all_empty(outcome.parsed):
+                response.metadata["structured_output_empty"] = True
         else:
             # Keep the raw answer, mark the run unsuccessful, and explain why —
             # consistent with the framework's no-silent-failure rule.
@@ -157,6 +161,31 @@ def _apply_structured_output(
 
         return response
 
+    @staticmethod
+    def _structured_all_empty(data: Any) -> bool:
+        """Return True when a schema-valid object carries no extracted values.
+
+        A reasoning model or a nested schema under a tight ``max_tokens`` can
+        return an object that validates but has every field empty
+        (``{"diagnosis": "", "medications": []}``). Flagging it lets a caller
+        distinguish "nothing to extract" from "budget ran out before the model
+        filled the fields" — a larger ``max_tokens`` usually resolves the latter.
+        """
+        def _empty(v: Any) -> bool:
+            if v is None or v == "" or v == [] or v == {}:
+                return True
+            if isinstance(v, dict):
+                return all(_empty(x) for x in v.values())
+            if isinstance(v, list | tuple):
+                return all(_empty(x) for x in v)
+            return False
+
+        if isinstance(data, dict):
+            return len(data) > 0 and all(_empty(v) for v in data.values())
+        if isinstance(data, list | tuple):
+            return len(data) == 0
+        return False
+
     @staticmethod
     def _parse_with_pydantic(model_class: Any, data: Any) -> Any:
         """Parse data into a Pydantic model instance.

effgen/domains/presets.py

@@ -144,7 +144,16 @@ def FinanceDomain(**overrides) -> Domain:  # noqa: N802
 
 
 def HealthDomain(**overrides) -> Domain:  # noqa: N802
-    """Medical, wellness, and nutrition domain."""
+    """Medical, wellness, and nutrition domain.
+
+    This is a consumer health-information assistant. It uses the ``standard``
+    guardrails, which redact well-formed and labeled identifiers but do not
+    guarantee removal of every free-text identifier. It is not a de-identification
+    tool. To strip personal identifiers before sending records to a model, pass
+    ``guardrails="phi"`` (or ``guardrails=phi_guardrails(strict=True)`` to fail
+    closed), and extend ``PIIGuardrail`` with ``custom_patterns``/``custom_terms``
+    for site-specific identifiers.
+    """
     defaults = {
         "name": "health",
         "description": "Medical science, wellness, nutrition, and public health.",

effgen/guardrails/__init__.py

@@ -13,11 +13,13 @@
 from .presets import (
     MINIMAL,
     NONE,
+    PHI,
     STANDARD,
     STRICT,
     get_guardrail_preset,
     minimal_guardrails,
     no_guardrails,
+    phi_guardrails,
     standard_guardrails,
     strict_guardrails,
 )
@@ -43,11 +45,13 @@
     # Presets
     "STRICT",
     "STANDARD",
+    "PHI",
     "MINIMAL",
     "NONE",
     "get_guardrail_preset",
     "strict_guardrails",
     "standard_guardrails",
+    "phi_guardrails",
     "minimal_guardrails",
     "no_guardrails",
 ]

effgen/guardrails/base.py

@@ -119,6 +119,11 @@ def check(
         start = time.monotonic()
         current_content = content
         results: list[GuardrailResult] = []
+        # Aggregate redaction detail from any modifying guardrail (e.g.
+        # PIIGuardrail in redact mode) so a caller can audit what a chain removed
+        # without reaching into each individual result.
+        pii_types: list[str] = []
+        pii_counts: dict[str, int] = {}
 
         for guardrail in self.guardrails:
             if not guardrail.enabled:
@@ -140,18 +145,29 @@ def check(
                 result.metadata["chain_elapsed_ms"] = elapsed * 1000
                 return result
 
+            for t in result.metadata.get("pii_types", []):
+                if t not in pii_types:
+                    pii_types.append(t)
+            for t, n in (result.metadata.get("pii_counts") or {}).items():
+                pii_counts[t] = pii_counts.get(t, 0) + n
+
             # If guardrail modified content, pass modified version to next
             if result.modified_content is not None:
                 current_content = result.modified_content
 
         elapsed = time.monotonic() - start
+        metadata: dict[str, Any] = {
+            "checks_run": len(results),
+            "chain_elapsed_ms": elapsed * 1000,
+        }
+        if pii_types:
+            metadata["pii_types"] = pii_types
+        if pii_counts:
+            metadata["pii_counts"] = pii_counts
         return GuardrailResult(
             passed=True,
             reason="",
             modified_content=current_content if current_content != content else None,
             guardrail_name="GuardrailChain",
-            metadata={
-                "checks_run": len(results),
-                "chain_elapsed_ms": elapsed * 1000,
-            },
+            metadata=metadata,
         )