Como Criar uma API REST com FastAPI e IA Integrada

Introdução

Depois de anos usando Flask para expor modelos de ML, migrei para FastAPI e não volto atrás. A diferença em produtividade é absurda — validação automática de dados, documentação interativa gerada sem esforço, suporte nativo a async. Hoje uso FastAPI como padrão para qualquer serviço de IA que precisa ser consumido por outros sistemas.

Neste tutorial, vou mostrar como eu estruturo uma API de IA do zero, com os padrões que aprendi trabalhando em projetos reais: separação de serviços, configurações via variáveis de ambiente, e tratamento de erros que fazem sentido. Ao final, você terá uma aplicação completa e pronta para deploy. Se você ainda não conhece os modelos de IA que vamos expor aqui, o tutorial de fine-tuning com Hugging Face dá uma boa base.

Construir APIs REST é uma habilidade fundamental para qualquer desenvolvedor que trabalha com Inteligência Artificial. Não basta treinar um modelo — é preciso disponibilizá-lo para que outras aplicações possam consumi-lo. O FastAPI é o framework Python mais moderno e performático para essa tarefa, oferecendo tipagem automática, validação de dados, documentação interativa e suporte nativo a operações assíncronas.

Pré-requisitos

Antes de começar, instale as dependências necessárias:

# requirements.txt
fastapi==0.109.0
uvicorn[standard]==0.27.0
pydantic==2.5.3
pydantic-settings==2.1.0
transformers==4.37.0
torch==2.1.2
openai==1.12.0
python-dotenv==1.0.0
httpx==0.26.0

Instale tudo com:

pip install -r requirements.txt

1. Estrutura do Projeto

Vamos organizar o projeto seguindo boas práticas de arquitetura:

minha-api-ia/
├── app/
│   ├── __init__.py
│   ├── main.py              # Ponto de entrada da aplicação
│   ├── config.py             # Configurações e variáveis de ambiente
│   ├── models/
│   │   ├── __init__.py
│   │   └── schemas.py        # Schemas Pydantic para validação
│   ├── services/
│   │   ├── __init__.py
│   │   ├── sentiment.py      # Serviço de análise de sentimento
│   │   ├── classifier.py     # Serviço de classificação de texto
│   │   └── generator.py      # Serviço de geração de conteúdo
│   └── routers/
│       ├── __init__.py
│       ├── sentiment.py       # Rotas de sentimento
│       ├── classifier.py      # Rotas de classificação
│       └── generator.py       # Rotas de geração
├── tests/
│   ├── test_sentiment.py
│   └── test_classifier.py
├── requirements.txt
├── .env
└── Dockerfile

2. Configuração Base

Primeiro, vamos criar o arquivo de configuração:

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """Configurações da aplicação carregadas de variáveis de ambiente."""
    
    app_name: str = "API de IA - Inteligência em Código"
    app_version: str = "1.0.0"
    debug: bool = False
    
    # API Keys
    openai_api_key: str = ""
    
    # Configurações do modelo local
    sentiment_model: str = "nlptown/bert-base-multilingual-uncased-sentiment"
    classifier_model: str = "facebook/bart-large-mnli"
    
    # Rate limiting
    max_requests_per_minute: int = 60
    max_text_length: int = 5000
    
    class Config:
        env_file = ".env"


@lru_cache
def get_settings() -> Settings:
    """Retorna instância cacheada das configurações."""
    return Settings()

3. Schemas de Validação com Pydantic

O FastAPI usa Pydantic para validação automática. Vamos definir nossos schemas:

# app/models/schemas.py
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional


class SentimentEnum(str, Enum):
    POSITIVO = "positivo"
    NEGATIVO = "negativo"
    NEUTRO = "neutro"


class TextInput(BaseModel):
    """Schema de entrada para análise de texto."""
    texto: str = Field(
        ...,
        min_length=1,
        max_length=5000,
        description="Texto a ser analisado",
        examples=["Adorei esse produto! Funciona perfeitamente."]
    )
    idioma: str = Field(
        default="pt",
        description="Código do idioma (pt, en, es)"
    )


