Da API ao Serviço em Produção

01De Script a Serviço

Porque um script não é um serviço

Um script corre uma vez, do início ao fim, e termina. Um serviço fica ligado, à espera de pedidos, um após outro, de pessoas diferentes, ao mesmo tempo. A diferença não é só técnica: um script que funciona no teu terminal não diz nada sobre o que acontece quando dez pessoas o chamam ao mesmo tempo, ou quando uma chamada falha a meio e mais ninguém fica a saber.

Este capítulo constrói o mínimo: um serviço HTTP real, em Python, que recebe um pedido e devolve uma resposta gerada por um LLM. O resto do curso vai construir à volta disto.

O mínimo de FastAPI

FastAPI é uma biblioteca Python para escrever serviços HTTP. Instala-se com pip install fastapi uvicorn (uvicorn é o programa que corre o serviço). Um serviço mínimo:

from fastapi import FastAPI

app = FastAPI()

@app.get("/saude")
def saude():
    return {"estado": "ok"}

Corre-se com uvicorn nome_do_ficheiro:app --reload, e fica à escuta em http://localhost:8000. @app.get("/saude") diz que esta função responde a pedidos GET no caminho /saude. O que a função devolve (aqui, um dicionário) é convertido automaticamente para JSON.

Um endpoint de saúde (/saude, ou /health) é uma convenção comum: um pedido simples que confirma que o serviço está vivo, sem fazer trabalho nenhum. Vais precisar disto mais tarde, quando outro sistema tiver de verificar se o teu serviço ainda responde.

Modelos de pedido e resposta

Um serviço a sério não recebe só texto solto, recebe uma estrutura definida. FastAPI usa Pydantic para isso: defines uma classe com os campos que esperas, e o FastAPI valida sozinho cada pedido contra ela.

from pydantic import BaseModel

class PedidoResumo(BaseModel):
    texto: str

class RespostaResumo(BaseModel):
    resumo: str

Isto não é só documentação: se alguém chamar o teu endpoint sem o campo texto, o FastAPI rejeita o pedido sozinho, antes de a tua função sequer correr, com uma resposta 422 a dizer exatamente o que falta. Nenhuma linha de código tua faz essa verificação; vem de graça por teres declarado o tipo.

O endpoint que chama o LLM

Junta os dois: um endpoint que recebe um PedidoResumo, chama a API da OpenAI (o Curso 4 desta casa ensina esta chamada desde o início, se precisares de rever), e devolve um RespostaResumo.

import os
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

class PedidoResumo(BaseModel):
    texto: str

class RespostaResumo(BaseModel):
    resumo: str

@app.post("/resumir", response_model=RespostaResumo)
def resumir(pedido: PedidoResumo):
    resposta = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Resume este texto numa frase: {pedido.texto}"}],
    )
    return RespostaResumo(resumo=resposta.choices[0].message.content)

response_model=RespostaResumo faz o mesmo do lado da resposta: se a tua função devolvesse um campo a mais ou a menos, o FastAPI avisava-te em vez de deixar passar uma resposta inconsistente para quem chamou.

Testar sem escrever nada de novo

Com o serviço a correr (uvicorn nome_do_ficheiro:app --reload), duas formas de o testar sem escrever um cliente: abrir http://localhost:8000/docs no browser, uma página interativa que o FastAPI gera sozinho a partir dos teus modelos, onde dá para experimentar o endpoint a clicar; ou pela linha de comandos:

curl -X POST http://localhost:8000/resumir \
  -H "Content-Type: application/json" \
  -d '{"texto": "O teu texto aqui."}'

02Chamadas que Falham

O que corre mal, e porquê

Uma chamada a um fornecedor de LLM externo pode falhar por razões que não têm nada a ver com o teu código: o fornecedor está a limitar quantos pedidos aceitas por minuto (rate limit), a rede demorou demasiado tempo a responder (timeout), ou o próprio servidor do fornecedor teve um problema temporário (erro 5xx). Nenhuma destas é permanente; tentar outra vez, momentos depois, costuma resolver.

