LLMOps: eval harness, drift detection e canary de prompts

# Estrutura simples
#   prompts/
#     extract_ticket/
#       v3.md          ← prompt em markdown com frontmatter YAML
#       v3.json        ← schema de input/output
#       v3.eval.jsonl  ← golden set opcional específico deste prompt
from pathlib import Path
import hashlib, yaml, json

class Prompt:
    def __init__(self, path: Path):
        text = path.read_text()
        if text.startswith("---"):
            meta_block, body = text.split("---", 2)[1:]
            self.meta = yaml.safe_load(meta_block)
        else:
            self.meta, body = {}, text
        self.body = body.strip()
        self.hash = hashlib.sha256(self.body.encode()).hexdigest()[:12]

    def format(self, **kwargs) -> str:
        return self.body.format(**kwargs)

# Uso
p = Prompt(Path("prompts/extract_ticket/v3.md"))
prompt_text = p.format(ticket=ticket_text)
# Log: prompt_hash=p.hash, version=p.meta.get("version"), model=p.meta.get("model")

# promptfoo é uma das opções prontas — config declarativa
prompts: [file://prompts/extract_ticket/v3.md]
providers:
  - id: anthropic:messages:claude-sonnet-4-6
    config: { temperature: 0 }

tests:
  - description: ticket simples
    vars: { ticket: "Meu cartão não passou na renovação" }
    assert:
      - type: javascript
        value: output.includes("billing") && output.includes("card_declined")
      - type: latency
        threshold: 3000
      - type: cost
        threshold: 0.01

defaultTest:
  assert:
    - type: llm-rubric
      value: A resposta deve ser JSON válido com campos category, urgency (1-5), summary.
      provider: openai:chat:gpt-4o-mini

# Rota simples com sticky assignment por user_id
import hashlib

def canary_prompt(user_id: str, percent: int, control_hash: str, canary_hash: str) -> str:
    bucket = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100
    return canary_hash if bucket < percent else control_hash

# Ao chamar LLM, log:
#   variant = "control" | "canary"
#   quality_score (estimado via eval online ou modelo judge)
#   cost_usd
#   latency_ms

# Regra de promoção (pseudo)
#   se canary_sample > 1000 E
#      quality_delta > -2pp E
#      cost_delta < +15% E
#      latency_p95_delta < +200ms:
#     promover percent +20 até 100
#   senão:
#     rollback para 0

# Exemplo simples de drift por distribuição de tópicos (PSI)
import numpy as np

def psi(expected: np.ndarray, actual: np.ndarray, eps=1e-6) -> float:
    expected = np.clip(expected, eps, None); expected /= expected.sum()
    actual   = np.clip(actual,   eps, None); actual   /= actual.sum()
    return float(np.sum((actual - expected) * np.log(actual / expected)))

# Usa topic distribution (cluster de embeddings) em janela móvel
# PSI < 0.1 estável, 0.1-0.25 atenção, > 0.25 drift significativo