IA & Agentes

# Arquitetura RAG mínima funcional
# 1. Indexação (roda offline, quando o conteúdo muda)
def index_document(doc: str, doc_id: str):
    # Chunking — blocos com overlap para não perder contexto
    chunks = chunk_with_overlap(doc, size=800, overlap=100)
    
    for i, chunk in enumerate(chunks):
        embedding = embed(chunk)  # via API: text-embedding-004
        vector_store.upsert(
            id=f"{doc_id}_{i}",
            vector=embedding,
            metadata={"text": chunk, "doc_id": doc_id}
        )

# 2. Retrieval (em tempo real, por query do usuário)
def retrieve(query: str, top_k: int = 5) -> list[str]:
    query_embedding = embed(query)
    
    # Busca semântica (vetores)
    semantic_results = vector_store.query(query_embedding, top_k=top_k)
    
    # Busca léxica (palavras exatas) — não ignore isso
    lexical_results = full_text_search(query, top_k=top_k)
    
    # Rerank híbrido — RRF (Reciprocal Rank Fusion)
    return reciprocal_rank_fusion([semantic_results, lexical_results])

# 3. Generation — contexto injetado no prompt
def answer(query: str) -> str:
    context_chunks = retrieve(query)
    context = "\n\n".join(context_chunks)
    
    prompt = f"""Responda baseado APENAS no contexto abaixo.
Se a resposta não estiver no contexto, diga explicitamente.

CONTEXTO:
{context}

PERGUNTA: {query}"""
    
    return llm.generate(prompt)

# Controle de custo real em produção

# 1. Compressão de histórico — não mande tudo
def compress_history(history: list[dict], max_turns: int = 4) -> str:
    if len(history) <= max_turns:
        return format_history(history)
    
    # Resumo dos turnos antigos + últimas N mensagens completas
    old_turns = history[:-max_turns]
    recent_turns = history[-max_turns:]
    
    summary = llm.generate(
        f"Resuma em 2-3 frases os pontos principais desta conversa: {old_turns}",
        config={"maxOutputTokens": 150}  # resumo curto, modelo pequeno
    )
    
    return f"[Resumo anterior: {summary}]\n{format_history(recent_turns)}"

# 2. Roteamento por complexidade — modelo certo para cada tarefa
def route_to_model(query: str) -> str:
    # Classificação rápida e barata
    is_complex = classifier.predict(query)  # modelo local ou flash
    
    return "gemini-2.5-pro" if is_complex else "gemini-2.5-flash"

# 3. Limite explícito de output
generation_config = {
    "maxOutputTokens": 600,   # suficiente para resposta útil
    "temperature": 0.3,       # mais determinístico = menos tokens desperdiçados
}

# Anonymization básica antes de enviar para LLM externo
import re

def anonymize_for_llm(text: str) -> tuple[str, dict]:
    """Remove PII antes de enviar. Retorna texto limpo + mapa para substituição."""
    replacements = {}
    counter = {"cpf": 0, "email": 0, "phone": 0}
    
    # CPF
    def replace_cpf(m):
        key = f"[CPF_{counter['cpf']}]"
        replacements[key] = m.group(0)
        counter["cpf"] += 1
        return key
    
    # Email  
    def replace_email(m):
        key = f"[EMAIL_{counter['email']}]"
        replacements[key] = m.group(0)
        counter["email"] += 1
        return key
    
    text = re.sub(r'\d{3}\.\d{3}\.\d{3}-\d{2}', replace_cpf, text)
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', replace_email, text)
    
    return text, replacements

# Uso
clean_text, pii_map = anonymize_for_llm(user_message)
llm_response = llm.generate(clean_text)

# Nunca logue pii_map — ele contém os dados originais
# Use apenas se precisar reconstituir a resposta com dados reais

# Abstração de provider: troca de cloud sem mexer na lógica
from abc import ABC, abstractmethod

class LLMProvider(ABC):
    @abstractmethod
    def generate(self, prompt: str, config: dict) -> str: ...

# Google Cloud
class VertexProvider(LLMProvider):
    def generate(self, prompt: str, config: dict) -> str:
        from vertexai.generative_models import GenerativeModel
        model = GenerativeModel(config.get("model", "gemini-2.5-flash"))
        response = model.generate_content(prompt)
        return response.text

# AWS
class BedrockProvider(LLMProvider):
    def generate(self, prompt: str, config: dict) -> str:
        import boto3
        client = boto3.client("bedrock-runtime")
        # Claude via Bedrock
        response = client.invoke_model(
            modelId=config.get("model", "anthropic.claude-3-haiku"),
            body={"prompt": prompt, "max_tokens": config.get("max_tokens", 600)}
        )
        return response["body"].read().decode()

# Local (Ollama)
class OllamaProvider(LLMProvider):
    def generate(self, prompt: str, config: dict) -> str:
        import requests
        r = requests.post("http://localhost:11434/api/generate", json={
            "model": config.get("model", "llama3.2"),
            "prompt": prompt, "stream": False
        })
        return r.json()["response"]

# Roteamento inteligente
def get_provider(task_type: str) -> LLMProvider:
    if task_type in ("classify", "extract", "moderate"):
        return OllamaProvider()   # local, rápido, privado
    if task_type == "generate":
        return VertexProvider()   # cloud, modelo grande
    return BedrockProvider()      # fallback multi-cloud