src/layers/context_policy.py

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
from pydantic import BaseModel
from typing import Optional
import numpy as np
import time

INTENT_PROFILES = {
    "general": [],           # No restrictions, always passes

    "coding_assistant": [
        "write code", "debug function", "explain algorithm",
        "code review", "fix bug", "refactor code",
        "explain error", "unit test", "implement feature",
        "programming language", "software architecture",
        "API design", "database query", "deployment"
    ],

    "recipe_bot": [
        "recipe", "ingredient", "cooking", "bake", "meal",
        "food preparation", "kitchen", "dish", "cuisine",
        "nutrition", "cooking technique", "flavor"
    ],

    "customer_support": [
        "order status", "refund", "shipping", "account help",
        "billing issue", "product return", "technical support",
        "subscription", "payment", "delivery", "complaint"
    ],

    "education": [
        "explain concept", "learn about", "study", "homework",
        "academic", "research", "history", "science", "math",
        "literature", "quiz", "exam preparation"
    ],

    "hr_assistant": [
        "leave request", "employee policy", "onboarding",
        "performance review", "payroll", "benefits",
        "job description", "hiring", "resignation"
    ],
}

class PolicyResult(BaseModel):
    triggered: bool
    app_context: str
    similarity_to_intent: float
    reason: str             # "out_of_scope" | "within_scope" | "no_policy"
    score: float
    latency_ms: float
    ran: bool = True

class ContextAwarePolicyLayer:
    # Out-of-scope threshold: if max similarity to ANY intent
    # example is below this, the prompt is out of scope.
    # Tune per app_context if needed.
    OUT_OF_SCOPE_THRESHOLD = 0.35

    def __init__(self, model):
        # Injected shared SentenceTransformer instance
        self.model = model
        self._profile_embeddings = {}
        self._build_profile_embeddings()

    def _build_profile_embeddings(self):
        """
        Pre-compute and cache embeddings for all intent profiles
        at startup. These never change at runtime.
        ~10ms one-time cost. Zero cost per request.
        """
        for profile, examples in INTENT_PROFILES.items():
            if examples:
                self._profile_embeddings[profile] = \
                    self.model.encode(examples, normalize_embeddings=True)

    def check(
        self,
        prompt: str,
        app_context: str = "general"
    ) -> PolicyResult:
        start = time.perf_counter()

        # general = no policy, always pass
        if app_context == "general" or \
           app_context not in self._profile_embeddings:
            latency = (time.perf_counter() - start) * 1000
            return PolicyResult(
                triggered=False,
                app_context=app_context,
                similarity_to_intent=1.0,
                reason="no_policy",
                score=0.0,
                latency_ms=round(latency, 3),
                ran=True
            )

        prompt_emb   = self.model.encode([prompt], normalize_embeddings=True)
        profile_embs = self._profile_embeddings[app_context]

        similarities = cosine_similarity(prompt_emb, profile_embs)[0]
        max_sim      = float(similarities.max())

        triggered = max_sim < self.OUT_OF_SCOPE_THRESHOLD
        latency   = (time.perf_counter() - start) * 1000

        return PolicyResult(
            triggered=triggered,
            app_context=app_context,
            similarity_to_intent=round(max_sim, 4),
            reason="out_of_scope" if triggered else "within_scope",
            score=round(1.0 - max_sim, 4) if triggered else 0.0,
            latency_ms=round(latency, 3),
            ran=True
        )

    def register_custom_profile(
        self,
        profile_name: str,
        examples: list[str]
    ):
        """
        Allows developers to register a custom intent profile
        via the API at runtime. Embeddings computed once on
        registration, cached for all subsequent requests.
        """
        self._profile_embeddings[profile_name] = \
            self.model.encode(examples, normalize_embeddings=True)
        INTENT_PROFILES[profile_name] = examples