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 openai (uvicorn é o programa que corre o serviço; a openai já a conheces). 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, HTTPException
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
API_KEY = os.environ.get("OPENAI_API_KEY")
# sem a chave, o serviço arranca na mesma (o /saude e o /docs funcionam);
# só o endpoint que chama o modelo é que precisa dela, e recusa-se com
# clareza em vez de deixar rebentar um erro de autenticação cru
client = OpenAI(api_key=API_KEY or "sem-chave")

class PedidoResumo(BaseModel):
    texto: str

class RespostaResumo(BaseModel):
    resumo: str

@app.post("/resumir", response_model=RespostaResumo)
def resumir(pedido: PedidoResumo):
    if not API_KEY:
        raise HTTPException(status_code=503, detail="OPENAI_API_KEY não definida")
    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 menos, o FastAPI rebentava com um erro em vez de deixar passar uma resposta incompleta; um campo a mais é simplesmente cortado da resposta final.

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 (o \ de continuação é sintaxe bash; no Windows, escreve o comando numa linha só, ou usa o /docs):

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 te aceita 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(max_retries=0)  # ver a nota abaixo: o SDK retenta sozinho por omissão

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:
            ultima = erro
            if tentativa < tentativas_maximas - 1:
                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") from ultima

RateLimitError, APIConnectionError e APITimeoutError são classes reais da biblioteca openai, cada uma correspondendo a um tipo de falha transitória (APITimeoutError é, tecnicamente, uma subclasse de APIConnectionError; apanhar as duas é redundante mas inofensivo, e torna a intenção explícita). Apanhá-las especificamente, em vez de um Exception genérico, é o que te permite decidir que só estas vale a pena repetir. E o from ultima no raise final preserva o erro original no traceback, em vez de o esconder. Já o max_retries=0 no cliente: o SDK da openai retenta sozinho, por omissão, mais 2 vezes por baixo de cada tentativa tua; com o retry manual em cima, as "3 tentativas" viravam até 9 pedidos reais. Desligar o retry interno devolve-te o controlo (e a fatura).

O perigo de retries a mais

Isto que não é óbvio para quem vem de APIs normais: um retry de uma chamada a um LLM pode custar dinheiro outra vez. Depende de onde a falha aconteceu: um 429 rejeitado à porta não processa tokens e não é cobrado; mas um timeout do teu lado pode apanhar o servidor com a geração já concluída e cobrada, e aí o retry paga o mesmo pedido segunda vez. É por isso que retries de chamadas caras merecem um limite baixo e timeouts bem escolhidos, não um "tenta até ir".

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, HTTPException
from pydantic import BaseModel

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

class PedidoResumo(BaseModel):
    texto: str

def processar(texto):
    # chamar_com_retry é a função do capítulo 2
    resposta = chamar_com_retry([{"role": "user", "content": f"Resume: {texto}"}])
    return resposta.choices[0].message.content

def trabalhador():
    while True:
        job_id, texto = fila.get()
        try:
            resultados[job_id] = processar(texto)
        except Exception as erro:
            # sem isto, a primeira falha matava a thread em silêncio
            # e todos os jobs seguintes ficavam "a processar" para sempre
            resultados[job_id] = f"FALHOU: {erro}"

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):
    if job_id not in resultados:
        raise HTTPException(status_code=404, detail="job_id desconhecido")
    resultado = resultados[job_id]
    if resultado is None:
        return {"estado": "a processar"}
    if isinstance(resultado, str) and resultado.startswith("FALHOU:"):
        return {"estado": "falhou", "erro": resultado}
    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. Repara nos três estados possíveis da consulta: a processar, concluído, e falhou, mais um 404 para um job_id que nunca existiu; sem eles, um job que falha e um job inexistente seriam indistinguíveis de um job ainda em curso.

04Prática: o Teu Primeiro Serviço

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

app = FastAPI()
API_KEY = os.environ.get("OPENAI_API_KEY")
# sem a chave, o serviço arranca na mesma; o guard no endpoint responde 503
client = OpenAI(api_key=API_KEY or "sem-chave", max_retries=0)


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:
            ultima = erro
            if tentativa < tentativas_maximas - 1:
                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") from ultima


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


@app.post("/resumir", response_model=RespostaResumo)
def resumir(pedido: PedidoResumo):
    if not API_KEY:
        raise HTTPException(status_code=503, detail="OPENAI_API_KEY não definida")
    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 mais de cinco 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. O exemplo usa a Anthropic como alternativo: pip install anthropic, e idealmente uma ANTHROPIC_API_KEY própria (é uma conta separada, em console.anthropic.com); sem essa segunda conta, acompanhas o capítulo na mesma, e a dica do Exercício 2 mostra como simular o alternativo para testares o caminho de fallback.

from openai import OpenAI
import anthropic

# o mesmo cliente que chamar_com_retry, do capítulo 2, já usa
client = OpenAI(max_retries=0)
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ó; e nunca esquece nem expira, apesar de o próprio capítulo avisar que respostas cacheadas envelhecem. Em produção seria Redis ou outro cache partilhado entre réplicas, com limite de tamanho e TTL (tempo de vida) de graça; 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. (Em rigor, um alvo destes usado internamente chama-se SLO; SLA é quando o prometes a um cliente, com consequências se falhares. A disciplina de medir é a mesma.)

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

# sem nível e sem handler, logger.info é descartado em silêncio (o nível
# por omissão é WARNING). Configurar o NOSSO logger, e não o root com
# basicConfig, garante que só as nossas linhas vão para servico.log, sem
# misturar os logs INFO de outras bibliotecas (httpx, uvicorn)
handler = logging.FileHandler("servico.log")
handler.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("servico_llm")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

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 (a mediana: metade dos pedidos responde abaixo disto) 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, HTTPException
from pydantic import BaseModel
from openai import OpenAI, RateLimitError, APIConnectionError, APITimeoutError
import anthropic

app = FastAPI()
API_KEY = os.environ.get("OPENAI_API_KEY")
client = OpenAI(api_key=API_KEY or "sem-chave", max_retries=0)
client_anthropic = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY", "sem-chave"))

handler = logging.FileHandler("servico.log")
handler.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("servico_llm")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

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) as erro:
            ultima = erro
            if tentativa < tentativas_maximas - 1:
                time.sleep(2 ** tentativa)
    raise RuntimeError("Falhou depois de todas as tentativas") from ultima


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):
    if not API_KEY:
        raise HTTPException(status_code=503, detail="OPENAI_API_KEY não definida")

    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])

    sucesso = False
    try:
        try:
            resposta = chamar_com_retry(mensagens)
            texto = resposta.choices[0].message.content
            # só o caminho principal entra no cache: a resposta do alternativo
            # é de outro modelo, e a chave diz gpt-4o-mini (capítulo 7)
            cache[chave] = texto
        except RuntimeError:
            try:
                resposta_alt = client_anthropic.messages.create(
                    model="claude-sonnet-5",
                    max_tokens=200,
                    messages=mensagens,
                )
                texto = resposta_alt.content[0].text
            except Exception as erro:
                raise HTTPException(status_code=502, detail="ambos os fornecedores falharam") from erro
        sucesso = True
        return RespostaResumo(resumo=texto)
    finally:
        # o finally corre sempre, com sucesso ou falha: é o que garante o
        # requisito de uma linha de log por pedido, aconteça o que acontecer
        registar_pedido(job_id, time.time() - inicio, sucesso=sucesso)