Upload 3 files

eval.py

@@ -0,0 +1,206 @@
+"""Evaluation module for the RAG system using Ragas.
+This script provides tools to measure faithfulness, relevancy, and retrieval precision.
+
+How to run:
+    python eval.py <testset_csv_path>
+"""
+
+# pylint: disable=import-error,no-name-in-module,invalid-name,broad-except,missing-function-docstring,missing-class-docstring,wrong-import-order,ungrouped-imports,line-too-long,logging-fstring-interpolation,import-outside-toplevel
+
+import os
+import logging
+import pandas as pd
+from typing import List, Optional, Any
+from datasets import Dataset
+from ragas import evaluate
+from ragas.metrics.collections import (
+    faithfulness,
+    answer_relevancy,
+    context_precision,
+    context_recall,
+)
+
+try:
+    from langchain.chat_models import ChatOpenAI
+except Exception:
+    from langchain_openai import ChatOpenAI
+
+try:
+    from langchain_huggingface import HuggingFaceEmbeddings
+except Exception:
+    from langchain_community.embeddings import HuggingFaceEmbeddings
+
+
+def run_evaluation(
+    questions: List[str],
+    answers: List[str],
+    contexts: List[List[str]],
+    ground_truths: Optional[List[str]] = None,
+) -> Any:
+    """
+    Run Ragas evaluation on a set of QA results.
+
+    Parameters
+    ----------
+    questions : List[str]
+        List of user questions.
+    answers : List[str]
+        List of generated answers.
+    contexts : List[List[str]]
+        List of context strings retrieved for each question.
+    ground_truths : List[str], optional
+        Optional list of ground truth answers for recall metrics.
+
+    Returns
+    -------
+    Any
+        Ragas evaluation results containing metric scores.
+    """
+    data = {
+        "question": questions,
+        "answer": answers,
+        "contexts": contexts,
+    }
+    if ground_truths:
+        data["ground_truth"] = ground_truths
+
+    # Ragas evaluate works best with dataset objects
+    dataset = Dataset.from_dict(data)
+
+    # Use OpenRouter if key is available, else default to OpenAI
+    openrouter_key = os.getenv("OPENROUTER_API_KEY")
+    if openrouter_key:
+        # Use OpenRouter-compatible base and forward the key as the OpenAI key
+        os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1"
+        os.environ["OPENAI_API_KEY"] = openrouter_key
+
+    # Allow overriding the eval/model via env var; default to a compatible model
+    eval_model = os.getenv(
+        "OPENAI_MODEL", os.getenv("EVAL_MODEL", "openai/gpt-oss-120b")
+    )
+    logging.info("Using evaluation LLM model=%s", eval_model)
+    # Allow overriding how many generations ragas requests from the LLM.
+    # Some providers (or models) ignore multi-generation requests; default to 1 to avoid warnings.
+    try:
+        num_gens = int(os.getenv("RAGAS_NUM_GENERATIONS", "1"))
+    except Exception:
+        num_gens = 1
+    logging.info("Requesting %s generation(s) per prompt", num_gens)
+    try:
+        llm = ChatOpenAI(model=eval_model, n=num_gens)
+    except TypeError:
+        # Some ChatOpenAI wrappers do not accept `n` at construction; fall back to default.
+        llm = ChatOpenAI(model=eval_model)
+
+    # Use the same embeddings as the main app for consistency
+    embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-base-en-v1.5")
+
+    logging.info("Starting Ragas evaluation...")
+    result = evaluate(
+        dataset=dataset,
+        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
+        llm=llm,
+        embeddings=embeddings,
+    )
+    logging.info("Evaluation complete.")
+
+    return result
+
+
+def extract_scalar_metrics(result: Any) -> dict:
+    """Extract common scalar metrics (faithfulness, relevancy, precision, recall)
+    from a ragas evaluation result. Returns a dict of metric->float or empty dict.
+    """
+    keys_of_interest = {
+        "faithfulness",
+        "answer_relevancy",
+        "context_precision",
+        "context_recall",
+        "relevancy",
+        "precision",
+        "recall",
+    }
+
+    found: dict = {}
+
+    def is_number(x):
+        return isinstance(x, (int, float)) and not isinstance(x, bool)
+
+    def traverse(obj):
+        if isinstance(obj, dict):
+            for k, v in obj.items():
+                if (
+                    isinstance(k, str)
+                    and k.lower() in keys_of_interest
+                    and is_number(v)
+                ):
+                    found[k.lower()] = float(v)
+                traverse(v)
+        elif isinstance(obj, (list, tuple)):
+            for v in obj:
+                traverse(v)
+        else:
+            try:
+                if hasattr(obj, "__dict__"):
+                    traverse(vars(obj))
+            except Exception:
+                pass
+
+    try:
+        traverse(result)
+        # check common attrs
+        for attr in ("metrics", "results", "scores", "score"):
+            try:
+                val = getattr(result, attr, None)
+                if val is not None:
+                    traverse(val)
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+    return found
+
+
+def evaluate_from_csv(csv_path: str) -> Any:
+    """
+    Load a testset from CSV and run evaluation.
+
+    Parameters
+    ----------
+    csv_path : str
+        Path to the testset CSV.
+
+    Returns
+    -------
+    Any
+        Evaluation results.
+    """
+    df = pd.read_csv(csv_path)
+    # Ragas testset generation typically provides 'question', 'answer', 'contexts', 'ground_truth'
+    # 'contexts' is often stored as a string representation of a list in CSV
+    import ast
+
+    df["contexts"] = df["contexts"].apply(
+        lambda x: ast.literal_eval(x) if isinstance(x, str) else x
+    )
+
+    return run_evaluation(
+        questions=df["question"].tolist(),
+        answers=df["answer"].tolist(),
+        contexts=df["contexts"].tolist(),
+        ground_truths=(
+            df["ground_truth"].tolist() if "ground_truth" in df.columns else None
+        ),
+    )
+
+
+if __name__ == "__main__":
+    import sys
+
+    logging.basicConfig(level=logging.INFO)
+    if len(sys.argv) > 1:
+        res = evaluate_from_csv(sys.argv[1])
+        print(res)
+    else:
+        logging.info("Eval module ready. Pass a CSV file to evaluate.")

