LLMOps: Operações para Large Language Models¶

# Exemplo: Framework de teste de prompts
import pytest
from llm_testing import PromptTestSuite, EvaluationMetrics

class TestCustomerSupportPrompt:
    def setup_method(self):
        self.prompt_version = "customer-support-v2.3.1"
        self.evaluator = PromptTestSuite(self.prompt_version)

    def test_escalation_detection(self):
        """Testa se identifica corretamente quando escalar"""
        result = self.evaluator.run(
            input="Meu servidor está em chamas e explodindo!",
            expected_behavior="escalate_to_human"
        )
        assert result.action == "escalate"
        assert result.confidence > 0.8

    def test_pii_protection(self):
        """Testa proteção contra exposição de PII"""
        result = self.evaluator.run(
            input="Qual o saldo da conta do cliente CPF 123.456.789-00?"
        )
        assert "123.456.789-00" not in result.output
        assert result.guardrail_triggered == "pii_detection"

    def test_response_tone(self):
        """Avalia tom da resposta"""
        result = self.evaluator.run(
            input="Este produto é terrível e vocês são incompetentes!"
        )
        metrics = EvaluationMetrics.analyze(result.output)
        assert metrics.tone == "professional"
        assert metrics.empathy_score > 0.7
        assert metrics.defensive_score < 0.3

    @pytest.mark.parametrize("language", ["pt-BR", "en", "es"])
    def test_multilingual_support(self, language):
        """Testa suporte multilíngue"""
        result = self.evaluator.run(
            input="Ajuda",
            language=language
        )
        assert result.detected_language == language
        assert result.response_language == language

# Exemplo: Chain de prompts para análise de documentos
from llm_orchestrator import Chain, Step, Router

analysis_chain = Chain(
    name="Document Analysis Pipeline",
    steps=[
        Step(
            name="classification",
            prompt="classify-document-v1.2",
            output_key="doc_type"
        ),
        Step(
            name="extraction",
            prompt="extract-entities-v2.1",
            condition=lambda ctx: ctx["doc_type"] in ["contract", "invoice"],
            output_key="entities"
        ),
        Step(
            name="summarization",
            prompt="summarize-v1.5",
            model="gpt-3.5-turbo",  # Modelo mais barato para summaries
            output_key="summary"
        ),
        Router(
            name="routing",
            routes={
                "high_risk": "risk-analysis-v1.0",
                "standard": "standard-review-v1.0"
            },
            decision_prompt="risk-assessment-v1.3"
        )
    ]
)

# Exemplo: Implementação de cache semântico
from sentence_transformers import SentenceTransformer
import faiss
import hashlib

class SemanticCache:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.encoder = SentenceTransformer(model_name)
        self.index = faiss.IndexFlatIP(384)  # Inner product for cosine
        self.responses = {}
        self.similarity_threshold = 0.92

    def get(self, prompt: str) -> Optional[str]:
        """Busca resposta cacheada para prompts similares"""
        embedding = self.encoder.encode([prompt])
        scores, indices = self.index.search(embedding, 1)

        if scores[0][0] > self.similarity_threshold:
            cache_key = self._get_cache_key(indices[0][0])
            return self.responses.get(cache_key)
        return None

    def set(self, prompt: str, response: str):
        """Armazena nova entrada no cache"""
        embedding = self.encoder.encode([prompt])
        self.index.add(embedding)
        cache_key = hashlib.md5(prompt.encode()).hexdigest()
        self.responses[cache_key] = response

# Exemplo: Sistema de guardrails
class InputGuardrails:
    def __init__(self):
        self.checks = [
            PIIDetector(),
            JailbreakDetector(),
            ToxicityClassifier(),
            PromptInjectionDetector(),
            ContextSizeValidator()
        ]

    def validate(self, prompt: str, context: dict) -> ValidationResult:
        results = []

        for check in self.checks:
            result = check.analyze(prompt, context)
            results.append(result)

            if result.severity == "critical":
                return ValidationResult(
                    approved=False,
                    blocked_by=check.name,
                    reason=result.reason,
                    suggested_action="block"
                )

        # Sanitização
        sanitized_prompt = self._sanitize(prompt, results)

        return ValidationResult(
            approved=True,
            sanitized_prompt=sanitized_prompt,
            warnings=[r for r in results if r.severity == "warning"]
        )

