""" explainability.py ================= Structured reasoning objects and template-based explanation generation. No API calls — fully deterministic. The explanation is built from the agent's actual reasoning trace (claims verified, sources checked, decision) so it aligns with every moderation decision and is auditable. Enterprise platforms (Meta, Instagram, etc.) can: - Replace `generate_explanation` with a call to their internal LLM - Feed `ReasoningObject` into dashboards / audit logs - Extend `DECISION_TEMPLATES` without touching engine code """ from __future__ import annotations import os import textwrap from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import yaml EXPLANATION_CONFIG_PATH = os.getenv("EXPLANATION_CONFIG", "explanation_templates.yaml") # --------------------------------------------------------------------------- # Structured reasoning object (what the agent built up during the episode) # --------------------------------------------------------------------------- @dataclass class SourceEvidence: source_id: str title: str stance: str # supporting | contradicting | neutral reliability: float key_snippet: str = "" @dataclass class ClaimAssessment: claim_id: str claim_text: str verification_status: str # true | false | partially_true | unknown confidence: float supporting_sources: List[str] = field(default_factory=list) contradicting_sources: List[str] = field(default_factory=list) @dataclass class ReasoningObject: """ Complete structured reasoning trace for one episode. Produced by the agent or inference script; consumed by the explainer. """ content_summary: str claims: List[ClaimAssessment] sources_used: List[SourceEvidence] decision: str # allow | flag | remove confidence: float # 0–1 step_count: int episode_id: str = "" difficulty: str = "" extra: Dict[str, Any] = field(default_factory=dict) # Derived helpers @property def contradicting_sources(self) -> List[SourceEvidence]: return [s for s in self.sources_used if s.stance == "contradicting"] @property def supporting_sources(self) -> List[SourceEvidence]: return [s for s in self.sources_used if s.stance == "supporting"] @property def high_reliability_sources(self) -> List[SourceEvidence]: return [s for s in self.sources_used if s.reliability >= 0.80] @property def false_claims(self) -> List[ClaimAssessment]: return [c for c in self.claims if c.verification_status in ("false", "partially_true")] @property def verified_fraction(self) -> float: if not self.claims: return 0.0 verified = sum( 1 for c in self.claims if c.verification_status != "unknown" ) return verified / len(self.claims) # --------------------------------------------------------------------------- # Template engine # --------------------------------------------------------------------------- # Built-in templates — can be fully overridden by explanation_templates.yaml _BUILTIN_TEMPLATES: Dict[str, str] = { "remove": textwrap.dedent("""\ MODERATION DECISION: REMOVE Content summary: {content_summary} After examining {n_claims} claim(s) across {n_sources} source(s): {claim_assessments} KEY EVIDENCE {evidence_list} REASONING {reasoning_narrative} CONCLUSION The content was removed because {removal_reason}. Confidence: {confidence_pct}%. Sources consulted: {n_sources}. """), "flag": textwrap.dedent("""\ MODERATION DECISION: FLAG FOR REVIEW Content summary: {content_summary} After examining {n_claims} claim(s) across {n_sources} source(s): {claim_assessments} KEY EVIDENCE {evidence_list} REASONING {reasoning_narrative} CONCLUSION The content was flagged because {flag_reason}. Confidence: {confidence_pct}%. Sources consulted: {n_sources}. """), "allow": textwrap.dedent("""\ MODERATION DECISION: ALLOW Content summary: {content_summary} After examining {n_claims} claim(s) across {n_sources} source(s): {claim_assessments} KEY EVIDENCE {evidence_list} REASONING {reasoning_narrative} CONCLUSION The content was allowed because {allow_reason}. Confidence: {confidence_pct}%. Sources consulted: {n_sources}. """), } class ExplanationTemplates: """Loads templates from YAML; falls back to built-ins.""" def __init__(self, path: str = EXPLANATION_CONFIG_PATH) -> None: self._templates = dict(_BUILTIN_TEMPLATES) if os.path.exists(path): with open(path, "r", encoding="utf-8") as fh: loaded = yaml.safe_load(fh) or {} self._templates.update(loaded.get("templates", {})) def get(self, decision: str) -> str: return self._templates.get(decision, self._templates["flag"]) # --------------------------------------------------------------------------- # Explainability engine # --------------------------------------------------------------------------- class ExplainabilityEngine: """ Converts a ReasoningObject into a structured, human-readable explanation. No ML or API calls. All text is derived deterministically from the agent's actual evidence trace. """ def __init__(self, templates: Optional[ExplanationTemplates] = None) -> None: self._tpl = templates or ExplanationTemplates() def generate_explanation(self, reasoning: ReasoningObject) -> str: """Return a formatted explanation string.""" template = self._tpl.get(reasoning.decision) variables = self._build_variables(reasoning) try: return template.format(**variables) except KeyError as exc: # Graceful fallback — missing template var return self._fallback(reasoning) + f"\n[template key missing: {exc}]" def generate_structured(self, reasoning: ReasoningObject) -> dict: """Return a machine-readable structured explanation for API responses.""" return { "decision": reasoning.decision, "confidence": round(reasoning.confidence, 3), "content_summary": reasoning.content_summary, "claims_assessed": [ { "id": c.claim_id, "text": c.claim_text, "status": c.verification_status, "conf": round(c.confidence, 3), } for c in reasoning.claims ], "sources_used": [ { "id": s.source_id, "title": s.title, "stance": s.stance, "reliability": round(s.reliability, 3), } for s in reasoning.sources_used ], "evidence_summary": { "contradicting": len(reasoning.contradicting_sources), "supporting": len(reasoning.supporting_sources), "high_reliability": len(reasoning.high_reliability_sources), }, "narrative": self.generate_explanation(reasoning), "steps_taken": reasoning.step_count, "episode_id": reasoning.episode_id, } # ------------------------------------------------------------------ # Template variable builder # ------------------------------------------------------------------ def _build_variables(self, r: ReasoningObject) -> dict: return { "content_summary": r.content_summary or "(no summary)", "n_claims": len(r.claims), "n_sources": len(r.sources_used), "confidence_pct": int(r.confidence * 100), "claim_assessments": self._format_claims(r.claims), "evidence_list": self._format_evidence(r.sources_used), "reasoning_narrative":self._narrative(r), "removal_reason": self._removal_reason(r), "flag_reason": self._flag_reason(r), "allow_reason": self._allow_reason(r), } @staticmethod def _format_claims(claims: List[ClaimAssessment]) -> str: if not claims: return " (no claims assessed)" lines = [] for c in claims: status_icon = { "true": "[TRUE]", "false": "[FALSE]", "partially_true": "[PARTIAL]", "unknown": "[UNKNOWN]", "unverified": "[NOT VERIFIED]", }.get(c.verification_status, "[?]") lines.append( f" {status_icon} Claim: {c.claim_text}\n" f" Status: {c.verification_status} " f"(confidence {int(c.confidence * 100)}%)" ) return "\n".join(lines) @staticmethod def _format_evidence(sources: List[SourceEvidence]) -> str: if not sources: return " (no sources checked)" lines = [] for s in sorted(sources, key=lambda x: -x.reliability): icon = {"contradicting": "✗", "supporting": "✓", "neutral": "·"}.get( s.stance, "·" ) rel_bar = "█" * int(s.reliability * 10) + "░" * (10 - int(s.reliability * 10)) snippet = (s.key_snippet[:100] + "…") if len(s.key_snippet) > 100 else s.key_snippet lines.append( f" {icon} [{s.stance.upper():<14}] rel={rel_bar} ({s.reliability:.2f})\n" f" {s.title}\n" + (f" \"{snippet}\"" if snippet else "") ) return "\n".join(lines) def _narrative(self, r: ReasoningObject) -> str: parts = [] n_false = len(r.false_claims) n_total = len(r.claims) parts.append( f"{n_false} of {n_total} claim(s) were found to be false or partially false." ) if r.contradicting_sources: top = max(r.contradicting_sources, key=lambda s: s.reliability) parts.append( f"The strongest contradicting source was '{top.title}' " f"(reliability {top.reliability:.2f}), which " f"directly contradicts the claims made in the content." ) if r.supporting_sources: top = max(r.supporting_sources, key=lambda s: s.reliability) parts.append( f"One supporting source was found: '{top.title}' " f"(reliability {top.reliability:.2f})." ) else: parts.append("No credible supporting sources were identified.") if r.high_reliability_sources: names = ", ".join( f"'{s.title[:40]}'" for s in r.high_reliability_sources[:3] ) parts.append( f"High-reliability sources consulted: {names}." ) parts.append( f"The agent took {r.step_count} reasoning step(s) " f"and verified {int(r.verified_fraction * 100)}% of claims before deciding." ) return " " + "\n ".join(parts) @staticmethod def _removal_reason(r: ReasoningObject) -> str: false_count = len(r.false_claims) hi_rel = len(r.high_reliability_sources) if false_count: claim_texts = "; ".join( f"'{c.claim_text[:50]}'" for c in r.false_claims[:2] ) return ( f"{false_count} claim(s) were definitively false according to " f"{hi_rel} high-reliability source(s), including: {claim_texts}" ) return "overwhelming contradicting evidence from authoritative sources was found" @staticmethod def _flag_reason(r: ReasoningObject) -> str: n_contra = len(r.contradicting_sources) n_support = len(r.supporting_sources) if n_contra > 0 and n_support > 0: return ( f"evidence was mixed ({n_contra} contradicting, {n_support} supporting), " "requiring human review to determine context and intent" ) if n_contra > 0: return ( f"{n_contra} contradicting source(s) were found but evidence is " "insufficient for removal; human review is recommended" ) return "the claim could not be fully verified with available sources" @staticmethod def _allow_reason(r: ReasoningObject) -> str: n_support = len(r.supporting_sources) hi_rel = len(r.high_reliability_sources) return ( f"{n_support} supporting source(s) (including {hi_rel} high-reliability) " "corroborate the claims and no significant contradicting evidence was found" ) @staticmethod def _fallback(r: ReasoningObject) -> str: return ( f"MODERATION DECISION: {r.decision.upper()}\n" f"Content: {r.content_summary}\n" f"Claims verified: {len(r.claims)}. Sources checked: {len(r.sources_used)}.\n" f"Decision confidence: {int(r.confidence * 100)}%." ) # --------------------------------------------------------------------------- # Factory helper used by inference.py and server.py # --------------------------------------------------------------------------- def build_reasoning_from_episode( obs, # Final Observation object decision: str, episode_id: str = "", difficulty: str = "", ) -> ReasoningObject: """ Convenience builder: converts a final Observation into a ReasoningObject. Compatible with any agent that used verify_claim + check_source + decide. """ # Collect checked sources from interaction history checked_source_ids = { r.action_payload.get("source_id") for r in obs.interaction_history if r.action_type == "check_source" } sources_used = [ SourceEvidence( source_id=s.id, title=s.title, stance=s.stance if isinstance(s.stance, str) else s.stance.value, reliability=s.reliability, key_snippet=s.text[:200], ) for s in obs.available_sources if s.id in checked_source_ids ] claims_assessed = [ ClaimAssessment( claim_id=c.id, claim_text=c.text, verification_status=( c.verification_status if isinstance(c.verification_status, str) else c.verification_status.value ), confidence=c.confidence, ) for c in obs.claims ] # Confidence = mean reliability of checked sources (or 0.5 fallback) confidence = ( sum(s.reliability for s in sources_used) / len(sources_used) if sources_used else 0.5 ) return ReasoningObject( content_summary=obs.content[:200], claims=claims_assessed, sources_used=sources_used, decision=decision, confidence=confidence, step_count=obs.step_count, episode_id=episode_id, difficulty=difficulty, )