Este tutorial detalha a criação de uma API robusta para integarr funcionalidades de busca semântica e geração de texto. Utilizaremos dois modelos de IA de ponta: GTE-Chinese-Large para compreensão de linguagem natural em chinês e SeqGPT-560m para geração de texto. O objetivo é expor essas capacidades através de interfaces HTTP padronizadas, facilitando sua integração em diversas aplicações.
Imagine um sistema capaz de entender consultas em linguagem natural, encontrar informações relevantes em uma base de conhecimento e até mesmo gerar conteúdo novo com base nesses achados. Essa é a funcionalidade que vamos implementar, utilizando o framework FastAPI para construir uma API limpa e eficiente.
Configuração do Ambiente e Instalação de Dependências
Requisitos do Sistema
- Versão do Python: 3.8 ou superior (recomendado 3.11+).
- Sistema Operacional: Linux, macOS ou Windows (WSL recomendado para Windows).
- Memória RAM: Mínimo de 8GB (16GB recomendado).
- Espaço em Disco: Mínimo de 10GB para arquivos de modelo.
Instalação das Dependências
É recomendado o uso de um ambiente virtual. Após a ativação, instale os pacotes necessários:
# Criação do ambiente virtual
python -m venv ai_api_env
source ai_api_env/bin/activate # Linux/macOS
# Ou
ai_api_env\Scripts\activate # Windows
# Instalação das dependências principais
pip install fastapi uvicorn transformers modelscope sentence-transformers
pip install "pydantic>=2.0" "python-multipart>=0.0.6"
Download e Configuração dos Modelos
Os modelos GTE e SeqGPT precisam ser baixados. O ModelScope facilita este processo, especialmente para usuários na China:
from modelscope import snapshot_download
# Baixa o modelo de embedding GTE
model_embedding_path = snapshot_download('iic/nlp_gte_sentence-embedding_chinese-large')
# Baixa o modelo de geração de texto SeqGPT
model_generation_path = snapshot_download('iic/nlp_seqgpt-560m')
Os modelos são geralmente salvos em ~/.cache/modelscope/hub/. Certifique-se de ter espaço em disco suficiente.
Desenvolvimento da API com FastAPI
Estrutura Básica da API
Crie o arquivo principle da aplicação FastAPI (main.py):
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any
import numpy as np
from transformers import AutoModel, AutoTokenizer
import torch
import asyncio
from functools import lru_cache
# Inicialização da aplicação FastAPI
app = FastAPI(
title="API de IA para Busca Semântica e Geração de Texto",
description="Serviço unificado para busca semântica e geração de texto.",
version="1.0.0"
)
# --- Definição dos Modelos de Dados (Pydantic) ---
class SearchQuery(BaseModel):
query_text: str
documents: List[str]
k_results: int = 3
class SearchResult(BaseModel):
document: str
similarity_score: float
original_index: int
class TextGenerationInput(BaseModel):
prompt_text: str
max_output_length: int = 100
class GenerationOutput(BaseModel):
generated_content: str
# --- Carregamento Otimizado dos Modelos ---
@lru_cache(maxsize=1)
def load_embedding_model():
"""Carrega o modelo GTE para embeddings."""
try:
model_name = 'iic/nlp_gte_sentence-embedding_chinese-large'
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name)
# Opcional: Mover para GPU se disponível
# if torch.cuda.is_available():
# model.to('cuda')
print("Modelo GTE carregado com sucesso.")
return model, tokenizer
except Exception as e:
print(f"Erro ao carregar modelo GTE: {e}")
raise RuntimeError(f"Falha ao inicializar modelo de embedding: {str(e)}")
@lru_cache(maxsize=1)
def load_generation_model():
"""Carrega o modelo SeqGPT para geração de texto."""
try:
model_name = 'iic/nlp_seqgpt-560m'
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name)
# Opcional: Mover para GPU se disponível
# if torch.cuda.is_available():
# model.to('cuda')
print("Modelo SeqGPT carregado com sucesso.")
return model, tokenizer
except Exception as e:
print(f"Erro ao carregar modelo SeqGPT: {e}")
raise RuntimeError(f"Falha ao inicializar modelo de geração: {str(e)}")
# --- Endpoint de Verificação de Saúde ---
@app.get("/health")
async def health_check():
"""Verifica o status da API e dos modelos carregados."""
# Verifica se os modelos estão prontos para serem carregados (sem carregá-los ainda)
# Uma verificação mais robusta poderia tentar carregar um pequeno segmento
return {
"status": "ok",
"message": "API está respondendo.",
"models_ready_to_load": {
"embedding_model": True, # Indica que o caminho é válido
"generation_model": True # Indica que o caminho é válido
}
}
# --- Inicialização de estado para carregamento preguiçoso ---
# Estes serão preenchidos na primeira requisição a cada endpoint
app.state.embedding_model_instance = None
app.state.generation_model_instance = None
Funcionalidades Principais
Endpoint de Busca Semântica
Este endpoint utiliza o modelo GTE para calcular a similaridade entre uma consulta e uma lista de documentos.
def calculate_embeddings(texts: List[str], model, tokenizer) -> np.ndarray:
"""Gera embeddings para uma lista de textos."""
encoded_input = tokenizer(texts, padding=True, truncation=True, return_tensors='pt', max_length=512)
with torch.no_grad():
model_output = model(**encoded_input)
# Pega o embedding do token [CLS] como representação da sentença
sentence_embeddings = model_output.last_hidden_state[:, 0]
# Normaliza os embeddings (importante para similaridade cosseno)
norm = torch.norm(sentence_embeddings, p=2, dim=1, keepdim=True)
normalized_embeddings = sentence_embeddings.cpu().numpy() / (norm.cpu().numpy() + 1e-9) # Adiciona epsilon para evitar divisão por zero
return normalized_embeddings
@app.post("/search", response_model=List[SearchResult])
async def perform_semantic_search(search_data: SearchQuery):
"""Realiza busca semântica baseada em similaridade de embeddings."""
try:
# Carregamento preguiçoso do modelo de embedding
if app.state.embedding_model_instance is None:
model, tokenizer = load_embedding_model()
# Guarda a instância para reutilização
app.state.embedding_model_instance = (model, tokenizer)
else:
model, tokenizer = app.state.embedding_model_instance
# Calcula o embedding da consulta
query_embedding = calculate_embeddings([search_data.query_text], model, tokenizer)[0]
# Calcula os embeddings dos documentos
doc_embeddings = calculate_embeddings(search_data.documents, model, tokenizer)
# Calcula a similaridade cosseno
similarities = np.dot(doc_embeddings, query_embedding)
# Junta documentos, scores e índices originais
indexed_similarities = list(enumerate(similarities))
# Ordena pela similaridade (decrescente)
indexed_similarities.sort(key=lambda x: x[1], reverse=True)
# Seleciona os top_k resultados
top_k_results = indexed_similarities[:search_data.k_results]
# Formata a resposta
response_results = []
for index, score in top_k_results:
response_results.append(SearchResult(
document=search_data.documents[index],
similarity_score=float(score),
original_index=index
))
return response_results
except RuntimeError as re:
raise HTTPException(status_code=503, detail=str(re)) # Service Unavailable
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erro interno no servidor durante a busca: {str(e)}")
Endpoint de Geração de Texto
Este endpoint usa o modelo SeqGPT para gerar texto a partir de um prompt.
@app.post("/generate", response_model=GenerationOutput)
async def generate_text_content(gen_data: TextGenerationInput):
"""Gera texto usando o modelo SeqGPT."""
try:
# Carregamento preguiçoso do modelo de geração
if app.state.generation_model_instance is None:
model, tokenizer = load_generation_model()
app.state.generation_model_instance = (model, tokenizer)
else:
model, tokenizer = app.state.generation_model_instance
# Codifica o prompt
input_ids = tokenizer(gen_data.prompt_text, return_tensors="pt").input_ids
# Opcional: Mover para GPU se disponível
# if torch.cuda.is_available():
# input_ids = input_ids.to('cuda')
# Gera a saída do modelo
with torch.no_grad():
output_sequences = model.generate(
input_ids=input_ids,
max_length=gen_data.max_output_length,
temperature=0.7, # Controla a aleatoriedade
do_sample=True, # Habilita amostragem
num_return_sequences=1,
pad_token_id=tokenizer.eos_token_id # Necessário para alguns modelos
)
# Decodifica a saída gerada
generated_text = tokenizer.decode(output_sequences[0], skip_special_tokens=True)
return GenerationOutput(generated_content=generated_text)
except RuntimeError as re:
raise HTTPException(status_code=503, detail=str(re)) # Service Unavailable
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erro interno no servidor durante a geração: {str(e)}")
Execução e Teste da API
Iniciando o Servidor
Use o Uvicorn para rodar a aplicação FastAPI.
# Cria um script executável (run_server.sh)
# #!/bin/bash
# uvicorn main:app --host 0.0.0.0 --port 8000 --reload
# Ou execute diretamente:
uvicorn main:app --host 0.0.0.0 --port 8000 --reload
Após iniciar, acesse:
- Documentação interativa da API (Swagger UI):
http://localhost:8000/docs - Verificação de saúde:
http://localhost:8000/health
Testando os Endpoints
Utilize curl ou um cliente HTTP para testar:
Teste de Busca Semântica
curl -X POST "http://localhost:8000/search" \
-H "Content-Type: application/json" \
-d '{
"query_text": "Como aprender programação em Python?",
"documents": [
"Guia completo de Python para iniciantes.",
"Introdução aos algoritmos e estruturas de dados.",
"Desenvolvimento web com Flask e Django.",
"Machine Learning com Scikit-learn.",
"Dicas para otimizar código Python."
],
"k_results": 3
}'
Teste de Geração de Texto
curl -X POST "http://localhost:8000/generate" \
-H "Content-Type: application/json" \
-d '{
"prompt_text": "Escreva um parágrafo sobre os benefícios da inteligência artificial:",
"max_output_length": 100
}'
Exemplo de Cliente Python
Um script Python simples para interagir com a API:
import requests
import json
API_BASE_URL = "http://localhost:8000"
def test_semantic_search_api():
search_endpoint = f"{API_BASE_URL}/search"
payload = {
"query_text": "Qual a previsão do tempo para amanhã?",
"documents": [
"O dia de hoje será ensolarado.",
"Amanhã há possibilidade de chuva em algumas regiões.",
"Recomendações de hardware para jogos.",
"Tutorial de configuração de rede."
],
"k_results": 2
}
try:
response = requests.post(search_endpoint, json=payload)
response.raise_for_status() # Lança exceção para códigos de erro HTTP
print("--- Resultado da Busca Semântica ---")
print(json.dumps(response.json(), indent=2, ensure_ascii=False))
except requests.exceptions.RequestException as e:
print(f"Erro ao testar busca semântica: {e}")
def test_generation_api():
generate_endpoint = f"{API_BASE_URL}/generate"
payload = {
"prompt_text": "Crie um pequeno poema sobre a primavera:",
"max_output_length": 60
}
try:
response = requests.post(generate_endpoint, json=payload)
response.raise_for_status()
print("\n--- Resultado da Geração de Texto ---")
print(response.json()["generated_content"])
except requests.exceptions.RequestException as e:
print(f"Erro ao testar geração de texto: {e}")
if __name__ == "__main__":
test_semantic_search_api()
test_generation_api()
Otimizações e Considerações Adicionais
Otimização de Carregamento de Modelo
O uso de @lru_cache garante que os modelos sejam carregados apenas uma vez, mesmo que os endpoints sejam chamados múltiplas vezes. O FastAPI gerencia o estado da aplicação (app.state) para manter instâncias dos modelos carregados entre as requisições, evitando recarregamentos desnecessários.
Suporte a Processamento em Lote (Batch Processing)
Para melhorar a eficiência em cenários de alta demanda, consdiere implementar endpoints que aceitem listas de consultas ou prompts para processamento em lote. Isso pode ser feito modificando os modelos de entrada e a lógica de processamento para iterar sobre as entradas e, se possível, utilizar as capacidades de batch dos modelos de Transformers.
# Exemplo de estrutura para batch search
class BatchSearchRequest(BaseModel):
queries: List[SearchQuery]
@app.post("/batch-search", response_model=List[List[SearchResult]])
async def perform_batch_semantic_search(batch_data: BatchSearchRequest):
# ... lógica para processar batch_data.queries ...
# Pode chamar o endpoint /search individualmente ou otimizar o cálculo de embeddings
all_results = []
for search_query in batch_data.queries:
results = await perform_semantic_search(search_query) # Reutiliza a lógica existente
all_results.append(results)
return all_results
Implementação de Cache
Para consultas frequentes com os mesmos inputs, um cache (como Redis) pode reduzir significativamente a carga sobre os modelos. A chave do cache pode ser um hash do input (consulta + documentos para busca, ou prompt para geração).
# Exemplo de lógica de cache (simplificada)
# import redis
# redis_client = redis.Redis(host='localhost', port=6379, db=0)
# CACHE_EXPIRY_SECONDS = 3600 # 1 hora
# @app.post("/search")
# async def perform_semantic_search(search_data: SearchQuery):
# cache_key = f"search:{hash(json.dumps(search_data.dict(), sort_keys=True))}"
# cached_data = redis_client.get(cache_key)
# if cached_data:
# return json.loads(cached_data)
# # ... (lógica de cálculo normal) ...
# # Armazena no cache
# redis_client.setex(cache_key, CACHE_EXPIRY_SECONDS, json.dumps(response_results, default=lambda o: o.__dict__))
# return response_results
Conclusão
Com esta estrutura, você tem uma API funcional para serviços de IA, combinando busca semântica com o modelo GTE e geração de texto com o SeqGPT, tudo servido eficientemente via FastAPI. A abordagem de carregamento preguiçoso e o uso de caches são essenciais para otimizar o desempenho e o uso de recursos em produção.