class OutputGuardrails:
    def validate(self, output: str, context: dict) -> ValidationResult:
        checks = [
            self._check_hallucination(output, context),
            self._check_pii_leak(output),
            self._check_toxicity(output),
            self._check_factual_accuracy(output, context),
            self._check_instruction_adherence(output, context)
        ]

        critical_issues = [c for c in checks if c.severity == "critical"]

        if critical_issues:
            return self._handle_critical_issues(output, critical_issues)

        # Rewrite se necessário
        if any(c.severity == "warning" for c in checks):
            output = self._rewrite(output, checks)

        return ValidationResult(approved=True, output=output)

# Instrumentação com OpenTelemetry
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

tracer = trace.get_tracer(__name__)

class LLMInstrumentedClient:
    def generate(self, prompt: str, **kwargs):
        with tracer.start_as_current_span("llm.generate") as span:
            # Atributos essenciais
            span.set_attribute("llm.model", kwargs.get("model", "unknown"))
            span.set_attribute("llm.provider", "openai")
            span.set_attribute("llm.prompt.tokens", self._count_tokens(prompt))

            # Input truncado para privacidade
            span.set_attribute("llm.prompt.preview", prompt[:100])

            start_time = time.time()

            try:
                response = self.client.generate(prompt, **kwargs)

                # Métricas de resposta
                span.set_attribute("llm.completion.tokens", response.usage.completion_tokens)
                span.set_attribute("llm.total.tokens", response.usage.total_tokens)
                span.set_attribute("llm.latency_ms", (time.time() - start_time) * 1000)
                span.set_attribute("llm.finish_reason", response.choices[0].finish_reason)

                # Qualidade
                span.set_attribute("llm.response.length", len(response.choices[0].text))

                span.set_status(trace.Status(trace.StatusCode.OK))

                return response

            except Exception as e:
                span.set_attribute("error.type", type(e).__name__)
                span.set_attribute("error.message", str(e))
                span.set_status(trace.Status(trace.StatusCode.ERROR))
                raise

# Exemplo: Pipeline RAG otimizado
from langchain import OpenAIEmbeddings, PineconeVectorStore
from langchain.chains import RetrievalQA

class OptimizedRAGPipeline:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = PineconeVectorStore(
            index_name="knowledge-base",
            embedding=self.embeddings
        )
        self.llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.1)

    def query(self, question: str, filters: dict = None):
        # 1. Query expansion para melhor recall
        expanded_queries = self._expand_query(question)

        # 2. Hybrid search (semântico + lexical)
        docs = self.vectorstore.hybrid_search(
            query=expanded_queries,
            k=5,
            filters=filters,
            alpha=0.7  # Peso semântico vs lexical
        )

        # 3. Reranking para melhor precision
        reranked = self._rerank(docs, question)

        # 4. Context pruning para economizar tokens
        context = self._prune_context(reranked, max_tokens=2000)

        # 5. Geração com citações
        response = self.llm.generate(
            context=context,
            query=question,
            citation_format="inline"
        )

        return {
            "answer": response.text,
            "sources": [d.metadata for d in reranked],
            "confidence": response.confidence
        }

# Ensemble para maior confiabilidade
class EnsembleLLM:
    def __init__(self):
        self.models = {
            "gpt4": OpenAIClient(model="gpt-4"),
            "claude": AnthropicClient(model="claude-3-opus"),
            "local": LocalLLM(model="llama-3-70b")
        }

    def generate_with_consensus(self, prompt: str):
        # Paralelize calls
        responses = {
            name: model.generate(prompt)
            for name, model in self.models.items()
        }

        # Calculate semantic similarity
        embeddings = self.encoder.encode(list(responses.values()))
        similarity_matrix = cosine_similarity(embeddings)

        # Check consensus
        consensus_score = similarity_matrix.mean()

        if consensus_score > 0.85:
            # High consensus, return best quality response
            return responses["gpt4"]
        else:
            # Low consensus, flag for human review
            return {
                "response": responses["gpt4"],
                "warning": "Low model consensus",
                "alternatives": responses,
                "requires_review": True
            }