2.2 — Conectando-se a Múltiplos Provedores de Modelos via API (Multi-provider)

Cliente → HTTP Request (JSON) → API Gateway → Load Balancer → GPU Cluster → Response

# Estrutura universal de um request LLM

{

    "model": str,           # identificador do modelo

    "messages": [           # histórico de conversa

        {"role": "system", "content": str},   # instrução do sistema (opcional)

        {"role": "user", "content": str},     # input do usuário

        {"role": "assistant", "content": str}, # resposta anterior (para multi-turn)

    ],

    "max_tokens": int,      # limite de tokens de output

    "temperature": float,   # 0.0 = determinístico, 1.0+ = criativo

    # ...parâmetros opcionais específicos de cada provedor

}

{

    "text": str,            # o texto gerado

    "input_tokens": int,    # tokens consumidos no input

    "output_tokens": int,   # tokens gerados no output

    # ...metadados específicos do provedor (stop reason, etc.)

}

from anthropic import Anthropic



client = Anthropic(api_key='...')  # usa ANTHROPIC_API_KEY do env se omitido



# Request básico

response = client.messages.create(

    model='claude-haiku-4-5-20251001',

    max_tokens=500,

    system='Você é um assistente técnico.',   # system prompt separado de messages

    messages=[

        {'role': 'user', 'content': 'O que é um tensor?'}

    ]

)



# Extraindo o texto

texto = response.content[0].text        # content é lista (pode ter múltiplos blocos)



# Extraindo uso de tokens

input_tokens = response.usage.input_tokens

output_tokens = response.usage.output_tokens



# Stop reason (importante para tool_use loops)

stop_reason = response.stop_reason     # 'end_turn' | 'max_tokens' | 'tool_use'

from openai import OpenAI



client = OpenAI(api_key='...')  # usa OPENAI_API_KEY do env



response = client.chat.completions.create(

    model='gpt-4o-mini',

    max_tokens=500,

    messages=[

        {'role': 'system', 'content': 'Você é um assistente técnico.'},  # system dentro de messages

        {'role': 'user', 'content': 'O que é um tensor?'}

    ]

)



texto = response.choices[0].message.content   # choices[0] (pode ter N completions)

input_tokens = response.usage.prompt_tokens

output_tokens = response.usage.completion_tokens

import google.generativeai as genai



genai.configure(api_key='...')  # usa GOOGLE_API_KEY do env



model = genai.GenerativeModel('gemini-1.5-flash')



# Gemini tem API diferente — não usa messages como lista diretamente

response = model.generate_content('O que é um tensor?')

texto = response.text



# Para multi-turn (chat)

chat = model.start_chat()

response = chat.send_message('O que é um tensor?')

texto = response.text



# Tokens de uso

input_tokens = response.usage_metadata.prompt_token_count

output_tokens = response.usage_metadata.candidates_token_count

# Anthropic

from anthropic import RateLimitError, APIError, APIConnectionError

# anthropic.RateLimitError → 429

# anthropic.APIError → 4xx/5xx genérico

# anthropic.APIConnectionError → timeout/rede



# OpenAI

from openai import RateLimitError, APIError, APIConnectionError

# mesmos nomes, implementações diferentes



# Google

from google.api_core.exceptions import ResourceExhausted, GoogleAPIError

# ResourceExhausted → 429 (rate limit)

# GoogleAPIError → erros genéricos da API



# Pattern unificado de retry

import time

import random



def com_retry(fn, max_tentativas=3, base_delay=1.0):

    for tentativa in range(max_tentativas):

        try:

            return fn()

        except (RateLimitError, ResourceExhausted):

            if tentativa == max_tentativas - 1:

                raise

            # Backoff exponencial com jitter (aleatoriedade previne "thundering herd")

            delay = base_delay * (2 ** tentativa) + random.uniform(0, 1)

            time.sleep(delay)

        except (APIConnectionError,) as e:

            if tentativa == max_tentativas - 1:

                raise

            time.sleep(base_delay)

import os

import time

import random

from abc import ABC, abstractmethod

from dataclasses import dataclass

from typing import Optional





@dataclass

class Resposta:

    texto: str

    tokens_input: int

    tokens_output: int

    modelo: str

    provedor: str



    @property

    def custo_estimado_usd(self) -> float:

        """Estimativa de custo baseada em preços públicos de 2025."""

        precos = {

            # (input_per_1M, output_per_1M)

            'anthropic/claude-haiku-4-5-20251001': (0.25, 1.25),

            'anthropic/claude-sonnet-4-6': (3.0, 15.0),

            'openai/gpt-4o-mini': (0.15, 0.60),

            'openai/gpt-4o': (2.50, 10.0),

            'google/gemini-1.5-flash': (0.075, 0.30),

        }

        key = f'{self.provedor}/{self.modelo}'

        if key not in precos:

            return 0.0

        inp, out = precos[key]

        return (self.tokens_input / 1_000_000) * inp + (self.tokens_output / 1_000_000) * out





