Fundamentos de IA Generativa

import openai
import os
from typing import Optional

# Configuração do cliente OpenAI
# A chave é lida de variável de ambiente — nunca hardcode em código
client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

def gerar_resposta(
    prompt: str,
    temperatura: float = 0.7,
    modelo: str = "gpt-4o-mini",
    max_tokens: int = 500,
    system_prompt: Optional[str] = None
) -> str:
    """
    Gera uma resposta usando a API da OpenAI.

    Args:
        prompt: A pergunta ou instrução para o modelo
        temperatura: Controla aleatoriedade (0.0 = determinístico, 2.0 = muito criativo)
        modelo: Identificador do modelo a usar
        max_tokens: Limite de tokens na resposta
        system_prompt: Instrução de sistema que define o comportamento do assistente

    Returns:
        Texto da resposta gerada
    """
    mensagens = []

    # O system prompt define a "personalidade" e restrições do modelo
    if system_prompt:
        mensagens.append({"role": "system", "content": system_prompt})

    mensagens.append({"role": "user", "content": prompt})

    resposta = client.chat.completions.create(
        model=modelo,
        messages=mensagens,
        temperature=temperatura,  # Float entre 0 e 2
        max_tokens=max_tokens,
        # top_p=0.9,              # Alternativa à temperatura (nucleus sampling)
        # frequency_penalty=0.1,  # Penaliza repetição de tokens frequentes
        # presence_penalty=0.1,   # Penaliza tokens que já apareceram
    )

    return resposta.choices[0].message.content

# ─── Demonstração do impacto da temperatura ───────────────────────────────────

prompt_criativo = "Descreva o futuro da programação em 2 frases."

print("=" * 60)
print("TEMPERATURA 0.0 — Resposta determinística e conservadora:")
print("=" * 60)
for _ in range(2):
    resposta = gerar_resposta(prompt_criativo, temperatura=0.0)
    print(f"  → {resposta}\n")
    # Com temp=0.0, ambas as respostas serão idênticas

print("=" * 60)
print("TEMPERATURA 1.0 — Balancede criatividade e coerência:")
print("=" * 60)
for _ in range(2):
    resposta = gerar_resposta(prompt_criativo, temperatura=1.0)
    print(f"  → {resposta}\n")
    # Com temp=1.0, respostas serão semelhantes mas com variações

print("=" * 60)
print("TEMPERATURA 1.8 — Muito criativo, pode ser incoerente:")
print("=" * 60)
resposta = gerar_resposta(prompt_criativo, temperatura=1.8)
print(f"  → {resposta}\n")

# ─── Exemplo com system prompt para comportamento especializado ───────────────

system_arquiteto = """
Você é um arquiteto de software sênior com 20 anos de experiência.
Responda sempre com exemplos práticos e mencione trade-offs.
Seja direto e evite respostas genéricas.
"""

resposta_tecnica = gerar_resposta(
    "Como devo estruturar uma aplicação que usa LLMs em produção?",
    temperatura=0.3,  # Baixa para respostas técnicas mais consistentes
    system_prompt=system_arquiteto
)
print("Resposta técnica especializada:")
print(resposta_tecnica)

import tiktoken
import json

# tiktoken é a biblioteca oficial da OpenAI para tokenização
# Instalar: pip install tiktoken

def analisar_tokens(texto: str, modelo: str = "gpt-4o") -> dict:
    """
    Analisa a tokenização de um texto para um modelo específico.
    Retorna contagem, custo estimado e visualização dos tokens.
    """
    # Cada modelo usa um encoding diferente
    # gpt-4, gpt-4o, gpt-3.5-turbo usam "cl100k_base"
    enc = tiktoken.encoding_for_model(modelo)

    # Encode: converte texto em lista de inteiros (IDs de tokens)
    token_ids = enc.encode(texto)

    # Decode individual: vê a representação de cada token
    tokens_visuais = []
    for token_id in token_ids:
        # decode_single_token_bytes retorna os bytes do token
        token_bytes = enc.decode_single_token_bytes(token_id)
        tokens_visuais.append({
            "id": token_id,
            "texto": token_bytes.decode("utf-8", errors="replace"),
            "bytes": len(token_bytes)
        })

    # Preços aproximados do GPT-4o (verificar https://openai.com/pricing)
    PRECO_INPUT_POR_MILHAO = 2.50   # USD
    PRECO_OUTPUT_POR_MILHAO = 10.00  # USD

    custo_estimado_input = (len(token_ids) / 1_000_000) * PRECO_INPUT_POR_MILHAO

    return {
        "texto_original": texto,
        "total_tokens": len(token_ids),
        "custo_estimado_input_usd": round(custo_estimado_input, 8),
        "tokens": tokens_visuais[:20],  # Primeiros 20 para visualização
        "ratio_chars_por_token": round(len(texto) / len(token_ids), 2)
    }