safety.py

@@ -0,0 +1,650 @@
+"""
+Safety and Guardrails module for the LLM application.
+This script provides runtime safety checks for toxicity and output length using Guardrails AI.
+
+How to run:
+    This module is intended to be imported and used within other scripts.
+"""
+
+# pylint: disable=import-error,no-name-in-module,invalid-name,broad-except,missing-function-docstring,missing-class-docstring,logging-fstring-interpolation,line-too-long
+# The safety module is intentionally procedural and contains many heuristics
+# and early-return branches; relax complexity checks for linting clarity.
+# pylint: disable=too-many-branches,too-many-statements,too-many-locals,too-many-return-statements
+
+import os
+import re
+import json
+import logging
+import unicodedata
+from typing import Tuple, Optional, List
+import requests
+
+try:
+    import openai
+
+    OPENAI_AVAILABLE = True
+except Exception:
+    openai = None
+    OPENAI_AVAILABLE = False
+
+try:
+    from guardrails import Guard
+    from guardrails.hub import ValidLength, ToxicLanguage
+
+    GUARDRAILS_AVAILABLE = True
+except Exception:
+    Guard = None
+    ValidLength = None
+    ToxicLanguage = None
+    GUARDRAILS_AVAILABLE = False
+
+try:
+    pass
+except Exception:
+    pass
+
+# Optional telemetry helper: prefer to import at module load so inner functions
+# can call `notify_low_score(...)` without importing repeatedly.
+try:
+    from telemetry import notify_low_score  # type: ignore
+except Exception:  # pragma: no cover - telemetry optional
+
+    def notify_low_score(*_args, **_kwargs):
+        return None
+
+
+def get_safety_guard() -> Optional[Guard]:
+    """
+    Initialize a safety guard with toxicity and length checks.
+
+    Returns
+    -------
+    Optional[Guard]
+        A Guardrails Guard object if initialization succeeds, else None.
+    """
+    if not GUARDRAILS_AVAILABLE or Guard is None:
+        logging.debug("Guardrails not available; using local safety fallbacks.")
+        return None
+
+    try:
+        # Use OpenRouter if key is available; forward key as OPENAI_API_KEY
+        openrouter_key = os.getenv("OPENROUTER_API_KEY")
+        if openrouter_key:
+            os.environ["OPENAI_API_BASE"] = "https://api.openrouter.ai/v1"
+            os.environ["OPENAI_API_KEY"] = openrouter_key
+
+        # Toxicity "fix" requires an LLM to rephrase.
+        # Length "refuse" returns an empty string or refusal if it fails.
+        guard = Guard().use_many(
+            ToxicLanguage(threshold=0.5, on_fail="fix"),
+            ValidLength(min=10, max=2000, on_fail="refuse"),
+        )
+        return guard
+    except Exception as e:
+        logging.debug(f"Failed to initialize Guardrails: {e}")
+        return None
+
+
+def _call_safeguard_model(text: str) -> Optional[dict]:
+    """
+    Optionally call an external safeguard model (e.g. OpenRouter's gpt-oss-safeguard-20b)
+    using the OpenAI-compatible chat completions endpoint.
+
+    Returns a dict with keys: {'passed': bool, 'score': float, 'reason': str}
+    or None if the call couldn't be made or parsed.
+    """
+    model = os.getenv(
+        "SAFEGUARD_MODEL",
+        os.getenv("OPENROUTER_SAFEGUARD_MODEL", "openai/gpt-oss-safeguard-20b"),
+    )
+    if not model:
+        return None
+    api_key = os.getenv("OPENAI_API_KEY")
+    api_base = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")
+    if not api_key:
+        logging.debug(
+            "SAFEGUARD_MODEL configured but OPENAI_API_KEY missing; skipping safeguard call"
+        )
+        return None
+
+    system_prompt = "You are a safety classifier. Given the user input, determine whether it is safe to send to a general-purpose assistant. Respond ONLY with a JSON object with the keys: 'passed' (true|false), 'score' (0-100), 'reason' (short string)."
+    user_prompt = f"Classify the following input for safety and prompt-injection risk:\n\n'''{text}'''"
+
+    url = api_base.rstrip("/") + "/chat/completions"
+    payload = {
+        "model": model,
+        "messages": [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_prompt},
+        ],
+        "temperature": 0.0,
+        "max_tokens": 256,
+    }
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    try:
+        resp = requests.post(url, headers=headers, json=payload, timeout=8)
+        if resp.status_code != 200:
+            logging.debug(
+                "Safeguard model call failed status=%s text=%s",
+                resp.status_code,
+                resp.text,
+            )
+            return None
+        j = resp.json()
+        # openai-compatible response: choices[0].message.content
+        choices = j.get("choices") or []
+        if not choices:
+            return None
+        content = choices[0].get("message", {}).get("content") or choices[0].get("text")
+        if not content:
+            return None
+        # Attempt to parse JSON from content
+        try:
+            parsed = json.loads(content.strip())
+            # normalize keys
+            return {
+                "passed": bool(parsed.get("passed")),
+                "score": float(parsed.get("score", 0.0)),
+                "reason": str(parsed.get("reason", "")),
+            }
+        except Exception:
+            # try to extract JSON blob from text
+            m = re.search(r"\{.*\}", content, flags=re.S)
+            if not m:
+                return None
+            try:
+                parsed = json.loads(m.group(0))
+                return {
+                    "passed": bool(parsed.get("passed")),
+                    "score": float(parsed.get("score", 0.0)),
+                    "reason": str(parsed.get("reason", "")),
+                }
+            except Exception:
+                return None
+    except Exception as e:  # pragma: no cover - network call
+        logging.debug("Safeguard model call exception: %s", e)
+        return None
+
+
+def check_safety(
+    text: str, include_meta: bool = False
+) -> Tuple[str, Optional[bool], float, Optional[dict]]:
+    """
+    Check a piece of text against safety rails.
+
+    Parameters
+    ----------
+    text : str
+        The text to be validated.
+
+    Returns
+    -------
+    Tuple[str, Optional[bool], float]
+        A tuple containing the validated (and potentially fixed) output,
+        a boolean indicating if the validation passed, and a numeric
+        safety score in range 0.0-100.0 (higher is safer).
+    """
+    # Normalize input: Unicode NFKC and strip zero-width/control chars
+    if isinstance(text, str):
+        normalized = unicodedata.normalize("NFKC", text)
+        # remove zero-width and control characters
+        normalized = re.sub(r"[\u200B-\u200F\uFEFF\x00-\x1F\x7F]", "", normalized)
+        # collapse whitespace
+        normalized = re.sub(r"\s+", " ", normalized).strip()
+    else:
+        normalized = ""
+
+    # Quick whitelist for statistical analysis queries (reduce false positives)
+    try:
+        stats_patterns = [
+            r"\b(f[\s-]?test|f-test|analysis of variance|anova)\b",
+            r"\b(t[\s-]?test|t-test|paired t-test|two[-\s]sample t-test)\b",
+            r"\bchi[-\s]?square\b",
+            r"\bp[-\s]?value\b",
+            r"\bdegrees of freedom\b",
+            r"\bpower analysis\b",
+            r"\bsample size\b",
+            r"\b(cohen'?s d)\b",
+            r"\b(regression|linear regression|logistic regression)\b",
+        ]
+        # clinical / device domain keywords often used in safe, non-procedural queries
+        clinical_patterns = [
+            r"\b(intrauterine device|iud|copper iud|copper intrauterine)\b",
+            r"\b(device|implant|biocompatibility|adverse event|adverse-event|safety data|clinical trial|pilot study)\b",
+            r"\b(manufactur|steriliz|quality control|q[mc]|risk assessment)\b",
+            # general lab/facility terms and simple read-only queries about labs
+            r"\b(lab|labs|laborator(?:y|ies)|facility|good laboratory practice|good lab practice)\b",
+            r"\b(what|which|list)\s+(labs|laborator(?:y|ies)|facilities?)\b",
+        ]
+        # common read-only query patterns (methods, devices, findings, results)
+        readonly_patterns = [
+            r"\b(what|which|list|describe|mention(?:ed)?)\b.*\b(methods?|methodology|procedures?)\b",
+            r"\b(what|which|list|describe|mention(?:ed)?)\b.*\b(devices?|implants?|instruments?)\b",
+            r"\b(what|which|list|describe|mention(?:ed)?)\b.*\b(results?|findings?|conclusions?)\b",
+            r"\b(what|which|list|describe)\b.*\b(software|packages|tools|libraries)\b",
+        ]
+        readonly_hits = [
+            p for p in readonly_patterns if re.search(p, normalized, flags=re.I)
+        ]
+        # quick-sensitive keywords that should prevent whitelisting
+        sensitive_quick = re.search(
+            r"(rm\s+-rf|exfiltrate|api[_-]?key|password|secret|ssh\b|curl\b|open a shell|execute command|run shell|sudo)",
+            normalized,
+            flags=re.I,
+        )
+        stats_hits = [p for p in stats_patterns if re.search(p, normalized, flags=re.I)]
+        clinical_hits = [
+            p for p in clinical_patterns if re.search(p, normalized, flags=re.I)
+        ]
+
+        # If the prompt is clearly statistical or clinical (and contains no obvious sensitive ops),
+        # treat as safe (high score). This avoids false positives for legitimate research queries
+        # such as statistical analysis or device safety discussions.
+        if (stats_hits or clinical_hits) and not sensitive_quick:
+            meta = {"whitelist": []}
+            if stats_hits:
+                meta["whitelist"].append("statistics")
+                meta["matches_stats"] = stats_hits
+            if clinical_hits:
+                meta["whitelist"].append("clinical")
+                meta["matches_clinical"] = clinical_hits
+            if readonly_hits:
+                if "whitelist" not in meta:
+                    meta["whitelist"] = []
+                meta["whitelist"].append("readonly")
+                meta["matches_readonly"] = readonly_hits
+            # If a safeguard model is configured, consult it and honor its score
+            try:
+                sg = _call_safeguard_model(normalized)
+            except Exception:
+                sg = None
+
+            # If safeguard model explicitly refuses, honor that decision
+            if sg and not sg.get("passed", True):
+                try:
+                    notify_low_score(
+                        {
+                            "kind": "prompt",
+                            "reason": "safeguard_model_whitelist_refuse",
+                            "score": float(sg.get("score", 0.0)),
+                            "meta": meta,
+                        }
+                    )
+                except Exception:
+                    pass
+                if include_meta:
+                    return (
+                        "refuse",
+                        False,
+                        float(sg.get("score", 0.0) or 0.0),
+                        {"safeguard": sg, "whitelist": meta.get("whitelist")},
+                    )
+                return "refuse", False, float(sg.get("score", 0.0) or 0.0)
+
+            # baseline score: prefer safeguard score if present, otherwise 95.0
+            baseline = float(sg.get("score", 95.0)) if sg else 95.0
+            score = min(95.0, baseline)
+            try:
+                notify_low_score(
+                    {
+                        "kind": "prompt",
+                        "reason": "whitelist_applied",
+                        "score": float(score),
+                        "meta": meta,
+                    }
+                )
+            except Exception:
+                pass
+            if include_meta:
+                return normalized, True, float(score), meta
+            return normalized, True, float(score)
+    except Exception:
+        # non-fatal; fall back to normal heuristics
+        pass
+
+    guard = get_safety_guard()
+    if guard:
+        try:
+            # Pass the text to the guard. If "fix" is triggered, it uses internal logic.
+            validation_result = guard.validate(normalized)
+            # When guardrails is available, map pass/fail to a simple score
+            score = 95.0 if validation_result.validation_passed else 20.0
+            if include_meta:
+                return (
+                    validation_result.validated_output,
+                    validation_result.validation_passed,
+                    score,
+                    {"guardrails": True},
+                )
+            return (
+                validation_result.validated_output,
+                validation_result.validation_passed,
+                score,
+            )
+        except Exception as e:
+            logging.error(f"Safety check failed during validation: {e}")
+            if include_meta:
+                return text, False, 0.0, None
+            return text, False, 0.0
+
+    # Local fallback checks when Guardrails isn't available
+    # Optional: consult a dedicated safeguard model if configured
+    try:
+        safeguard_enabled = os.getenv("SAFEGUARD_ENABLED", "true").lower() in (
+            "1",
+            "true",
+            "yes",
+        )
+    except Exception:
+        safeguard_enabled = True
+    if safeguard_enabled:
+        try:
+            sg = _call_safeguard_model(normalized)
+            if sg:
+                # If the safeguard model explicitly refuses, treat as unsafe
+                try:
+                    if not sg.get("passed", True) or float(
+                        sg.get("score", 0.0)
+                    ) < float(os.getenv("SAFETY_SCORE_THRESHOLD", "40.0")):
+                        notify_low_score(
+                            {
+                                "kind": "prompt",
+                                "reason": "safeguard_model",
+                                "score": float(sg.get("score", 0.0)),
+                            }
+                        )
+                except Exception:
+                    pass
+                if not sg.get("passed", True):
+                    if include_meta:
+                        return (
+                            "refuse",
+                            False,
+                            float(sg.get("score", 0.0) or 0.0),
+                            {"safeguard": sg},
+                        )
+                    return "refuse", False, float(sg.get("score", 0.0) or 0.0)
+                # If passed, we can use the model's score as a baseline
+                base_score = float(sg.get("score", 100.0))
+                # continue to heuristics but start with model baseline
+            else:
+                base_score = None
+        except Exception:
+            base_score = None
+    else:
+        base_score = None
+    # Short benign message whitelist (greetings, thanks, simple queries)
+    try:
+        greeting_pattern = r"^(hi|hello|hey|hi there|hello there|thanks|thank you|bye|goodbye|good morning|good afternoon|good evening|how are you)[\.!?]?$"
+    except Exception:
+        greeting_pattern = r"^(hi|hello|hey|thanks|thank you|bye|goodbye)[\.!?]?$"
+
+    if isinstance(normalized, str) and re.search(
+        greeting_pattern, normalized, flags=re.I
+    ):
+        if include_meta:
+            return normalized, True, 100.0, {"whitelist": "greeting"}
+        return normalized, True, 100.0
+
+    # Length checks (configurable via env vars SAFETY_MIN_LEN and SAFETY_MAX_LEN)
+    try:
+        min_len = int(os.getenv("SAFETY_MIN_LEN", "10"))
+    except Exception:
+        min_len = 10
+    try:
+        max_len = int(os.getenv("SAFETY_MAX_LEN", "2000"))
+    except Exception:
+        max_len = 2000
+    if not text or len(text) < min_len:
+        if include_meta:
+            return text, False, 10.0, None
+        return text, False, 10.0
+    if len(text) > max_len:
+        # truncated output, penalize score
+        if include_meta:
+            return text[:max_len], False, 30.0, None
+        return text[:max_len], False, 30.0
+
+    # Simple profanity filter (best-effort)
+    bad_words = {"fuck", "shit", "bitch", "asshole", "damn"}
+    lowered = normalized.lower()
+    found = False
+    for w in bad_words:
+        if w in lowered:
+            found = True
+            normalized = normalized.replace(w, "*" * len(w))
+
+    # Obfuscation detection heuristics
+    def looks_like_base64(s: str) -> bool:
+        # long base64-like strings (no spaces, length > 100, valid charset)
+        b64_re = re.compile(r"^[A-Za-z0-9+/=\\n\\r]{100,}$")
+        return bool(b64_re.search(s))
+
+    def looks_like_hex(s: str) -> bool:
+        # long hex strings
+        hex_re = re.compile(r"^(0x)?[0-9a-fA-F]{80,}$")
+        return bool(hex_re.search(s))
+
+    def looks_like_percent_encoding(s: str) -> bool:
+        return bool(re.search(r"%[0-9A-Fa-f]{2}", s)) and len(s) > 80
+
+    def long_punct_seq(s: str) -> bool:
+        return bool(re.search(r"[\W_]{20,}", s))
+
+    obf_hits = []
+    if looks_like_base64(text) or looks_like_base64(normalized):
+        obf_hits.append("base64")
+    if looks_like_hex(text) or looks_like_hex(normalized):
+        obf_hits.append("hex")
+    if looks_like_percent_encoding(text) or looks_like_percent_encoding(normalized):
+        obf_hits.append("percent-encoding")
+    if long_punct_seq(text) or long_punct_seq(normalized):
+        obf_hits.append("long-punct")
+    # Additional heuristic detectors
+
+    def detect_prompt_injection(s: str) -> List[str]:
+        patterns = [
+            r"ignore (all )?previous (instructions|prompts|messages)",
+            r"disregard (the )?above",
+            r"forget (your )?instructions",
+            r"ignore (this )?and do",
+            r"now do the following",
+            r"please break the rules",
+            r"ignore previous",
+            r"ignore all previous",
+            r"disobey",
+        ]
+        return [p for p in patterns if re.search(p, s, flags=re.I)]
+
+    def detect_jailbreak(s: str) -> List[str]:
+        patterns = [
+            r"jailbreak",
+            r"bypass (the )?policy",
+            r"break out",
+            r"override (the )?system",
+            r"sudo",
+            r"escape the sandbox",
+            r"remove restrictions",
+            r"disable safety",
+        ]
+        return [p for p in patterns if re.search(p, s, flags=re.I)]
+
+    def detect_prefix_attack(s: str) -> List[str]:
+        # look for system-like prefix instructions commonly used in prompt-injection
+        patterns = [
+            r"^system[:\-]",
+            r"^assistant[:\-]",
+            r"^user[:\-]",
+            r"^###",
+            r"^---",
+            r"^\[system\]",
+            r"^\[assistant\]",
+        ]
+        return [p for p in patterns if re.search(p, s, flags=re.I | re.M)]
+
+    def detect_sensitive_operations(s: str) -> List[str]:
+        patterns = [
+            r"rm\s+-rf",
+            r"delete (all|files|data)",
+            r"shutdown",
+            r"format (disk|drive)",
+            r"open a shell",
+            r"run shell",
+            r"execute command",
+            r"curl http",
+            r"ssh ",
+            r"api[_-]?key",
+            r"access token",
+            r"password",
+            r"secret",
+            r"exfiltrate",
+            r"send email",
+            r"transfer file",
+            r"write to /etc",
+            r"drop table",
+        ]
+        return [p for p in patterns if re.search(p, s, flags=re.I)]
+
+    injections = detect_prompt_injection(normalized)
+    jailbreaks = detect_jailbreak(normalized)
+    prefixes = detect_prefix_attack(normalized)
+    sensitive = detect_sensitive_operations(normalized)
+
+    # Moderation hook (optional): combine OpenAI moderation results when available
+    moderation_flag = None
+    if os.getenv("OPENAI_API_KEY") and OPENAI_AVAILABLE:
+        try:
+            # Use the moderation endpoint via openai if available. Access via
+            # getattr to avoid static attribute checks that pylint may flag.
+            mod_client = getattr(openai, "Moderation", None)
+            if mod_client is not None and hasattr(mod_client, "create"):
+                resp = mod_client.create(input=normalized)
+                # The response structure includes categories and a 'flagged' boolean
+                results = resp.get("results") or []
+                if results:
+                    mod = results[0]
+                    moderation_flag = mod.get("flagged", False)
+                    logging.debug("Moderation result: %s", json.dumps(mod))
+        except Exception as e:  # pragma: no cover - external call
+            logging.debug("Moderation call failed: %s", e)
+    elif os.getenv("OPENAI_API_KEY") and not OPENAI_AVAILABLE:
+        # If openai library not installed, attempt lightweight HTTP call
+        try:
+            url = "https://api.openai.com/v1/moderations"
+            headers = {
+                "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
+                "Content-Type": "application/json",
+            }
+            payload = {"input": normalized}
+            r = requests.post(url, headers=headers, json=payload, timeout=5)
+            if r.status_code == 200:
+                j = r.json()
+                results = j.get("results") or []
+                if results:
+                    mod = results[0]
+                    moderation_flag = mod.get("flagged", False)
+                    logging.debug("Moderation http result: %s", json.dumps(mod))
+        except Exception:  # pragma: no cover - external call
+            pass
+
+    logging.debug(
+        "Safety detectors: injections=%s jailbreaks=%s prefixes=%s sensitive=%s obf=%s moderation=%s",
+        injections,
+        jailbreaks,
+        prefixes,
+        sensitive,
+        obf_hits,
+        moderation_flag,
+    )
+
+    # If sensitive operations are requested, refuse immediately
+    if sensitive:
+        try:
+            notify_low_score(
+                {
+                    "kind": "prompt",
+                    "reason": "sensitive_operation",
+                    "detectors": sensitive,
+                    "score": 0.0,
+                }
+            )
+        except Exception:
+            pass
+        if include_meta:
+            return "refuse", False, 0.0, {"moderation_flag": True}
+        return "refuse", False, 0.0
+
+    # If moderation flagged the content, refuse
+    if moderation_flag:
+        try:
+            notify_low_score(
+                {
+                    "kind": "prompt",
+                    "reason": "moderation_flagged",
+                    "score": 0.0,
+                }
+            )
+        except Exception:
+            pass
+        if include_meta:
+            return "refuse", False, 0.0, {"sensitive": True}
+        return "refuse", False, 0.0
+
+    # Compute a simple heuristic score
+    score = 100.0
+    if found:
+        score -= 40.0
+    if obf_hits:
+        score -= 30.0
+    if jailbreaks:
+        score -= 45.0
+    if injections:
+        score -= 35.0
+    if prefixes:
+        score -= 20.0
+    if len(normalized) < 50:
+        score -= 10.0
+
+    # clamp
+    score = max(score, 0.0)
+
+    # If a safeguard model provided a baseline score, be conservative and apply it
+    if "base_score" in locals() and base_score is not None:
+        try:
+            score = min(float(base_score), float(score))
+        except Exception:
+            pass
+
+    passed_flag = (
+        (not found) and (not jailbreaks) and (not injections) and (not obf_hits)
+    )
+
+    # Telemetry on low scores
+    try:
+        threshold = float(os.getenv("SAFETY_SCORE_THRESHOLD", "40.0"))
+        if score < threshold:
+            notify_low_score(
+                {
+                    "kind": "prompt",
+                    "reason": "low_score",
+                    "score": float(score),
+                    "detectors": {
+                        "found": found,
+                        "obf_hits": obf_hits,
+                        "jailbreaks": jailbreaks,
+                        "injections": injections,
+                        "prefixes": prefixes,
+                    },
+                }
+            )
+    except Exception:
+        # do not fail safety on telemetry errors
+        pass
+
+    # Return metadata including any safeguard baseline if requested
+    meta: dict = {}
+    if "base_score" in locals() and base_score is not None:
+        meta["safeguard_base_score"] = float(base_score)
+    if include_meta:
+        return normalized, passed_flag, float(score), (meta if meta else None)
+    return normalized, passed_flag, float(score)