class SentimentResponse(BaseModel):
    """Resposta da análise de sentimento."""
    texto: str
    sentimento: SentimentEnum
    confianca: float = Field(ge=0, le=1)
    detalhes: dict


class ClassificationInput(BaseModel):
    """Entrada para classificação de texto."""
    texto: str = Field(..., min_length=1, max_length=5000)
    categorias: list[str] = Field(
        ...,
        min_length=2,
        max_length=20,
        description="Lista de categorias possíveis",
        examples=[["tecnologia", "esportes", "política", "entretenimento"]]
    )


class ClassificationResponse(BaseModel):
    """Resposta da classificação."""
    texto: str
    categoria_principal: str
    scores: dict[str, float]


class GenerationInput(BaseModel):
    """Entrada para geração de conteúdo."""
    prompt: str = Field(..., min_length=10, max_length=2000)
    max_tokens: int = Field(default=500, ge=50, le=4000)
    temperatura: float = Field(default=0.7, ge=0, le=2)
    estilo: Optional[str] = Field(
        default=None,
        description="Estilo do texto: formal, informal, técnico, criativo"
    )


class GenerationResponse(BaseModel):
    """Resposta da geração de conteúdo."""
    prompt_original: str
    texto_gerado: str
    tokens_usados: int
    modelo: str


class HealthResponse(BaseModel):
    """Resposta do health check."""
    status: str
    version: str
    modelos_carregados: list[str]


class BatchInput(BaseModel):
    """Entrada para processamento em lote."""
    textos: list[str] = Field(..., min_length=1, max_length=100)


class ErrorResponse(BaseModel):
    """Schema padrão de erro."""
    detail: str
    status_code: int

4. Serviço de Análise de Sentimento

Agora vamos criar o serviço que carrega e utiliza o modelo de sentimento:

# app/services/sentiment.py
from transformers import pipeline
from app.config import get_settings
import logging

logger = logging.getLogger(__name__)


class SentimentService:
    """Serviço de análise de sentimento usando Transformers."""
    
    def __init__(self):
        self._pipeline = None
    
    def load_model(self):
        """Carrega o modelo de sentimento na memória."""
        settings = get_settings()
        logger.info(f"Carregando modelo de sentimento: {settings.sentiment_model}")
        
        self._pipeline = pipeline(
            "sentiment-analysis",
            model=settings.sentiment_model,
            device=-1  # CPU; use 0 para GPU
        )
        
        logger.info("Modelo de sentimento carregado com sucesso!")
    
    def analyze(self, texto: str) -> dict:
        """
        Analisa o sentimento de um texto.
        
        Args:
            texto: Texto para análise
            
        Returns:
            Dicionário com sentimento, confiança e detalhes
        """
        if self._pipeline is None:
            self.load_model()
        
        resultado = self._pipeline(texto)[0]
        
        # O modelo retorna estrelas (1-5), convertemos para sentimento
        label = resultado["label"]
        score = resultado["score"]
        
        # Extrair número de estrelas
        try:
            stars = int(label.split()[0])
        except (ValueError, IndexError):
            stars = 3
        
        # Mapear para sentimento
        if stars >= 4:
            sentimento = "positivo"
        elif stars <= 2:
            sentimento = "negativo"
        else:
            sentimento = "neutro"
        
        return {
            "sentimento": sentimento,
            "confianca": round(score, 4),
            "detalhes": {
                "label_original": label,
                "score_original": score,
                "estrelas": stars
            }
        }
    
    def analyze_batch(self, textos: list[str]) -> list[dict]:
        """Analisa sentimento de múltiplos textos."""
        return [self.analyze(texto) for texto in textos]


# Instância global (singleton)
sentiment_service = SentimentService()

5. Serviço de Classificação Zero-Shot

# app/services/classifier.py
from transformers import pipeline
from app.config import get_settings
import logging

logger = logging.getLogger(__name__)