# ─── Comparação de tokenização PT vs EN ───────────────────────────────────────

textos = {
    "Inglês simples": "The quick brown fox jumps over the lazy dog",
    "Português simples": "O rato roeu a roupa do rei de Roma",
    "Código Python": "def fibonacci(n):\n    return n if n <= 1 else fibonacci(n-1) + fibonacci(n-2)",
    "Texto técnico PT": "implementação de microsserviços com orquestração de contêineres",
    "JSON estruturado": '{"nome": "João", "empresa": "TechCorp", "cargo": "Desenvolvedor Sênior"}',
}

for descricao, texto in textos.items():
    resultado = analisar_tokens(texto)
    print(f"\n{'─'*50}")
    print(f"📝 {descricao}")
    print(f"   Caracteres: {len(texto)}")
    print(f"   Tokens:     {resultado['total_tokens']}")
    print(f"   Ratio:      {resultado['ratio_chars_por_token']} chars/token")

    # Visualiza os primeiros tokens
    tokens_str = " | ".join(
        f"[{t['texto']!r}]" for t in resultado['tokens'][:10]
    )
    print(f"   Tokens:     {tokens_str}")

# ─── Calculando limite de contexto ────────────────────────────────────────────

def verificar_limite_contexto(
    mensagens: list[dict],
    limite_contexto: int = 128_000  # GPT-4o tem 128K de contexto
) -> dict:
    """Verifica se as mensagens cabem no contexto do modelo."""
    enc = tiktoken.encoding_for_model("gpt-4o")

    # Cada mensagem tem overhead de ~4 tokens para metadados
    total_tokens = 3  # overhead inicial

    for msg in mensagens:
        total_tokens += 4  # overhead por mensagem
        total_tokens += len(enc.encode(msg.get("content", "")))
        if msg.get("name"):
            total_tokens += 1

    tokens_disponiveis = limite_contexto - total_tokens
    percentual_usado = (total_tokens / limite_contexto) * 100

    return {
        "tokens_usados": total_tokens,
        "tokens_disponiveis": tokens_disponiveis,
        "percentual_usado": round(percentual_usado, 2),
        "cabe_no_contexto": total_tokens < limite_contexto
    }

mensagens_exemplo = [
    {"role": "system", "content": "Você é um assistente especializado em Python."},
    {"role": "user", "content": "Como implementar um decorator de cache com TTL?"},
    {"role": "assistant", "content": "Para implementar um decorator de cache com TTL..."},
]

status = verificar_limite_contexto(mensagens_exemplo)
print(f"\n✅ Status do contexto: {json.dumps(status, indent=2, ensure_ascii=False)}")

import openai
import numpy as np
from typing import Union
import os

client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

def gerar_embedding(texto: str, modelo: str = "text-embedding-3-small") -> list[float]:
    """
    Gera um embedding para o texto usando a API da OpenAI.

    Modelos disponíveis:
    - text-embedding-3-small: 1536 dimensões, mais barato (~$0.02/1M tokens)
    - text-embedding-3-large: 3072 dimensões, mais preciso (~$0.13/1M tokens)
    - text-embedding-ada-002: modelo legado, ainda amplamente usado
    """
    resposta = client.embeddings.create(
        input=texto,
        model=modelo,
        # encoding_format="float"  # ou "base64" para transferência eficiente
    )
    return resposta.data[0].embedding

def similaridade_cosseno(vec_a: list[float], vec_b: list[float]) -> float:
    """
    Calcula a similaridade do cosseno entre dois vetores.

    Fórmula: cos(θ) = (A · B) / (||A|| × ||B||)

    Resultado:
    -  1.0: Vetores idênticos (textos semanticamente equivalentes)
    -  0.0: Sem relação semântica
    - -1.0: Semanticamente opostos
    """
    a = np.array(vec_a)
    b = np.array(vec_b)

    # Produto escalar dividido pelo produto das normas
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def busca_semantica(
    query: str,
    documentos: list[dict],
    top_k: int = 3
) -> list[dict]:
    """
    Realiza busca semântica: encontra os documentos mais similares à query.

    Args:
        query: Texto da busca
        documentos: Lista de dicts com 'titulo', 'conteudo' e 'embedding'
        top_k: Número de resultados a retornar

    Returns:
        Top-k documentos mais similares com seus scores
    """
    # Gera embedding da query
    query_embedding = gerar_embedding(query)

    # Calcula similaridade com cada documento
    resultados = []
    for doc in documentos:
        score = similaridade_cosseno(query_embedding, doc["embedding"])
        resultados.append({
            **doc,
            "score": score
        })

    # Ordena por score decrescente e retorna top-k
    resultados.sort(key=lambda x: x["score"], reverse=True)
    return resultados[:top_k]