class LLMClient(ABC):

    """Interface abstrata — todos os clientes implementam esta."""



    @abstractmethod

    def completar(

        self,

        mensagens: list[dict],

        max_tokens: int = 500,

        temperature: float = 0.7,

        system: Optional[str] = None,

    ) -> Resposta:

        pass



    def completar_com_retry(self, *args, max_tentativas=3, **kwargs) -> Resposta:

        """Retry com backoff exponencial + jitter."""

        for tentativa in range(max_tentativas):

            try:

                return self.completar(*args, **kwargs)

            except Exception as e:

                nome_erro = type(e).__name__

                # Só retenta em erros recuperáveis

                if 'RateLimit' in nome_erro or 'ResourceExhausted' in nome_erro or 'Timeout' in nome_erro:

                    if tentativa < max_tentativas - 1:

                        delay = (2 ** tentativa) + random.uniform(0, 0.5)

                        print(f"[retry] {nome_erro} — aguardando {delay:.1f}s (tentativa {tentativa+1}/{max_tentativas})")

                        time.sleep(delay)

                        continue

                raise  # não recuperável — propaga imediatamente





class ClienteAnthropic(LLMClient):

    """Cliente para API da Anthropic (Claude)."""



    def __init__(self, modelo: str = 'claude-haiku-4-5-20251001'):

        # TODO 1: Inicialize o cliente Anthropic

        # from anthropic import Anthropic

        # self.client = Anthropic(api_key=os.environ.get('ANTHROPIC_API_KEY'))

        # self.modelo = modelo

        self.modelo = modelo

        self._client = None  # substituir por cliente real no TODO



    def completar(self, mensagens, max_tokens=500, temperature=0.7, system=None) -> Resposta:

        # TODO 2: Use self.client.messages.create para chamar a API

        # - model=self.modelo

        # - max_tokens=max_tokens

        # - messages=mensagens

        # - system=system (se fornecido)

        # Retorne Resposta(texto, tokens_input, tokens_output, self.modelo, 'anthropic')

        raise NotImplementedError("TODO: implemente ClienteAnthropic.completar()")





class ClienteOpenAI(LLMClient):

    """Cliente para API da OpenAI (GPT)."""



    def __init__(self, modelo: str = 'gpt-4o-mini'):

        # TODO 3: Inicialize o cliente OpenAI

        # from openai import OpenAI

        # self.client = OpenAI(api_key=os.environ.get('OPENAI_API_KEY'))

        # self.modelo = modelo

        self.modelo = modelo



    def completar(self, mensagens, max_tokens=500, temperature=0.7, system=None) -> Resposta:

        # TODO 4: Use self.client.chat.completions.create

        # ATENÇÃO: OpenAI coloca system DENTRO de messages, não separado

        # Se system for fornecido, adicione {'role': 'system', 'content': system} no INÍCIO de mensagens

        # response.choices[0].message.content para o texto

        # response.usage.prompt_tokens / completion_tokens para tokens

        raise NotImplementedError("TODO: implemente ClienteOpenAI.completar()")





class ClienteMultiProvider(LLMClient):

    """Orquestra múltiplos clientes com fallback automático."""



    def __init__(self, clientes: list[LLMClient]):

        # Ordem de preferência: primeiro cliente é o primário

        self.clientes = clientes



    def completar(self, mensagens, max_tokens=500, temperature=0.7, system=None) -> Resposta:

        erros = []

        for cliente in self.clientes:

            try:

                return cliente.completar_com_retry(

                    mensagens, max_tokens=max_tokens,

                    temperature=temperature, system=system

                )

            except Exception as e:

                erros.append(f"{type(cliente).__name__}: {e}")

                print(f"[fallback] {type(cliente).__name__} falhou: {e}")

                continue



        raise RuntimeError(

            f"Todos os provedores falharam:\n" + "\n".join(erros)

        )

# Implementação completa (depois de completar os TODOs)



from anthropic import Anthropic as AnthropicSDK

from openai import OpenAI as OpenAISDK

import google.generativeai as genai