É diferente de um erro de autenticação (a tua chave está errada) ou de um pedido malformado (mandaste um campo que o modelo não aceita): esses não se resolvem tentando outra vez, resolvem-se corrigindo o que enviaste. Retry só faz sentido para falhas transitórias.

Retry com backoff exponencial

Backoff exponencial significa esperar cada vez mais entre tentativas: 1 segundo, depois 2, depois 4. Isto evita bombardear um serviço já sobrecarregado com mais pedidos imediatos, o que só pioraria a situação.

import time
from openai import OpenAI, RateLimitError, APIConnectionError, APITimeoutError

client = OpenAI()

def chamar_com_retry(mensagens, tentativas_maximas=3):
    for tentativa in range(tentativas_maximas):
        try:
            return client.chat.completions.create(model="gpt-4o-mini", messages=mensagens)
        except (RateLimitError, APIConnectionError, APITimeoutError) as erro:
            espera = 2 ** tentativa
            print(f"Tentativa {tentativa + 1} falhou ({erro.__class__.__name__}), a esperar {espera}s...")
            time.sleep(espera)
    raise RuntimeError("Falhou depois de todas as tentativas")

RateLimitError, APIConnectionError e APITimeoutError são classes reais da biblioteca openai, cada uma correspondendo a um tipo de falha transitória. Apanhá-las especificamente, em vez de um Exception genérico, é o que te permite decidir que só estas vale a pena repetir.

O perigo de retries a mais

Isto que não é óbvio para quem vem de APIs normais: cada retry de uma chamada a um LLM custa dinheiro outra vez, como se fosse um pedido novo. Retentar um pedido de rede normal é essencialmente grátis; retentar uma chamada a um LLM três vezes é pagar três vezes pelo mesmo pedido, mesmo que só a última tentativa tenha tido sucesso.

Por isso tentativas_maximas tem um limite baixo (3, no exemplo acima), não 10 ou 20. Se o fornecedor continuar a falhar depois de 3 tentativas com backoff, o problema provavelmente não se resolve tentando mais vezes, e estás só a acumular custo.

03Processamento Assíncrono e Filas

Porque bloquear o pedido é mau design

Um pedido HTTP normal responde em milissegundos. Uma chamada a um LLM pode demorar vários segundos. Se o teu endpoint espera pela resposta do LLM antes de responder ao utilizador, estás a ocupar essa ligação (e os recursos do servidor) durante todo esse tempo. Com poucos pedidos ao mesmo tempo, não repara ninguém; com tráfego real, o serviço fica lento ou deixa de aceitar pedidos novos.

O padrão: aceitar, processar, consultar

Em vez de esperar, o endpoint aceita o pedido, devolve logo um identificador (job_id), e processa a chamada ao LLM em segundo plano. Quem pediu consulta o resultado mais tarde, por esse identificador, tantas vezes quantas precisar, até estar pronto.

import uuid
import threading
import queue
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
fila = queue.Queue()
resultados = {}

class PedidoResumo(BaseModel):
    texto: str

def processar(texto):
    return chamar_com_retry([{"role": "user", "content": f"Resume: {texto}"}])

def trabalhador():
    while True:
        job_id, texto = fila.get()
        resultados[job_id] = processar(texto)

threading.Thread(target=trabalhador, daemon=True).start()

@app.post("/resumir")
def pedir_resumo(pedido: PedidoResumo):
    job_id = str(uuid.uuid4())
    resultados[job_id] = None
    fila.put((job_id, pedido.texto))
    return {"job_id": job_id}

@app.get("/resumir/{job_id}")
def consultar_resumo(job_id: str):
    resultado = resultados.get(job_id)
    if resultado is None:
        return {"estado": "a processar"}
    return {"estado": "concluido", "resumo": resultado}

