4. Guardrails e Métricas

"""The safety mesh of the assistant: checks before the model (input) and after it (output)."""
import re

INJECTION = re.compile(
    r"ignore (all |any )?(the )?(previous|prior|above) (instructions|rules)"
    r"|disregard (the )?(system|previous) (prompt|instructions)"
    r"|(reveal|print|show) (me )?(your|the) (system prompt|instructions)"
    r"|you are now|developer mode|jailbreak",
    re.IGNORECASE,
)
PII = [  # replaced BEFORE the text reaches the model, the logs or a third-party provider
    (re.compile(r"\b\d{3}\.?\d{3}\.?\d{3}-?\d{2}\b"), "[CPF]"),
    (re.compile(r"\b(?:\d[ -]?){13,16}\b"), "[CARD]"),
    (re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "[EMAIL]"),
]
CITATION = re.compile(r"\[(event:\d+)\]")


class Blocked(Exception):
    def __init__(self, kind: str, message: str):
        super().__init__(message)
        self.kind = kind


def check_input(text: str, max_chars: int) -> str:
    """Reject what must never reach the model; redact what must not leave the system."""
    if not text.strip():
        raise Blocked("empty", "Please ask a question.")
    if len(text) > max_chars:
        raise Blocked("too_long", f"Please keep the question under {max_chars} characters.")
    if INJECTION.search(text):
        raise Blocked("prompt_injection", "I can only help with BoxOffice events, tickets and orders.")
    for pattern, label in PII:
        text = pattern.sub(label, text)
    return text


DONT_KNOW = re.compile(r"(don't|do not) know|não sei|no information|não encontrei", re.IGNORECASE)


def check_output(answer: str, evidence: set[str]) -> tuple[str, bool]:
    """Grounded = cites at least one document it received and no document it did not.

    Invented citations are replaced by "[unverified]"; "I don't know" needs no citation.
    """
    cited = set(CITATION.findall(answer))
    invented = cited - evidence
    for doc_id in invented:
        answer = answer.replace(f"[{doc_id}]", "[unverified]")
    grounded = not invented and (bool(cited) or bool(DONT_KNOW.search(answer)))
    return answer, grounded

"""LLM metrics with OpenTelemetry, pushed by OTLP to otel-lgtm (lab 8).

In Prometheus they appear as assistant_llm_tokens_total, assistant_llm_cost_usd_total,
assistant_ttft_seconds_bucket, assistant_duration_seconds_bucket, assistant_tool_calls_total,
assistant_guardrail_blocks_total and assistant_requests_total.
"""
import os

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.metrics.view import ExplicitBucketHistogramAggregation, View
from opentelemetry.sdk.resources import Resource

# LLM latencies are seconds, not milliseconds: the default buckets (0, 5, 10, 25 ...) would be useless
SECONDS = ExplicitBucketHistogramAggregation([0.1, 0.25, 0.5, 1, 2, 4, 8, 16, 32, 64])

readers = []
if os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT"):  # no backend configured: metrics stay in memory
    from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter

    readers.append(PeriodicExportingMetricReader(OTLPMetricExporter()))

provider = MeterProvider(
    resource=Resource.create(),  # reads OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES
    metric_readers=readers,
    views=[View(instrument_name="assistant.ttft", aggregation=SECONDS),
           View(instrument_name="assistant.duration", aggregation=SECONDS)],
)
metrics.set_meter_provider(provider)
meter = metrics.get_meter("boxoffice.assistant")

requests = meter.create_counter("assistant.requests", description="questions answered, by outcome")
tokens = meter.create_counter("assistant.llm.tokens", description="LLM tokens, by model and type")
cost = meter.create_counter("assistant.llm.cost_usd", description="estimated LLM cost in US dollars")
ttft = meter.create_histogram("assistant.ttft", unit="s", description="time to first answer token")
duration = meter.create_histogram("assistant.duration", unit="s", description="time to the complete answer")
tool_calls = meter.create_counter("assistant.tool.calls", description="tool calls, by tool and outcome")
guardrail_blocks = meter.create_counter("assistant.guardrail.blocks", description="inputs or outputs blocked")

# US dollars per million tokens (input, output). Local models cost electricity, not tokens.
PRICES = {
    "anthropic/claude-opus-5-5": (4.00, 20.00),
    "anthropic/claude-sonnet-5-5": (2.00, 10.00),
    "anthropic/claude-haiku-4-5": (1.00, 5.00),
}


def price(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    per_in, per_out = PRICES.get(model, (0.0, 0.0))
    return (prompt_tokens * per_in + completion_tokens * per_out) / 1_000_000