class ClassifierService:
    """Classificação zero-shot usando BART/NLI."""
    
    def __init__(self):
        self._pipeline = None
    
    def load_model(self):
        """Carrega o modelo de classificação."""
        settings = get_settings()
        logger.info(f"Carregando modelo de classificação: {settings.classifier_model}")
        
        self._pipeline = pipeline(
            "zero-shot-classification",
            model=settings.classifier_model,
            device=-1
        )
        
        logger.info("Modelo de classificação carregado!")
    
    def classify(self, texto: str, categorias: list[str]) -> dict:
        """
        Classifica um texto nas categorias fornecidas.
        
        Args:
            texto: Texto para classificar
            categorias: Lista de categorias possíveis
            
        Returns:
            Dicionário com categoria principal e scores
        """
        if self._pipeline is None:
            self.load_model()
        
        resultado = self._pipeline(
            texto,
            candidate_labels=categorias,
            hypothesis_template="Este texto é sobre {}."
        )
        
        # Montar dicionário de scores
        scores = {}
        for label, score in zip(resultado["labels"], resultado["scores"]):
            scores[label] = round(score, 4)
        
        return {
            "categoria_principal": resultado["labels"][0],
            "scores": scores
        }


classifier_service = ClassifierService()

6. Serviço de Geração com OpenAI

# app/services/generator.py
from openai import OpenAI, AsyncOpenAI
from app.config import get_settings
import logging

logger = logging.getLogger(__name__)


class GeneratorService:
    """Geração de texto usando API da OpenAI."""
    
    def __init__(self):
        self._client = None
    
    def _get_client(self) -> OpenAI:
        """Retorna cliente OpenAI configurado."""
        if self._client is None:
            settings = get_settings()
            self._client = OpenAI(api_key=settings.openai_api_key)
        return self._client
    
    def generate(
        self,
        prompt: str,
        max_tokens: int = 500,
        temperatura: float = 0.7,
        estilo: str | None = None
    ) -> dict:
        """
        Gera texto com base no prompt fornecido.
        
        Args:
            prompt: Texto de entrada para geração
            max_tokens: Máximo de tokens na resposta
            temperatura: Criatividade (0=determinístico, 2=muito criativo)
            estilo: Estilo do texto gerado
            
        Returns:
            Dicionário com texto gerado e metadados
        """
        client = self._get_client()
        
        system_message = "Você é um assistente especializado em criar conteúdo de alta qualidade em português brasileiro."
        
        if estilo:
            estilos_map = {
                "formal": "Escreva de forma formal e profissional.",
                "informal": "Escreva de forma informal e descontraída.",
                "técnico": "Escreva de forma técnica com terminologia especializada.",
                "criativo": "Escreva de forma criativa e envolvente."
            }
            system_message += f" {estilos_map.get(estilo, '')}"
        
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_message},
                {"role": "user", "content": prompt}
            ],
            max_tokens=max_tokens,
            temperature=temperatura
        )
        
        return {
            "texto_gerado": response.choices[0].message.content,
            "tokens_usados": response.usage.total_tokens,
            "modelo": response.model
        }


generator_service = GeneratorService()

7. Rotas da API

Agora criamos as rotas organizadas por funcionalidade:

# app/routers/sentiment.py
from fastapi import APIRouter, HTTPException
from app.models.schemas import (
    TextInput, SentimentResponse, BatchInput, ErrorResponse
)
from app.services.sentiment import sentiment_service

router = APIRouter(prefix="/api/v1/sentiment", tags=["Análise de Sentimento"])