threading.Thread(..., daemon=True) arranca um trabalhador que corre para sempre em segundo plano, a tirar pedidos da fila um a um. queue.Queue() já vem pronta a lidar com vários pedidos a chegar ao mesmo tempo, sem teres de escrever essa parte à mão.

04Prática: o Teu Primeiro Serviço

import os
import time
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI, RateLimitError, APIConnectionError, APITimeoutError

app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])


class PedidoResumo(BaseModel):
    texto: str


class RespostaResumo(BaseModel):
    resumo: str


def chamar_com_retry(mensagens, tentativas_maximas=3):
    for tentativa in range(tentativas_maximas):
        try:
            return client.chat.completions.create(
                model="gpt-4o-mini",
                messages=mensagens,
                timeout=10,
            )
        except (RateLimitError, APIConnectionError, APITimeoutError) as erro:
            espera = 2 ** tentativa
            print(f"Tentativa {tentativa + 1} falhou ({erro.__class__.__name__}), a esperar {espera}s...")
            time.sleep(espera)
    raise RuntimeError("Falhou depois de todas as tentativas")


@app.get("/saude")
def saude():
    return {"estado": "ok"}


@app.post("/resumir", response_model=RespostaResumo)
def resumir(pedido: PedidoResumo):
    resposta = chamar_com_retry(
        [{"role": "user", "content": f"Resume este texto numa frase: {pedido.texto}"}]
    )
    return RespostaResumo(resumo=resposta.choices[0].message.content)

05Quanto Custa e Quanto Demora

Custo por pedido

Um LLM cobra por token, tanto do que envias (prompt) como do que ele devolve (completion). Cada resposta da API da OpenAI inclui essa contagem em resposta.usage:

resposta = client.chat.completions.create(model="gpt-4o-mini", messages=mensagens)

print(resposta.usage.prompt_tokens)      # tokens que enviaste
print(resposta.usage.completion_tokens)  # tokens que recebeste

PRECO_POR_1M_TOKENS_ENTRADA = 0.15  # exemplo; confirma o preço atual na documentação do fornecedor
PRECO_POR_1M_TOKENS_SAIDA = 0.60

custo = (
    resposta.usage.prompt_tokens / 1_000_000 * PRECO_POR_1M_TOKENS_ENTRADA
    + resposta.usage.completion_tokens / 1_000_000 * PRECO_POR_1M_TOKENS_SAIDA
)
print(f"Custo: ${custo:.6f}")

Os preços por token mudam com frequência e variam por modelo; os valores acima são só um exemplo de cálculo, não uma referência a copiar. Confirma sempre os preços atuais na página do fornecedor antes de os usar a sério.

Latência: a média engana

Imagina 100 pedidos ao teu serviço: 95 respondem em 1.2 segundos, 5 demoram 8 segundos (o fornecedor teve um dia mau nesses). A média sai perfeitamente aceitável:

import statistics

latencias = [1.2] * 95 + [8.0] * 5  # 100 pedidos: 95 rápidos, 5 lentos
latencias.sort()

indice_p95 = int(len(latencias) * 0.95)
p95 = latencias[indice_p95]
media = statistics.mean(latencias)

print(f"média: {media:.2f}s, p95: {p95:.2f}s")
# média: 1.54s, p95: 8.00s

A média (1.54s) esconde por completo o problema. O p95, o valor abaixo do qual ficam 95% dos pedidos, mostra exatamente o que a média escondeu: 5% dos teus utilizadores estão a ter uma experiência oito vezes pior do que a média sugere. É por isto que um SLA (capítulo 8) se compromete a p95 ou p99, nunca só à média.

06Quando o Fornecedor Falha

Fallback entre fornecedores

Depender de um único fornecedor de LLM é depender de um ponto único de falha: se a OpenAI tiver uma quebra, o teu serviço para com ela, mesmo que o problema não seja teu. Um fallback tenta o fornecedor principal e, se falhar mesmo depois dos retries, tenta um segundo.

from openai import OpenAI
import anthropic

client_openai = OpenAI()
client_anthropic = anthropic.Anthropic()

