Guardrails Skill

You are an expert in LLM safety and output validation, implementing the guardrail

patterns used by Anthropic's Constitutional AI, Meta's Llama Guard, and enterprise

AI safety teams. You write production-ready guardrail systems in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — Input Guardrails

import re
from dataclasses import dataclass
from enum import Enum


class GuardrailAction(Enum):
    ALLOW = "allow"
    BLOCK = "block"
    REDACT = "redact"
    WARN = "warn"


@dataclass
class GuardrailResult:
    action: GuardrailAction
    reason: str
    modified_input: str | None = None


# ── Topic relevance check ─────────────────────────────────────────────────────
IMI_ALLOWED_TOPICS = [
    "sports", "fan", "football", "rugby", "cricket", "tennis", "golf",
    "sponsorship", "brand", "marketing", "research", "survey", "data",
    "audience", "engagement", "stadium", "club", "athlete", "team",
    "analysis", "report", "insight", "segment", "behaviour", "sentiment"
]

IMI_BLOCKED_TOPICS = [
    "gambling odds", "betting tips", "personal financial advice",
    "medical diagnosis", "legal advice", "political campaigning",
    "competitor pricing", "insider information"
]


def check_topic_relevance(user_input: str, threshold: float = 0.1) -> GuardrailResult:
    """Check if input is relevant to IMI research domain."""
    input_lower = user_input.lower()

    # Check blocked topics
    for blocked in IMI_BLOCKED_TOPICS:
        if blocked in input_lower:
            return GuardrailResult(
                action=GuardrailAction.BLOCK,
                reason=f"Input contains blocked topic: '{blocked}'"
            )

    # Check relevance
    matches = sum(1 for topic in IMI_ALLOWED_TOPICS if topic in input_lower)
    relevance = matches / len(IMI_ALLOWED_TOPICS)

    if relevance < threshold and len(user_input) > 50:
        return GuardrailResult(
            action=GuardrailAction.WARN,
            reason="Input may be outside IMI research scope"
        )

    return GuardrailResult(action=GuardrailAction.ALLOW, reason="Input passes topic check")


# ── Input length and complexity checks ───────────────────────────────────────
def check_input_constraints(
    user_input: str,
    max_length: int = 5000,
    min_length: int = 3
) -> GuardrailResult:
    """Enforce input length and basic quality constraints."""
    if len(user_input) < min_length:
        return GuardrailResult(action=GuardrailAction.BLOCK, reason="Input too short")
    if len(user_input) > max_length:
        truncated = user_input[:max_length]
        return GuardrailResult(
            action=GuardrailAction.REDACT,
            reason=f"Input truncated to {max_length} chars",
            modified_input=truncated
        )
    return GuardrailResult(action=GuardrailAction.ALLOW, reason="Length OK")


# ── Full input pipeline ───────────────────────────────────────────────────────
def apply_input_guardrails(user_input: str) -> tuple[str | None, list[GuardrailResult]]:
    """
    Run all input guardrails.
    Returns (processed_input, results_list) — None input means blocked.
    """
    results = []
    current_input = user_input

    # Import from security-pii skill
    try:
        from presidio_analyzer import AnalyzerEngine
        from presidio_anonymizer import AnonymizerEngine
        analyzer = AnalyzerEngine()
        anonymizer = AnonymizerEngine()
        pii_results = analyzer.analyze(text=current_input, language="en")
        if pii_results:
            anonymized = anonymizer.anonymize(text=current_input, analyzer_results=pii_results)
            current_input = anonymized.text
            results.append(GuardrailResult(action=GuardrailAction.REDACT, reason="PII redacted"))
    except ImportError:
        pass  # Presidio not installed — skip PII check

    checks = [
        check_input_constraints(current_input),
        check_topic_relevance(current_input),
    ]

    for check in checks:
        results.append(check)
        if check.action == GuardrailAction.BLOCK:
            return None, results
        if check.action == GuardrailAction.REDACT and check.modified_input:
            current_input = check.modified_input

    return current_input, results

Part 2 — Output Guardrails

import anthropic
import json, re


def llm_output_safety_check(
    output: str,
    context: str = "",
    model: str = "claude-3-haiku-20240307"
) -> dict:
    """
    LLM-as-judge safety check on output.
    Checks: factual grounding, harmful content, off-topic, hallucination risk.
    Fast and cheap using Haiku.
    """
    client = anthropic.Anthropic()
    response = client.messages.create(
        model=model,
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": (
                f"Evaluate this AI output for safety and quality.\n\n"
                f"CONTEXT PROVIDED: {context[:500]}\n\n"
                f"AI OUTPUT: {output[:1000]}\n\n"
                "Check for:\n"
                "1. Hallucination: does output contain claims not supported by context?\n"
                "2. Harmful content: offensive, discriminatory, or dangerous information?\n"
                "3. Off-topic: unrelated to sports fan research/sponsorship?\n"
                "4. Overconfidence: presents uncertain claims as fact?\n\n"
                'Reply with JSON only: {"safe": true/false, "issues": [], "confidence": 0-1}'
            )
        }]
    )

    text = response.content[0].text
    match = re.search(r'\{.*\}', text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group())
        except json.JSONDecodeError:
            pass

    return {"safe": True, "issues": [], "confidence": 0.5}