@router.post(
    "/analyze",
    response_model=SentimentResponse,
    summary="Analisar sentimento de um texto",
    responses={422: {"model": ErrorResponse}}
)
async def analyze_sentiment(input_data: TextInput):
    """
    Analisa o sentimento de um texto em português, inglês ou espanhol.
    
    Retorna classificação (positivo, negativo ou neutro) com score de confiança.
    """
    try:
        resultado = sentiment_service.analyze(input_data.texto)
        return SentimentResponse(
            texto=input_data.texto,
            sentimento=resultado["sentimento"],
            confianca=resultado["confianca"],
            detalhes=resultado["detalhes"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erro na análise: {str(e)}")


@router.post("/batch", summary="Analisar sentimento em lote")
async def analyze_batch(input_data: BatchInput):
    """Analisa o sentimento de múltiplos textos de uma vez."""
    resultados = []
    for texto in input_data.textos:
        resultado = sentiment_service.analyze(texto)
        resultados.append({
            "texto": texto,
            "sentimento": resultado["sentimento"],
            "confianca": resultado["confianca"]
        })
    return {"resultados": resultados, "total": len(resultados)}

# app/routers/classifier.py
from fastapi import APIRouter, HTTPException
from app.models.schemas import ClassificationInput, ClassificationResponse
from app.services.classifier import classifier_service

router = APIRouter(prefix="/api/v1/classifier", tags=["Classificação de Texto"])


@router.post(
    "/classify",
    response_model=ClassificationResponse,
    summary="Classificar texto em categorias"
)
async def classify_text(input_data: ClassificationInput):
    """
    Classifica um texto nas categorias fornecidas usando zero-shot classification.
    
    Não requer treinamento prévio — basta fornecer as categorias desejadas.
    """
    try:
        resultado = classifier_service.classify(
            input_data.texto,
            input_data.categorias
        )
        return ClassificationResponse(
            texto=input_data.texto,
            categoria_principal=resultado["categoria_principal"],
            scores=resultado["scores"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erro na classificação: {str(e)}")

# app/routers/generator.py
from fastapi import APIRouter, HTTPException
from app.models.schemas import GenerationInput, GenerationResponse
from app.services.generator import generator_service

router = APIRouter(prefix="/api/v1/generator", tags=["Geração de Conteúdo"])


@router.post(
    "/generate",
    response_model=GenerationResponse,
    summary="Gerar texto com IA"
)
async def generate_text(input_data: GenerationInput):
    """
    Gera texto usando IA com base no prompt fornecido.
    
    Suporta diferentes estilos: formal, informal, técnico, criativo.
    """
    try:
        resultado = generator_service.generate(
            prompt=input_data.prompt,
            max_tokens=input_data.max_tokens,
            temperatura=input_data.temperatura,
            estilo=input_data.estilo
        )
        return GenerationResponse(
            prompt_original=input_data.prompt,
            texto_gerado=resultado["texto_gerado"],
            tokens_usados=resultado["tokens_usados"],
            modelo=resultado["modelo"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erro na geração: {str(e)}")

8. Aplicação Principal

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import get_settings
from app.routers import sentiment, classifier, generator
from app.services.sentiment import sentiment_service
from app.services.classifier import classifier_service
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Gerencia o ciclo de vida da aplicação."""
    logger.info("🚀 Iniciando API de IA...")
    
    # Carregar modelos na inicialização
    sentiment_service.load_model()
    classifier_service.load_model()
    
    logger.info("✅ Todos os modelos carregados!")
    yield
    
    logger.info("🛑 Encerrando API...")


settings = get_settings()

app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    description="API REST com serviços de IA: análise de sentimento, classificação e geração de texto.",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc"
)

# CORS para permitir acesso de frontends
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Registrar rotas
app.include_router(sentiment.router)
app.include_router(classifier.router)
app.include_router(generator.router)


@app.get("/", tags=["Root"])
async def root():
    """Endpoint raiz com informações da API."""
    return {
        "nome": settings.app_name,
        "versao": settings.app_version,
        "documentacao": "/docs",
        "endpoints": {
            "sentimento": "/api/v1/sentiment/analyze",
            "classificacao": "/api/v1/classifier/classify",
            "geracao": "/api/v1/generator/generate"
        }
    }


@app.get("/health", tags=["Health"])
async def health_check():
    """Verifica se a API está funcionando corretamente."""
    return {
        "status": "healthy",
        "version": settings.app_version,
        "modelos_carregados": ["sentiment", "classifier"]
    }

9. Dockerfile para Deploy

# Dockerfile
# FROM python:3.11-slim
#
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt
#
# COPY app/ app/
# COPY .env .env
#
# EXPOSE 8000
# CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

10. Testando a API

Execute a aplicação:

uvicorn app.main:app --reload --port 8000

Agora teste com Python usando httpx:

import httpx

BASE_URL = "http://localhost:8000"

# Teste 1: Análise de Sentimento
response = httpx.post(f"{BASE_URL}/api/v1/sentiment/analyze", json={
    "texto": "Este produto é maravilhoso! Recomendo a todos.",
    "idioma": "pt"
})
print("Sentimento:", response.json())

# Teste 2: Classificação de Texto
response = httpx.post(f"{BASE_URL}/api/v1/classifier/classify", json={
    "texto": "O novo processador da Intel usa arquitetura avançada de 2nm.",
    "categorias": ["tecnologia", "esportes", "política", "saúde"]
})
print("Classificação:", response.json())

# Teste 3: Geração de Conteúdo
response = httpx.post(f"{BASE_URL}/api/v1/generator/generate", json={
    "prompt": "Escreva uma introdução sobre redes neurais convolucionais.",
    "max_tokens": 300,
    "temperatura": 0.7,
    "estilo": "técnico"
})
print("Geração:", response.json())

# Teste 4: Análise em lote
response = httpx.post(f"{BASE_URL}/api/v1/sentiment/batch", json={
    "textos": [
        "Excelente experiência de compra!",
        "Produto terrível, chegou com defeito.",
        "Entrega no prazo, produto dentro do esperado."
    ]
})
print("Lote:", response.json())

11. Testes Automatizados

# tests/test_sentiment.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_analyze_positive():
    """Testa análise de texto positivo."""
    response = client.post("/api/v1/sentiment/analyze", json={
        "texto": "Adorei esse produto! Funciona perfeitamente."
    })
    assert response.status_code == 200
    data = response.json()
    assert data["sentimento"] == "positivo"
    assert 0 <= data["confianca"] <= 1


def test_analyze_negative():
    """Testa análise de texto negativo."""
    response = client.post("/api/v1/sentiment/analyze", json={
        "texto": "Péssimo produto, não funciona de jeito nenhum."
    })
    assert response.status_code == 200
    data = response.json()
    assert data["sentimento"] == "negativo"


def test_empty_text_rejected():
    """Texto vazio deve retornar erro 422."""
    response = client.post("/api/v1/sentiment/analyze", json={
        "texto": ""
    })
    assert response.status_code == 422


def test_health_check():
    """Health check deve retornar status healthy."""
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Resultado Esperado

Após subir o servidor com uvicorn app.main:app --reload, acesse http://localhost:8000/docs e você verá a documentação interativa gerada automaticamente. Ao chamar o endpoint de análise de sentimento via curl ou pelo Swagger UI, a resposta se parece com isso:

{
  "text": "O FastAPI é incrível para construir APIs de IA!",
  "sentiment": "POSITIVE",
  "score": 0.9823,
  "processing_time_ms": 45
}

O endpoint de geração de texto responderá:

{
  "prompt": "Explique o que é machine learning em uma frase",
  "generated_text": "Machine learning é um subcampo da IA onde sistemas aprendem padrões a partir de dados para fazer previsões ou tomar decisões sem serem explicitamente programados.",
  "model": "gpt-4o-mini",
  "tokens_used": 42
}

E o health check confirma que todos os serviços estão operacionais:

{
  "status": "healthy",
  "services": {
    "sentiment_model": "loaded",
    "classifier_model": "loaded",
    "openai_client": "connected"
  },
  "uptime_seconds": 137
}

Conclusão

Neste tutorial, construímos uma API REST completa com FastAPI que integra três serviços de IA diferentes: análise de sentimento com Transformers, classificação zero-shot e geração de texto com OpenAI. A arquitetura modular facilita a manutenção e a adição de novos endpoints.

Os pontos-chave que abordamos foram:

• Estruturação profissional do projeto com separação em camadas (routers, services, models)
• Validação automática de dados com Pydantic e schemas tipados
• Carregamento inteligente de modelos com lifecycle management do FastAPI
• Documentação automática via Swagger UI e ReDoc
• Testes automatizados com TestClient do FastAPI

Próximos Passos

• Adicionar autenticação com JWT ou API Keys
• Implementar rate limiting com SlowAPI
• Configurar cache com Redis para respostas frequentes
• Adicionar monitoramento com Prometheus e Grafana
• Deploy na AWS Lambda com Mangum ou no Docker com Kubernetes
• Implementar streaming de respostas para geração de texto longo