def gerar_resposta(prompt):
    try:
        resposta = chamar_com_retry([{"role": "user", "content": prompt}])
        return resposta.choices[0].message.content
    except RuntimeError:
        print("OpenAI esgotou as tentativas, a usar o fornecedor alternativo...")
        resposta = client_anthropic.messages.create(
            model="claude-sonnet-5",
            max_tokens=200,
            messages=[{"role": "user", "content": prompt}],
        )
        return resposta.content[0].text

O nome do modelo (aqui, claude-sonnet-5) muda com o tempo; confirma o modelo atual na documentação da Anthropic antes de usar isto a sério. gerar_resposta chama chamar_com_retry do capítulo 2, que já esgota as tentativas do fornecedor principal antes de este bloco tentar o alternativo.

Circuit breaker, em conceito

Sem mais nada, o teu serviço tenta sempre o fornecedor principal primeiro, mesmo que ele esteja em baixo há dez minutos, desperdiçando tempo e retries num pedido condenado. Um circuit breaker regista falhas e, a partir de um limite, "abre o circuito": salta logo para o fallback sem sequer tentar o principal, até passar um tempo de recuperação.

import time

class CircuitBreaker:
    def __init__(self, limite_falhas=3, tempo_recuperacao=30):
        self.limite_falhas = limite_falhas
        self.tempo_recuperacao = tempo_recuperacao
        self.falhas = 0
        self.aberto_desde = None

    def esta_aberto(self):
        if self.aberto_desde is None:
            return False
        if time.time() - self.aberto_desde > self.tempo_recuperacao:
            self.falhas = 0
            self.aberto_desde = None
            return False
        return True

    def registar_falha(self):
        self.falhas += 1
        if self.falhas >= self.limite_falhas:
            self.aberto_desde = time.time()

    def registar_sucesso(self):
        self.falhas = 0
        self.aberto_desde = None

Quando esta_aberto() devolve True, o teu código salta logo para o fallback. registar_sucesso() reinicia a contagem assim que o fornecedor principal voltar a responder bem, para o circuito não ficar aberto para sempre por engano.

07Cache

Quando cachear é seguro

Cachear a resposta de um LLM faz sentido quando o mesmo pedido, feito outra vez, devia dar uma resposta igualmente válida: classificar um texto, extrair um campo de um documento, tarefas onde o valor certo não muda consoante o dia. Poupas custo e latência de repetir uma chamada cujo resultado já tens.

Quando cachear é perigoso

É errado quando a variedade é o objetivo (uma resposta de conversa que deve soar natural, não repetida) ou quando o prompt depende de algo que muda (a data de hoje, um preço em tempo real): uma resposta cacheada fica desatualizada e incorreta sem nenhum aviso.

Desenho da chave de cache

A chave tem de capturar tudo o que influencia a resposta: o modelo, o conteúdo exato da mensagem, e parâmetros como a temperatura. Esquecer um destes campos na chave faz duas chamadas diferentes partilharem cache por engano.

import hashlib
import json

def chave_de_cache(modelo, mensagens, temperatura):
    dados = json.dumps(
        {"modelo": modelo, "mensagens": mensagens, "temperatura": temperatura},
        sort_keys=True,
    )
    return hashlib.sha256(dados.encode("utf-8")).hexdigest()

cache = {}

def chamar_com_cache(modelo, mensagens, temperatura=0):
    chave = chave_de_cache(modelo, mensagens, temperatura)
    if chave in cache:
        return cache[chave]
    resposta = client.chat.completions.create(model=modelo, messages=mensagens, temperature=temperatura)
    texto = resposta.choices[0].message.content
    cache[chave] = texto
    return texto

sort_keys=True garante que a mesma mensagem produz sempre a mesma chave, independentemente da ordem dos campos no dicionário. Tal como a fila do capítulo 3, este cache vive na memória de um processo só; em produção seria Redis ou outro cache partilhado entre réplicas, mas o desenho da chave não muda.