def check_factual_grounding(output: str, source_context: str) -> dict:
    """
    Check if output is grounded in provided sources.
    Detects hallucinations in RAG pipelines.
    """
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": (
                f"SOURCE DOCUMENTS:\n{source_context[:1000]}\n\n"
                f"GENERATED ANSWER:\n{output[:500]}\n\n"
                "Is every factual claim in the answer supported by the source documents? "
                'Reply JSON only: {"grounded": true/false, "unsupported_claims": []}'
            )
        }]
    )

    text = response.content[0].text
    match = re.search(r'\{.*\}', text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group())
        except json.JSONDecodeError:
            pass
    return {"grounded": True, "unsupported_claims": []}


# ── Rule-based output checks ──────────────────────────────────────────────────
HARMFUL_OUTPUT_PATTERNS = [
    r'\b(kill|murder|harm|attack)\b.*\b(person|people|fan)\b',
    r'\b(discriminat|racist|sexist)\b',
    r'\bpersonal (?:address|phone|email)\b.*\b\d',
]

OVERCONFIDENCE_PHRASES = [
    "definitely", "certainly", "100%", "guaranteed", "without doubt",
    "always true", "never wrong", "proven fact"
]

IMI_COMPETITOR_MENTIONS = ["nielsen sports", "repucom", "sponsorpulse", "octagon", "two circles"]


def rule_based_output_check(output: str) -> dict:
    """Fast, deterministic rule checks on output."""
    issues = []
    output_lower = output.lower()

    for pattern in HARMFUL_OUTPUT_PATTERNS:
        if re.search(pattern, output_lower):
            issues.append(f"Potentially harmful pattern: {pattern}")

    overconfident = [p for p in OVERCONFIDENCE_PHRASES if p in output_lower]
    if overconfident:
        issues.append(f"Overconfidence language: {', '.join(overconfident)}")

    competitor_mentions = [c for c in IMI_COMPETITOR_MENTIONS if c in output_lower]
    if competitor_mentions:
        issues.append(f"Competitor mention: {', '.join(competitor_mentions)}")

    return {"passed": len(issues) == 0, "issues": issues}

Part 3 — NeMo-Style Rail Configuration

# pip install nemoguardrails

# colang/imi_rails.co — topic and safety rails
NEMO_RAILS_CONFIG = """
define user ask about sports research
  "Tell me about football fan behaviour"
  "What drives sponsorship engagement?"
  "Analyse this survey data"

define user ask off-topic
  "What are the best gambling sites?"
  "Give me financial investment advice"
  "Help me write a political speech"

define flow off-topic check
  user ask off-topic
  bot refuse off-topic

define bot refuse off-topic
  "I'm focused on sports fan intelligence research for IMI. I'm not able to help with that topic."

define flow
  user ask about sports research
  bot provide research assistance
"""

# rails_config.yml
NEMO_CONFIG_YAML = """
models:
  - type: main
    engine: anthropic
    model: claude-3-5-sonnet-20241022

rails:
  input:
    flows:
      - off-topic check
  output:
    flows:
      - output moderation
"""


def build_nemo_guardrails():
    """Initialise NeMo Guardrails from config."""
    from nemoguardrails import RailsConfig, LLMRails
    config = RailsConfig.from_content(
        colang_content=NEMO_RAILS_CONFIG,
        yaml_content=NEMO_CONFIG_YAML
    )
    return LLMRails(config)

Part 4 — Full Guardrailed Pipeline

def guardrailed_imi_query(
    user_input: str,
    context: str = "",
    use_llm_check: bool = True
) -> dict:
    """
    Full guardrailed query pipeline:
    input check → LLM call → output check → return safe response
    """
    # 1. Input guardrails
    processed_input, input_results = apply_input_guardrails(user_input)
    if processed_input is None:
        block_reason = next((r.reason for r in input_results if r.action == GuardrailAction.BLOCK), "Blocked")
        return {"response": None, "blocked": True, "reason": block_reason}

    # 2. LLM call
    import anthropic
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=800,
        system="You are an IMI sports fan intelligence research assistant. Use British English.",
        messages=[{"role": "user", "content": processed_input}]
    )
    output = response.content[0].text

    # 3. Output guardrails
    rule_check = rule_based_output_check(output)
    if not rule_check["passed"]:
        return {"response": None, "blocked": True, "reason": f"Output blocked: {rule_check['issues']}"}

    if use_llm_check:
        safety = llm_output_safety_check(output, context)
        if not safety.get("safe", True):
            return {"response": None, "blocked": True, "reason": f"Safety check failed: {safety.get('issues')}"}

    return {"response": output, "blocked": False, "input_results": input_results}

Output Standards

• Layer order: input rails → LLM call → output rails (always both ends)
• Fast checks first: rule-based checks before expensive LLM-as-judge
• Fail safe: when in doubt, block and explain rather than pass
• Transparency: always return reason for blocking to aid debugging
• IMI policy: block gambling, betting, financial advice, competitor mentions in outputs
• British English in all guardrail messages and explanations

pip install

pip install anthropic nemoguardrails presidio-analyzer presidio-anonymizer

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)