# ─── Exemplo prático: base de conhecimento técnica ────────────────────────────

base_conhecimento = [
    "Python é uma linguagem de programação de alto nível com sintaxe limpa",
    "Kubernetes orquestra containers Docker em clusters distribuídos",
    "Redis é um banco de dados em memória usado para cache e mensageria",
    "A API da OpenAI permite integrar modelos de linguagem em aplicações",
    "GraphQL é uma linguagem de consulta para APIs que permite buscar dados específicos",
    "PostgreSQL é um banco relacional ACID-compliant com suporte a JSON e arrays",
    "Transformers usam mecanismo de atenção para processar sequências de texto",
    "Docker empacota aplicações em containers portáteis e isolados",
]

print("📚 Gerando embeddings para a base de conhecimento...")
documentos_com_embeddings = []
for i, texto in enumerate(base_conhecimento):
    embedding = gerar_embedding(texto)
    documentos_com_embeddings.append({
        "id": i,
        "conteudo": texto,
        "embedding": embedding
    })
    print(f"  ✓ Documento {i+1}/{len(base_conhecimento)}: {texto[:50]}...")

# Testa buscas semânticas com diferentes queries
queries_teste = [
    "Como usar IA em minha aplicação?",           # Deve encontrar OpenAI
    "Banco de dados para alta performance",        # Deve encontrar Redis/PostgreSQL
    "Deploy de aplicações em produção",            # Deve encontrar Kubernetes/Docker
    "Como modelos de linguagem funcionam?",        # Deve encontrar Transformers
]

print("\n" + "=" * 60)
for query in queries_teste:
    print(f"\n🔍 Query: '{query}'")
    resultados = busca_semantica(query, documentos_com_embeddings, top_k=2)

    for rank, resultado in enumerate(resultados, 1):
        barra_score = "█" * int(resultado["score"] * 20)
        print(f"  {rank}. [{barra_score:<20}] {resultado['score']:.4f}")
        print(f"     {resultado['conteudo'][:70]}...")

import openai
import numpy as np
import os
from dataclasses import dataclass, field

client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

@dataclass
class Chunk:
    """Representa um pedaço de documento indexado."""
    id: str
    conteudo: str
    metadados: dict
    embedding: list[float] = field(default_factory=list)

class VectorStoreSimples:
    """
    Implementação simples de vector store em memória.
    Em produção, use Pinecone, Weaviate, Qdrant ou pgvector.
    """
    def __init__(self):
        self.chunks: list[Chunk] = []

    def adicionar(self, chunk: Chunk):
        self.chunks.append(chunk)

    def buscar(self, query_embedding: list[float], top_k: int = 3) -> list[tuple[Chunk, float]]:
        """Retorna os top-k chunks mais similares à query."""
        if not self.chunks:
            return []

        query_vec = np.array(query_embedding)
        resultados = []

        for chunk in self.chunks:
            chunk_vec = np.array(chunk.embedding)
            # Similaridade cosseno
            score = float(
                np.dot(query_vec, chunk_vec) /
                (np.linalg.norm(query_vec) * np.linalg.norm(chunk_vec))
            )
            resultados.append((chunk, score))

        # Ordena por score e retorna top-k
        resultados.sort(key=lambda x: x[1], reverse=True)
        return resultados[:top_k]