08SLA e Observabilidade

O que é um SLA para uma funcionalidade de IA

Um SLA (service level agreement) é uma promessa mensurável, não uma intenção vaga: "95% dos pedidos respondem em menos de 3 segundos", não "o serviço é rápido". Para uma funcionalidade de IA, o custo por pedido e mesmo uma taxa aceitável de erro fazem parte dessa promessa, tanto quanto a disponibilidade.

Logging estruturado

Em vez de print() solto, um registo estruturado (aqui, uma linha JSON) é pesquisável e agregável mais tarde, por ferramenta ou à mão.

import logging
import json
import time

logger = logging.getLogger("servico_llm")

def registar_pedido(job_id, modelo, tokens_entrada, tokens_saida, duracao_segundos, sucesso):
    logger.info(json.dumps({
        "job_id": job_id,
        "modelo": modelo,
        "tokens_entrada": tokens_entrada,
        "tokens_saida": tokens_saida,
        "duracao_segundos": round(duracao_segundos, 3),
        "sucesso": sucesso,
        "timestamp": time.time(),
    }))

Métricas básicas a acompanhar

• Contagem de pedidos, por hora ou por dia.
• Taxa de erro: quantos pedidos falharam mesmo depois dos retries.
• Latência p50 e p95, não só a média (capítulo 5).
• Custo acumulado por dia, para apanhar uma subida anormal antes da fatura chegar.

As quatro métricas acima calculam-se a partir de registos estruturados como o exemplo anterior: um script que lê o ficheiro de log e soma já é observabilidade, mesmo sem nenhuma ferramenta dedicada.

09Prática: Junta Tudo

import os
import time
import json
import hashlib
import logging
import uuid
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI, RateLimitError, APIConnectionError, APITimeoutError
import anthropic

app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
client_anthropic = anthropic.Anthropic()
logger = logging.getLogger("servico_llm")
cache = {}


class PedidoResumo(BaseModel):
    texto: str


class RespostaResumo(BaseModel):
    resumo: str


def chave_de_cache(modelo, mensagens, temperatura):
    dados = json.dumps({"modelo": modelo, "mensagens": mensagens, "temperatura": temperatura}, sort_keys=True)
    return hashlib.sha256(dados.encode("utf-8")).hexdigest()


def chamar_com_retry(mensagens, tentativas_maximas=3):
    for tentativa in range(tentativas_maximas):
        try:
            return client.chat.completions.create(model="gpt-4o-mini", messages=mensagens, timeout=10)
        except (RateLimitError, APIConnectionError, APITimeoutError):
            time.sleep(2 ** tentativa)
    raise RuntimeError("Falhou depois de todas as tentativas")


def registar_pedido(job_id, duracao_segundos, sucesso):
    logger.info(json.dumps({
        "job_id": job_id,
        "duracao_segundos": round(duracao_segundos, 3),
        "sucesso": sucesso,
        "timestamp": time.time(),
    }))


@app.get("/saude")
def saude():
    return {"estado": "ok"}


@app.post("/resumir", response_model=RespostaResumo)
def resumir(pedido: PedidoResumo):
    job_id = str(uuid.uuid4())
    inicio = time.time()
    mensagens = [{"role": "user", "content": f"Resume este texto numa frase: {pedido.texto}"}]

    chave = chave_de_cache("gpt-4o-mini", mensagens, 0)
    if chave in cache:
        registar_pedido(job_id, time.time() - inicio, sucesso=True)
        return RespostaResumo(resumo=cache[chave])

    try:
        resposta = chamar_com_retry(mensagens)
        texto = resposta.choices[0].message.content
    except RuntimeError:
        resposta_alt = client_anthropic.messages.create(
            model="claude-sonnet-5",
            max_tokens=200,
            messages=mensagens,
        )
        texto = resposta_alt.content[0].text

    cache[chave] = texto
    registar_pedido(job_id, time.time() - inicio, sucesso=True)
    return RespostaResumo(resumo=texto)