class ClienteAnthropic(LLMClient):

    def __init__(self, modelo='claude-haiku-4-5-20251001'):

        self.client = AnthropicSDK(api_key=os.environ['ANTHROPIC_API_KEY'])

        self.modelo = modelo



    def completar(self, mensagens, max_tokens=500, temperature=0.7, system=None):

        kwargs = {

            'model': self.modelo,

            'max_tokens': max_tokens,

            'messages': mensagens,

        }

        if system:

            kwargs['system'] = system



        response = self.client.messages.create(**kwargs)



        return Resposta(

            texto=response.content[0].text,

            tokens_input=response.usage.input_tokens,

            tokens_output=response.usage.output_tokens,

            modelo=self.modelo,

            provedor='anthropic',

        )





class ClienteOpenAI(LLMClient):

    def __init__(self, modelo='gpt-4o-mini'):

        self.client = OpenAISDK(api_key=os.environ['OPENAI_API_KEY'])

        self.modelo = modelo



    def completar(self, mensagens, max_tokens=500, temperature=0.7, system=None):

        # OpenAI: system vai DENTRO de messages, não separado

        msgs = list(mensagens)  # cópia para não mutar o original

        if system:

            msgs = [{'role': 'system', 'content': system}] + msgs



        response = self.client.chat.completions.create(

            model=self.modelo,

            max_tokens=max_tokens,

            messages=msgs,

        )



        return Resposta(

            texto=response.choices[0].message.content,

            tokens_input=response.usage.prompt_tokens,

            tokens_output=response.usage.completion_tokens,

            modelo=self.modelo,

            provedor='openai',

        )





# Comparação de qualidade entre provedores

def comparar_provedores(pergunta: str):

    clientes = {

        'Claude Haiku': ClienteAnthropic('claude-haiku-4-5-20251001'),

        'GPT-4o mini': ClienteOpenAI('gpt-4o-mini'),

    }



    mensagem = [{'role': 'user', 'content': pergunta}]

    system = 'Responda de forma técnica e concisa em 2-3 frases.'



    print(f"\nPergunta: {pergunta}\n" + "="*50)



    for nome, cliente in clientes.items():

        start = time.time()

        resposta = cliente.completar(mensagem, max_tokens=200, system=system)

        latencia = time.time() - start



        print(f"\n{nome}:")

        print(f"  Resposta: {resposta.texto}")

        print(f"  Tokens: {resposta.tokens_input} in / {resposta.tokens_output} out")

        print(f"  Custo: ${resposta.custo_estimado_usd:.6f}")

        print(f"  Latência: {latencia:.2f}s")





comparar_provedores("Explique o mecanismo de atenção em Transformers em 2 frases.")

# CORRETO: lê do ambiente

client = Anthropic(api_key=os.environ.get('ANTHROPIC_API_KEY'))



# ERRADO: expõe a chave no código (vai parar no git)

client = Anthropic(api_key='sk-ant-...')

# Em produção

cliente = ClienteAnthropic()



# Em testes

class ClienteMock(LLMClient):

    def completar(self, mensagens, **kwargs):

        return Resposta(texto="resposta mock", tokens_input=10, tokens_output=5, ...)



# Testes não fazem chamada de API real — rápidos e baratos

# Após cada request:

logger.info("llm_call", extra={

    "provedor": resposta.provedor,

    "modelo": resposta.modelo,

    "tokens_in": resposta.tokens_input,

    "tokens_out": resposta.tokens_output,

    "custo_usd": resposta.custo_estimado_usd,

    "latencia_ms": latencia_ms,

})

# ERRADO: usa Sonnet (caro) até para classificação simples

modelo = 'claude-sonnet-4-6'  # $3/$15 por 1M



# CORRETO: Haiku para classificação, Sonnet para geração complexa

modelo_classificacao = 'claude-haiku-4-5-20251001'   # $0.25/$1.25 — 12x mais barato

modelo_geracao = 'claude-sonnet-4-6'

# Pode retornar None se stop reason for 'content_filter'

texto = response.choices[0].message.content

if texto is None:  # ← precisa checar!

    raise ValueError(f"Conteúdo filtrado. Stop reason: {response.choices[0].finish_reason}")

# ERRADO: modifica o original (bug sutil em loops multi-turn)

if system:

    mensagens.insert(0, {'role': 'system', 'content': system})



# CORRETO: trabalha em cópia

msgs = [{'role': 'system', 'content': system}] + list(mensagens)

# ERRADO: comparação de string pode falhar se houver versão nova

if response.stop_reason == 'end_turn':  # ok hoje



# MELHOR: checar o que importa (não é tool_use)

if response.stop_reason != 'tool_use':

    # terminamos