📚 Tutoriais•11 min de leitura
Como Criar uma API REST com FastAPI e IA Integrada
Guia completo para construir uma API REST profissional com FastAPI, integrando modelos de IA para classificação de texto, geração de conteúdo e análise de sentimento.
#fastapi#api#python#ia#rest
Introdução
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.
Neste tutorial, vamos construir uma API REST completa que integra múltiplos serviços de IA: classificação de texto, análise de sentimento e geração de conteúdo. Ao final, você terá uma aplicação pronta para deploy em produção.
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
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"
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