feat: add optional Semantic Guard plugin for variable-level alignment…

pievo/group/evolveflow.py

@@ -68,6 +68,7 @@ def __init__(
         initial_absolute_sigma: float = 0.01,
         llm_judge: bool = False,
         off_pievo: bool = False,
+        guard: Optional[Any] = None,
     ):
         self.submissions: List[Dict[str, Any]] = []
         self.output_dir = output_dir
@@ -156,6 +157,7 @@ def __init__(
 
         self.hypothesis_form = {}
         self.off_pievo = off_pievo
+        self.guard = guard  # Optional SemanticGuard for variable-level alignment checks
         self.extra_clients: Dict[str, OpenAIChatCompletionClient] = {}
 
     def register_client(self, client: Any, name: str):
@@ -851,6 +853,50 @@ async def update_pievo_state(self) -> None:
 
         # Extract the new principle from the self.submissions, and assign the initial belief for handling.
         new_principle_ids = self._extract_principles()
+
+        # ── Semantic Guard: validate new principles before Bayesian update ──
+        if self.guard is not None and new_principle_ids:
+            from pievo.guard import Principle as GuardPrinciple
+            phenomenon = self.task  # Use task description as phenomenon context
+            validated_ids = []
+            blocked_ids = []
+            for pid in new_principle_ids:
+                ptext = self.principles.get(pid, "")
+                new_p = GuardPrinciple(id=pid, text=ptext, phenomenon=phenomenon)
+                decisions = await self.guard.check_against_all(
+                    new_p, self.principles, phenomenon
+                )
+                if self.guard.any_blocked(decisions):
+                    blocked_ids.append(pid)
+                    blocked_reasons = [
+                        d.reason for d in decisions if d.action == "BLOCK"
+                    ]
+                    pievo_log(
+                        f"SEMANTIC GUARD: BLOCKED principle '{pid}' — "
+                        f"semantic conflict with existing principles. "
+                        f"Reasons: {blocked_reasons}",
+                        source="semantic_guard",
+                        tag="blocked",
+                        level="WARN",
+                    )
+                    logger.warning(
+                        f"SEMANTIC GUARD: BLOCKED principle '{pid}' — {blocked_reasons}"
+                    )
+                else:
+                    validated_ids.append(pid)
+                    pievo_log(
+                        f"SEMANTIC GUARD: ACCEPTED principle '{pid}'",
+                        source="semantic_guard",
+                        tag="proceed",
+                    )
+
+            if blocked_ids:
+                # Remove blocked principles from self.principles so they do not
+                # participate in the Bayesian update.
+                for pid in blocked_ids:
+                    self.principles.pop(pid, None)
+                new_principle_ids = validated_ids
+
         self._handle_new_principles(new_principle_ids)
         self._extract_history()
 

pievo/guard/__init__.py

@@ -0,0 +1,50 @@
+"""
+Semantic Guard for PiEvo -- Optional plugin for variable-level semantic alignment
+detection between LLM-generated principles before Bayesian comparison.
+
+Usage:
+    from pievo.guard import SemanticGuard, StrictPolicy, LLMExtractor
+    guard = SemanticGuard(extractor=LLMExtractor(client=openai_client), policy=StrictPolicy())
+    pievo = PiEvo(task=..., client=..., guard=guard)
+"""
+
+from pievo.guard.signature import (
+    OperationalDefinition,
+    VariableSignature,
+    EvidentialStatus,
+    Principle,
+    PerVariableAlignment,
+    AlignmentReport,
+    NO_SHARED_VARIABLES,
+)
+from pievo.guard.alignment import (
+    SimilarityMethod,
+    text_similarity,
+    variable_alignment,
+    compute_alignment_vector,
+)
+from pievo.guard.policy import (
+    GuardPolicy,
+    StrictPolicy,
+    PermissivePolicy,
+    ConservativePolicy,
+)
+from pievo.guard.extractor import (
+    SignatureExtractor,
+    LLMExtractor,
+    ManualExtractor,
+    ExtractionError,
+)
+from pievo.guard.guard import (
+    SemanticGuard,
+    AlignmentDecision,
+)
+
+__all__ = [
+    "OperationalDefinition", "VariableSignature", "EvidentialStatus",
+    "Principle", "PerVariableAlignment", "AlignmentReport", "NO_SHARED_VARIABLES",
+    "SimilarityMethod", "text_similarity", "variable_alignment", "compute_alignment_vector",
+    "GuardPolicy", "StrictPolicy", "PermissivePolicy", "ConservativePolicy",
+    "SignatureExtractor", "LLMExtractor", "ManualExtractor", "ExtractionError",
+    "SemanticGuard", "AlignmentDecision",
+]

pievo/guard/alignment.py

@@ -0,0 +1,192 @@
+"""
+Multi-dimensional alignment vector computation (Spec Section 3.3).
+"""
+
+from __future__ import annotations
+
+import logging
+from enum import Enum
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple
+
+from pievo.guard.signature import (
+    VariableSignature,
+    OperationalDefinition,
+    PerVariableAlignment,
+    AlignmentReport,
+    NO_SHARED_VARIABLES,
+    op_def_eq,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SimilarityMethod(Enum):
+    EXACT = "exact"
+    TOKEN = "token"
+    EMBEDDING = "embedding"
+
+
+def _tokenize(text: str) -> Set[str]:
+    return set(text.lower().split())
+
+
+def text_similarity(
+    text1: str,
+    text2: str,
+    method: SimilarityMethod = SimilarityMethod.EXACT,
+    threshold: float = 0.7,
+    embedding_fn: Optional[Callable[[str], List[float]]] = None,
+) -> Tuple[float, str]:
+    """Compute similarity between two text strings."""
+    if method == SimilarityMethod.EXACT:
+        return (1.0 if text1 == text2 else 0.0), "exact"
+
+    elif method == SimilarityMethod.TOKEN:
+        tokens1 = _tokenize(text1)
+        tokens2 = _tokenize(text2)
+        if not tokens1 or not tokens2:
+            return 0.0, "token"
+        intersection = tokens1 & tokens2
+        union = tokens1 | tokens2
+        jaccard = len(intersection) / len(union)
+        return jaccard if jaccard >= threshold else 0.0, "token"
+
+    elif method == SimilarityMethod.EMBEDDING:
+        if embedding_fn is None:
+            logger.warning("EMBEDDING mode requires embedding_fn, falling back to TOKEN")
+            return text_similarity(text1, text2, SimilarityMethod.TOKEN, threshold)
+        vec1 = embedding_fn(text1)
+        vec2 = embedding_fn(text2)
+        if not vec1 or not vec2:
+            return 0.0, "embedding"
+        dot = sum(a * b for a, b in zip(vec1, vec2))
+        norm1 = sum(a * a for a in vec1) ** 0.5
+        norm2 = sum(b * b for b in vec2) ** 0.5
+        if norm1 == 0 or norm2 == 0:
+            return 0.0, "embedding"
+        cosine = max(0.0, min(1.0, dot / (norm1 * norm2)))
+        return cosine if cosine >= threshold else 0.0, "embedding"
+
+    raise ValueError(f"Unknown similarity method: {method}")
+
+
+def variable_alignment(
+    s1: VariableSignature,
+    s2: VariableSignature,
+    similarity_method: SimilarityMethod = SimilarityMethod.EXACT,
+    similarity_threshold: float = 0.7,
+    embedding_fn: Optional[Callable[[str], List[float]]] = None,
+) -> Dict[str, Any]:
+    """Per-variable alignment with separate confidence reporting."""
+    if similarity_method == SimilarityMethod.EXACT:
+        ref_match = 1.0 if s1.referent == s2.referent else 0.0
+    else:
+        ref_match, _ = text_similarity(
+            s1.referent, s2.referent,
+            method=similarity_method,
+            threshold=similarity_threshold,
+            embedding_fn=embedding_fn,
+        )
+
+    op_match = 1.0 if op_def_eq(s1.operational_definition, s2.operational_definition) else 0.0
+    ev_match = 1.0 if s1.evidential_status == s2.evidential_status else 0.0
+    confidence = min(s1.confidence, s2.confidence)
+    uncertain = confidence < 0.8
+
+    return {
+        "referential": ref_match,
+        "operational": op_match,
+        "evidential": ev_match,
+        "confidence": confidence,
+        "uncertain": uncertain,
+        "referent_1": s1.referent,
+        "referent_2": s2.referent,
+        "op_1": s1.operational_definition.to_dict(),
+        "op_2": s2.operational_definition.to_dict(),
+        "ref_match_raw": ref_match,
+        "op_match_raw": op_match,
+        "ev_match_raw": ev_match,
+    }
+
+
+def compute_alignment_vector(
+    signatures_1: List[VariableSignature],
+    signatures_2: List[VariableSignature],
+    phenomenon: str,
+    similarity_method: SimilarityMethod = SimilarityMethod.EXACT,
+    similarity_threshold: float = 0.7,
+    embedding_fn: Optional[Callable[[str], List[float]]] = None,
+) -> AlignmentReport:
+    """Compute 3D alignment vector between two sets of variable signatures."""
+    sig1_by_name: Dict[str, VariableSignature] = {s.name: s for s in signatures_1}
+    sig2_by_name: Dict[str, VariableSignature] = {s.name: s for s in signatures_2}
+
+    shared_names = sorted(set(sig1_by_name.keys()) & set(sig2_by_name.keys()))
+
+    if not shared_names:
+        return AlignmentReport(
+            vector=(0.0, 0.0, 0.0),
+            per_variable=[],
+            shared_count=0,
+            phenomenon=phenomenon,
+            policy_recommendation=NO_SHARED_VARIABLES,
+            has_uncertain=False,
+        )
+
+    per_variable_results: List[PerVariableAlignment] = []
+    ref_scores: List[float] = []
+    op_scores: List[float] = []
+    ev_scores: List[float] = []
+    has_uncertain = False
+
+    for name in shared_names:
+        s1 = sig1_by_name[name]
+        s2 = sig2_by_name[name]
+
+        result = variable_alignment(
+            s1, s2,
+            similarity_method=similarity_method,
+            similarity_threshold=similarity_threshold,
+            embedding_fn=embedding_fn,
+        )
+
+        ref_scores.append(result["referential"])
+        op_scores.append(result["operational"])
+        ev_scores.append(result["evidential"])
+        if result["uncertain"]:
+            has_uncertain = True
+
+        pva = PerVariableAlignment(
+            name=name,
+            referential=result["referential"],
+            operational=result["operational"],
+            evidential=result["evidential"],
+            uncertain=result["uncertain"],
+            confidence=result["confidence"],
+            details={
+                "referent_1": result["referent_1"],
+                "referent_2": result["referent_2"],
+                "op_1": result["op_1"],
+                "op_2": result["op_2"],
+                "ref_match_raw": result["ref_match_raw"],
+                "op_match_raw": result["op_match_raw"],
+                "ev_match_raw": result["ev_match_raw"],
+                "extraction_confidence": result["confidence"],
+            },
+        )
+        per_variable_results.append(pva)
+
+    n = len(shared_names)
+    vector = (
+        sum(ref_scores) / n,
+        sum(op_scores) / n,
+        sum(ev_scores) / n,
+    )
+
+    return AlignmentReport(
+        vector=vector,
+        per_variable=per_variable_results,
+        shared_count=n,
+        phenomenon=phenomenon,
+        has_uncertain=has_uncertain,
+    )