Spaces:
Sleeping
Sleeping
| """ | |
| 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) | |
| # --------------------------------------------------------------------------- | |
| class SourceEvidence: | |
| source_id: str | |
| title: str | |
| stance: str # supporting | contradicting | neutral | |
| reliability: float | |
| key_snippet: str = "" | |
| 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) | |
| 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 | |
| def contradicting_sources(self) -> List[SourceEvidence]: | |
| return [s for s in self.sources_used if s.stance == "contradicting"] | |
| def supporting_sources(self) -> List[SourceEvidence]: | |
| return [s for s in self.sources_used if s.stance == "supporting"] | |
| def high_reliability_sources(self) -> List[SourceEvidence]: | |
| return [s for s in self.sources_used if s.reliability >= 0.80] | |
| def false_claims(self) -> List[ClaimAssessment]: | |
| return [c for c in self.claims if c.verification_status in ("false", "partially_true")] | |
| 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), | |
| } | |
| 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) | |
| 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) | |
| 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" | |
| 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" | |
| 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" | |
| ) | |
| 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, | |
| ) | |