FastAPI para Dados e IA: Do Zero à Produção com OpenAI e Groq
Aprenda a criar APIs modernas com FastAPI para projetos de Dados e Inteligência Artificial. Tutorial completo com exemplos práticos de integração com OpenAI e Groq.
Introdução
Se você trabalha com Dados ou Inteligência Artificial, provavelmente já enfrentou o desafio de expor seus modelos e pipelines como serviços acessíveis. É aqui que o FastAPI se destaca como uma das ferramentas mais poderosas do ecossistema Python.
O FastAPI combina performance excepcional, tipagem moderna e documentação automática em um framework que simplesmente funciona. Não é à toa que se tornou a escolha preferida para servir modelos de ML, criar APIs para LLMs e construir backends de aplicações inteligentes.
Neste artigo, vamos explorar o FastAPI do zero, passando pela instalação, configuração e chegando a exemplos práticos de integração com OpenAI e Groq. Ao final, você terá uma base sólida para criar suas próprias APIs de IA.
Por que FastAPI para Dados e IA?
Antes de mergulharmos no código, vale entender o que torna o FastAPI especial para nosso contexto:
Performance Assíncrona: Construído sobre Starlette e UVICORN, o FastAPI suporta operações assíncronas nativamente. Isso é crucial quando suas APIs precisam fazer chamadas a serviços externos como OpenAI ou Groq, permitindo lidar com múltiplas requisições simultâneas sem bloquear o servidor.
Validação Automática com Pydantic: A integração com Pydantic significa que seus dados de entrada e saída são validados automaticamente. Em projetos de IA, onde a qualidade dos dados é fundamental, isso evita erros silenciosos e facilita o debugging.
Documentação Interativa: O FastAPI gera automaticamente documentação Swagger UI e ReDoc. Seus colegas de equipe (e você mesmo no futuro) agradecerão por poder testar endpoints diretamente no navegador.
Type Hints Nativos: O uso de type hints do Python não é apenas decorativo. O FastAPI usa essas anotações para validação, serialização e documentação, criando um ciclo virtuoso de código limpo e funcional.
Instalação e Configuração Inicial
Vamos começar configurando nosso ambiente de desenvolvimento. Recomendo sempre usar ambientes virtuais para isolar as dependências do projeto.
Criando o Ambiente
# Criar diretório do projeto
mkdir fastapi-ia-demo
cd fastapi-ia-demo
# Criar e ativar ambiente virtual
python -m venv venv
source venv/bin/activate # Linux/Mac
# ou
venv\Scripts\activate # Windows
Instalando Dependências
# FastAPI e servidor ASGI
pip install fastapi uvicorn[standard]
# Dependências para IA
pip install openai groq
# Utilitários
pip install python-dotenv pydantic-settings
Estrutura do Projeto
Para projetos mais organizados, sugiro a seguinte estrutura:
fastapi-ia-demo/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── models/
│ │ └── schemas.py
│ └── routers/
│ ├── openai_router.py
│ └── groq_router.py
├── .env
├── .gitignore
└── requirements.txt
Configuração de Variáveis de Ambiente
Crie um arquivo .env na raiz do projeto para armazenar suas chaves de API:
OPENAI_API_KEY=sua-chave-openai-aqui
GROQ_API_KEY=sua-chave-groq-aqui
E o arquivo config.py para carregar essas configurações:
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
app_name: str = "FastAPI IA Demo"
openai_api_key: str
groq_api_key: str
class Config:
env_file = ".env"
@lru_cache
def get_settings():
return Settings()
O decorator @lru_cache garante que as configurações sejam carregadas apenas uma vez, otimizando a performance.
Sua Primeira API com FastAPI
Vamos criar nossa aplicação principal. Este será o ponto de entrada da API:
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(
title="FastAPI para Dados e IA",
description="API demonstrando integrações com OpenAI e Groq",
version="1.0.0"
)
# Configurar CORS para permitir requisições do frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def root():
return {"message": "API de IA funcionando!", "status": "healthy"}
@app.get("/health")
async def health_check():
return {"status": "ok"}
Para executar a aplicação:
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
Acesse http://localhost:8000/docs para ver a documentação interativa gerada automaticamente.
Definindo Schemas com Pydantic
Antes de criar os endpoints de IA, vamos definir os schemas que validarão nossas requisições e respostas:
# app/models/schemas.py
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum
class ModelProvider(str, Enum):
OPENAI = "openai"
GROQ = "groq"
class ChatMessage(BaseModel):
role: str = Field(..., description="Papel da mensagem: system, user ou assistant")
content: str = Field(..., description="Conteúdo da mensagem")
class ChatRequest(BaseModel):
messages: list[ChatMessage] = Field(..., description="Lista de mensagens do chat")
model: Optional[str] = Field(None, description="Modelo específico a ser usado")
temperature: float = Field(0.7, ge=0, le=2, description="Criatividade das respostas")
max_tokens: int = Field(1000, ge=1, le=4000, description="Máximo de tokens na resposta")
class ChatResponse(BaseModel):
content: str = Field(..., description="Resposta gerada pelo modelo")
model: str = Field(..., description="Modelo utilizado")
provider: str = Field(..., description="Provedor do modelo")
usage: dict = Field(..., description="Informações de uso de tokens")
class EmbeddingRequest(BaseModel):
text: str = Field(..., description="Texto para gerar embedding")
class EmbeddingResponse(BaseModel):
embedding: list[float] = Field(..., description="Vetor de embedding")
dimensions: int = Field(..., description="Dimensões do vetor")
Os schemas do Pydantic garantem que dados inválidos sejam rejeitados automaticamente com mensagens de erro claras, antes mesmo de chegarem à lógica do seu código.
Integração com OpenAI
Agora vamos criar os endpoints para interagir com a API da OpenAI:
# app/routers/openai_router.py
from fastapi import APIRouter, HTTPException, Depends
from openai import OpenAI
from app.config import get_settings, Settings
from app.models.schemas import (
ChatRequest,
ChatResponse,
EmbeddingRequest,
EmbeddingResponse
)
router = APIRouter(prefix="/openai", tags=["OpenAI"])
def get_openai_client(settings: Settings = Depends(get_settings)):
return OpenAI(api_key=settings.openai_api_key)
@router.post("/chat", response_model=ChatResponse)
async def openai_chat(
request: ChatRequest,
client: OpenAI = Depends(get_openai_client)
):
"""
Endpoint para chat completion usando modelos da OpenAI.
"""
try:
model = request.model or "gpt-4o-mini"
response = client.chat.completions.create(
model=model,
messages=[msg.model_dump() for msg in request.messages],
temperature=request.temperature,
max_tokens=request.max_tokens
)
return ChatResponse(
content=response.choices[0].message.content,
model=response.model,
provider="openai",
usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/embeddings", response_model=EmbeddingResponse)
async def openai_embeddings(
request: EmbeddingRequest,
client: OpenAI = Depends(get_openai_client)
):
"""
Gera embeddings de texto usando text-embedding-3-small da OpenAI.
Útil para sistemas de busca semântica e RAG.
"""
try:
response = client.embeddings.create(
model="text-embedding-3-small",
input=request.text
)
embedding = response.data[0].embedding
return EmbeddingResponse(
embedding=embedding,
dimensions=len(embedding)
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
O uso de Depends para injeção de dependências torna o código testável e facilita a troca de implementações (por exemplo, usar um mock durante testes).
Integração com Groq
O Groq oferece inferência ultrarrápida para modelos open-source como Llama e Mixtral. A integração segue um padrão similar:
# app/routers/groq_router.py
from fastapi import APIRouter, HTTPException, Depends
from groq import Groq
from app.config import get_settings, Settings
from app.models.schemas import ChatRequest, ChatResponse
router = APIRouter(prefix="/groq", tags=["Groq"])
def get_groq_client(settings: Settings = Depends(get_settings)):
return Groq(api_key=settings.groq_api_key)
@router.post("/chat", response_model=ChatResponse)
async def groq_chat(
request: ChatRequest,
client: Groq = Depends(get_groq_client)
):
"""
Endpoint para chat completion usando modelos do Groq.
Oferece inferência ultrarrápida com modelos como Llama e Mixtral.
"""
try:
# Modelos disponíveis no Groq
model = request.model or "llama-3.1-70b-versatile"
response = client.chat.completions.create(
model=model,
messages=[msg.model_dump() for msg in request.messages],
temperature=request.temperature,
max_tokens=request.max_tokens
)
return ChatResponse(
content=response.choices[0].message.content,
model=response.model,
provider="groq",
usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/models")
async def list_groq_models():
"""
Lista os modelos mais populares disponíveis no Groq.
"""
return {
"models": [
{"id": "llama-3.1-70b-versatile", "description": "Llama 3.1 70B - Melhor equilíbrio"},
{"id": "llama-3.1-8b-instant", "description": "Llama 3.1 8B - Mais rápido"},
{"id": "mixtral-8x7b-32768", "description": "Mixtral 8x7B - Contexto longo"},
{"id": "gemma2-9b-it", "description": "Gemma 2 9B - Google"}
]
}
Registrando os Routers
Atualize o main.py para incluir os routers:
# app/main.py (atualizado)
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import openai_router, groq_router
app = FastAPI(
title="FastAPI para Dados e IA",
description="API demonstrando integrações com OpenAI e Groq",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Registrar routers
app.include_router(openai_router.router)
app.include_router(groq_router.router)
@app.get("/")
async def root():
return {"message": "API de IA funcionando!", "status": "healthy"}
@app.get("/health")
async def health_check():
return {"status": "ok"}
Streaming de Respostas
Para aplicações de chat em tempo real, o streaming de respostas é essencial. Veja como implementar:
# Adicionar ao openai_router.py
from fastapi.responses import StreamingResponse
import json
@router.post("/chat/stream")
async def openai_chat_stream(
request: ChatRequest,
client: OpenAI = Depends(get_openai_client)
):
"""
Endpoint com streaming para respostas em tempo real.
Ideal para interfaces de chat.
"""
async def generate():
try:
model = request.model or "gpt-4o-mini"
stream = client.chat.completions.create(
model=model,
messages=[msg.model_dump() for msg in request.messages],
temperature=request.temperature,
max_tokens=request.max_tokens,
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
yield f"data: {json.dumps({'content': content})}\n\n"
yield "data: [DONE]\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream"
)
Endpoint Unificado: Escolha seu Provider
Para facilitar o uso, podemos criar um endpoint que abstrai a escolha do provedor:
# Adicionar ao main.py ou criar novo router
from app.models.schemas import ModelProvider
@app.post("/chat/unified", response_model=ChatResponse)
async def unified_chat(
request: ChatRequest,
provider: ModelProvider = ModelProvider.OPENAI
):
"""
Endpoint unificado que permite escolher entre OpenAI e Groq.
"""
settings = get_settings()
if provider == ModelProvider.OPENAI:
client = OpenAI(api_key=settings.openai_api_key)
model = request.model or "gpt-4o-mini"
else:
client = Groq(api_key=settings.groq_api_key)
model = request.model or "llama-3.1-70b-versatile"
response = client.chat.completions.create(
model=model,
messages=[msg.model_dump() for msg in request.messages],
temperature=request.temperature,
max_tokens=request.max_tokens
)
return ChatResponse(
content=response.choices[0].message.content,
model=response.model,
provider=provider.value,
usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
)
Testando a API
Com a aplicação rodando, você pode testar usando curl ou a documentação interativa:
# Teste básico de saúde
curl http://localhost:8000/health
# Chat com OpenAI
curl -X POST http://localhost:8000/openai/chat \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "system", "content": "Você é um assistente especializado em dados."},
{"role": "user", "content": "O que é um Data Lake?"}
],
"temperature": 0.7
}'
# Chat com Groq
curl -X POST http://localhost:8000/groq/chat \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "user", "content": "Explique Apache Spark em 3 frases."}
],
"model": "llama-3.1-8b-instant"
}'
# Gerar embeddings
curl -X POST http://localhost:8000/openai/embeddings \
-H "Content-Type: application/json" \
-d '{
"text": "FastAPI é um framework moderno para construir APIs"
}'
Boas Práticas para Produção
Antes de colocar sua API em produção, considere as seguintes práticas:
Rate Limiting: Proteja sua API contra abusos usando bibliotecas como slowapi:
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@router.post("/chat")
@limiter.limit("10/minute")
async def chat(request: Request, ...):
...
Logging Estruturado: Use logging adequado para monitoramento:
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@router.post("/chat")
async def chat(request: ChatRequest, ...):
logger.info(f"Chat request received - model: {request.model}")
# ... resto do código
Tratamento de Erros Global: Configure handlers para exceções:
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled error: {exc}")
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"}
)
Cache de Respostas: Para respostas determinísticas, considere usar cache:
from functools import lru_cache
# ou redis para ambientes distribuídos
Conclusão
O FastAPI provou ser uma escolha excepcional para projetos de Dados e IA. Sua combinação de performance, tipagem forte e documentação automática acelera o desenvolvimento sem sacrificar a qualidade.
Neste artigo, cobrimos desde a instalação básica até integrações práticas com OpenAI e Groq. Os conceitos apresentados aqui servem como fundação para projetos mais complexos, como sistemas RAG completos, pipelines de processamento de dados em tempo real ou APIs de inferência para modelos customizados.
Os próximos passos naturais incluem adicionar autenticação (JWT ou API Keys), implementar cache com Redis, configurar CI/CD para deploy automatizado e adicionar testes automatizados com pytest.
O ecossistema FastAPI continua evoluindo, e a comunidade está sempre trazendo novas integrações e padrões. Mantenha-se atualizado e não hesite em experimentar.
Recursos Adicionais
Tem dúvidas ou sugestões? Deixe seu comentário abaixo ou me encontre no LinkedIn. Vamos trocar ideias sobre Dados e IA!