class RagPipeline:
    """
    Pipeline RAG (Retrieval-Augmented Generation) completo.

    Fluxo:
    1. INDEXAÇÃO: Divide documentos → gera embeddings → armazena
    2. CONSULTA: Embed da query → busca → constrói prompt → gera resposta
    """

    def __init__(self, modelo_embedding="text-embedding-3-small", modelo_llm="gpt-4o-mini"):
        self.modelo_embedding = modelo_embedding
        self.modelo_llm = modelo_llm
        self.vector_store = VectorStoreSimples()

    def _gerar_embedding(self, texto: str) -> list[float]:
        resp = client.embeddings.create(input=texto, model=self.modelo_embedding)
        return resp.data[0].embedding

    def _dividir_em_chunks(self, texto: str, tamanho: int = 500, sobreposicao: int = 50) -> list[str]:
        """
        Divide texto em chunks com sobreposição para não perder contexto.
        Em produção, use RecursiveCharacterTextSplitter do LangChain.
        """
        palavras = texto.split()
        chunks = []

        for inicio in range(0, len(palavras), tamanho - sobreposicao):
            chunk = " ".join(palavras[inicio:inicio + tamanho])
            if chunk:
                chunks.append(chunk)

        return chunks

    def indexar_documento(self, conteudo: str, metadados: dict = None):
        """
        FASE 1: Indexação
        Divide o documento em chunks, gera embeddings e armazena.
        """
        print(f"📚 Indexando documento: {metadados.get('titulo', 'Sem título')}")
        chunks = self._dividir_em_chunks(conteudo)

        for i, texto_chunk in enumerate(chunks):
            # Gera embedding para o chunk
            embedding = self._gerar_embedding(texto_chunk)

            chunk = Chunk(
                id=f"{metadados.get('id', 'doc')}_{i}",
                conteudo=texto_chunk,
                metadados=metadados or {},
                embedding=embedding
            )
            self.vector_store.adicionar(chunk)
            print(f"  ✓ Chunk {i+1}/{len(chunks)} indexado ({len(texto_chunk.split())} palavras)")

    def consultar(self, pergunta: str, top_k: int = 3) -> dict:
        """
        FASE 2: Consulta
        1. Gera embedding da pergunta
        2. Busca chunks mais relevantes
        3. Constrói prompt com contexto recuperado
        4. Gera resposta grounded nos documentos
        """
        print(f"\n🔍 Consultando: '{pergunta}'")

        # Passo 1: Embed da query
        query_embedding = self._gerar_embedding(pergunta)

        # Passo 2: Recupera contexto relevante
        chunks_relevantes = self.vector_store.buscar(query_embedding, top_k=top_k)

        print(f"  Encontrados {len(chunks_relevantes)} chunks relevantes:")
        for chunk, score in chunks_relevantes:
            print(f"  - [{score:.3f}] {chunk.metadados.get('titulo', '')} — {chunk.conteudo[:60]}...")

        # Passo 3: Constrói prompt com contexto (o "augmented" do RAG)
        contexto = "\n\n---\n\n".join([
            f"Fonte: {c.metadados.get('titulo', 'Desconhecida')}\n{c.conteudo}"
            for c, _ in chunks_relevantes
        ])

        prompt_sistema = """
        Você é um assistente especializado. Responda APENAS com base nos documentos fornecidos.
        Se a informação não estiver nos documentos, diga explicitamente que não possui essa informação.
        Cite a fonte quando possível.
        """

        prompt_usuario = f"""
        Documentos de contexto:
        {contexto}

        Pergunta: {pergunta}

        Responda de forma clara e objetiva, citando as fontes relevantes.
        """

        # Passo 4: Gera resposta com contexto
        resposta = client.chat.completions.create(
            model=self.modelo_llm,
            messages=[
                {"role": "system", "content": prompt_sistema},
                {"role": "user", "content": prompt_usuario}
            ],
            temperature=0.1  # Baixa temperatura para respostas factuais
        )

        return {
            "pergunta": pergunta,
            "resposta": resposta.choices[0].message.content,
            "chunks_usados": [
                {"fonte": c.metadados.get("titulo"), "score": round(s, 4)}
                for c, s in chunks_relevantes
            ],
            "tokens_usados": resposta.usage.total_tokens
        }

# ─── Uso do Pipeline RAG ──────────────────────────────────────────────────────

rag = RagPipeline()

# Indexa documentos (em produção: PDFs, páginas web, banco de dados)
documentos = [
    {
        "id": "doc1",
        "titulo": "Política de Férias 2025",
        "conteudo": """
        Todo colaborador tem direito a 30 dias de férias anuais após 12 meses de trabalho.
        As férias podem ser divididas em até 3 períodos, sendo um mínimo de 14 dias corridos.
        O pedido deve ser feito com 30 dias de antecedência pelo sistema de RH.
        Férias não gozadas são pagas em dobro no mês seguinte ao vencimento.
        """,
        "metadados": {"tipo": "rh", "versao": "2025.1"}
    },
    {
        "id": "doc2",
        "titulo": "Guia de Onboarding Tech",
        "conteudo": """
        Novos desenvolvedores devem configurar: VPN corporativa, acesso ao GitHub Enterprise,
        Docker Desktop, VSCode com extensões padrão. O setup completo leva aproximadamente 2 horas.
        Todo código deve passar por code review de pelo menos 2 aprovações antes do merge.
        Deploy em produção requer aprovação do tech lead.
        """,
        "metadados": {"tipo": "tech", "versao": "2025.3"}
    },
]

for doc in documentos:
    metadados = {**doc["metadados"], "titulo": doc["titulo"], "id": doc["id"]}
    rag.indexar_documento(doc["conteudo"], metadados)

# Consultas
resultado = rag.consultar("Quantos dias de férias tenho direito?")
print(f"\n💬 Resposta: {resultado['resposta']}")
print(f"📊 Tokens usados: {resultado['tokens_usados']}")