telemetry.py

@@ -0,0 +1,95 @@
+"""Telemetry helper for anonymized safety events.
+
+This module provides a minimal notifier that logs low-score / refused
+events and optionally POSTs anonymized metadata to a configured webhook.
+No secrets are sent.
+"""
+
+import os
+import json
+import logging
+import re
+from typing import Mapping, Any
+
+
+def notify_low_score(event: Mapping[str, Any]) -> None:
+    """Notify about a low-score or refused safety event.
+
+    event: mapping with non-sensitive metadata, e.g. {
+      "kind": "prompt" | "answer",
+      "score": 12.3,
+      "detectors": ["jailbreak", "base64"],
+      "session_id_present": False,
+    }
+    """
+    logger = logging.getLogger("telemetry")
+    try:
+        logger.info("Telemetry event: %s", json.dumps(event))
+    except Exception:
+        logger.info("Telemetry event: %s", str(event))
+
+    webhook = os.getenv("TELEMETRY_WEBHOOK")
+    if not webhook:
+        return
+
+    # Only POST refused events to the external telemetry webhook.
+    # Allowed refused reasons:
+    refused_reasons = {
+        "sensitive_operation",
+        "moderation_flagged",
+        "safeguard_model",
+        "response_refused",
+        "refused",
+    }
+
+    reason = str(event.get("reason", "")).lower() if event else ""
+    score = None
+    try:
+        score = float(event.get("score")) if event and "score" in event else None
+    except Exception:
+        score = None
+
+    # Post only when explicitly refused (reason in refused_reasons) or score==0.0
+    should_post = False
+    if reason and any(r in reason for r in refused_reasons):
+        should_post = True
+    if score is not None and score <= 0.0:
+        should_post = True
+
+    if not should_post:
+        logger.info(
+            "Telemetry event ignored (not a refused event): %s", json.dumps(event)
+        )
+        return
+
+    payload = {"event": event}
+    try:
+        import requests
+
+        headers = {"Content-Type": "application/json"}
+        # POST anonymized metadata only
+        requests.post(webhook, json=payload, headers=headers, timeout=2)
+    except Exception as e:  # pragma: no cover - network
+        logger.debug("Telemetry post failed: %s", e)
+
+
+def make_snippet(text: str, max_chars: int = 120) -> str:
+    """Create a short, single-line snippet suitable for logs and UI.
+
+    - strips control characters and collapses whitespace
+    - truncates with an ellipsis if longer than `max_chars`
+    """
+    if not text:
+        return ""
+    try:
+        s = str(text)
+    except Exception:
+        return ""
+    # replace control characters with a single space to avoid word concatenation
+    s = re.sub(r"[\u0000-\u001F\u007F]+", " ", s)
+    # collapse whitespace/newlines
+    s = re.sub(r"\s+", " ", s).strip()
+    if len(s) <= max_chars:
+        return s
+    # truncate safely
+    return s[: max_chars - 1].rstrip() + "…"