Tutorial: Servidor IA con Sanic (normalizador)
Guía completa para crear, ejecutar y desplegar un microservicio IA construido con Sanic y Google Generative AI. Código listo para copiar/pegar.
1. Crear entorno y proyecto
mkdir normalizador
cd normalizador
python3 -m venv venv
source venv/bin/activate # En Windows: venv\Scripts\activate
2. Instalar dependencias
pip install sanic google-generativeai python-dateutil
# Opcional (para Swagger / sanic-ext):
pip install "sanic[ext]"
3. Estructura de carpetas
Crear la siguiente estructura:
normalizador/
├── venv/
├── app/
│ ├── __init__.py
│ ├── app.py
│ ├── config.py
│ ├── services.py
│ ├── normalizador_funcion.py
│ ├── google_keys.json
└── requirements.txt (opcional)
4. Código principal: app/app.py
from sanic import Sanic, response
from sanic.request import Request
from sanic.response import json, text
from app.services import analyze_question_with_ai, generate_rag_response, get_gemma_response
import time
app = Sanic("normalizador")
# CORS
@app.middleware("response")
async def cors_headers(request, resp):
resp.headers["Access-Control-Allow-Origin"] = "*"
resp.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
resp.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
return resp
@app.route("/")
async def hello(request: Request):
return text("Servidor IA con Sanic operativo ✅")
@app.post("/generate/")
async def generate_text_with_gemma(request: Request):
try:
data = request.json
if not data or "prompt" not in data:
return json({"error": "Falta el campo 'prompt'"}, status=400)
prompt = data["prompt"]
gemma_output = get_gemma_response(prompt)
return json({"response": gemma_output})
except Exception as e:
return json({"error": f"Error al generar contenido con Gemma: {str(e)}"}, status=500)
@app.post("/generate_rag")
async def generate_response(request: Request):
try:
data = request.json or {}
if "user_query" not in data:
return json({
"error": "Se requiere 'user_query' en el JSON",
"ejemplo": {
"user_query": "¿Cuántos días de vacaciones tengo?",
"context": "Opcional",
"datos": "Opcional"
}
}, status=400)
user_query = data["user_query"]
context = data.get("context", "")
datos = data.get("datos", "")
rag_response = generate_rag_response(user_query, context, datos)
return json({"response": rag_response})
except Exception as e:
return json({"error": f"Error interno: {str(e)}"}, status=500)
@app.post("/analyze_question/")
async def analyze_question(request: Request):
try:
data = request.json or {}
if "question" not in data:
return json({"error": "El campo 'question' es requerido"}, status=400)
user_question = data["question"].strip()
if not user_question:
return json({"error": "La pregunta no puede estar vacía"}, status=400)
analysis_result = analyze_question_with_ai(user_question)
return json(analysis_result, ensure_ascii=False)
except Exception as e:
return json({"error": f"Error interno al procesar la pregunta: {str(e)}"}, status=500)
@app.route("/health")
async def health_check(request: Request):
return json({"status": "ok"})
@app.route("/ping")
async def ping(request: Request):
start = time.time()
end = time.time()
return json({
"message": "pong",
"response_time_ms": round((end - start) * 1000, 2)
})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, workers=2, access_log=False)
5. Configuración de IA: app/config.py
Usá tu código actual; aquí un ejemplo mínimo para cargar claves:
import json
from pathlib import Path
import google.generativeai as genai
import random
def load_api_keys():
try:
with open(Path(__file__).parent / "google_keys.json") as f:
return json.load(f).get("GOOGLE_API_KEYS", [])
except:
return []
class Settings:
def __init__(self):
self.google_api_keys = load_api_keys()
self.gemma_model_name = "gemini-2.5-flash-lite"
self.gemma_temperature = 0
self.gemma_top_p = 1.0
self.gemma_top_k = 40
self.gemma_max_output_tokens = 2048
def get_random_key(self):
if not self.google_api_keys:
raise ValueError("No hay claves API configuradas")
selected_key = random.choice(self.google_api_keys)
genai.configure(api_key=selected_key)
return selected_key
settings = Settings()
Archivo app/google_keys.json ejemplo:
{
"GOOGLE_API_KEYS": [
"TU_API_KEY_AQUI"
]
}
6. Normalizador: app/normalizador_funcion.py
def normalizar_nombre(nombre: str) -> str:
grupos = {
"asistente_nombre": ["quien_soy", "identidad"],
"saludo": ["saludo", "hola", "buen_dia"],
"obtener_facturacion": [
"obtener_facturacion_anual",
"facturacion_anual",
"obtener_facturacion_por_anio",
"obtener_facturacion_por_año",
],
"clientes_mas_compras": [
"obtener_clientes_mas_compras",
"obtener_clientes_mas_compradores",
"obtener_clientes_top_compras",
"obtener_clientes_con_mas_compras",
"obtener_clientes_mas_compraron",
"obtener_clientes_top_compradores",
"listar_compradores_top",
"obtener_compradores_top"
],
}
for funcion_canonica, sinonimos in grupos.items():
if nombre in sinonimos:
return funcion_canonica
return nombre
7. Servicios IA: app/services.py
Pegá tu código existente. Asegurate que el import del normalizador apunte así:
from app.normalizador_funcion import normalizar_nombre
from app.config import settings
import google.generativeai as genai
# ... resto de tus funciones (get_gemma_response, generate_rag_response, analyze_question_with_ai, ...)
8. Ejecutar el servidor
Desarrollo:
python app/app.pyProducción (CLI de Sanic):
sanic app.app:app --host=0.0.0.0 --port=8000 --workers=2
9. Probar la API
curl -X POST http://localhost:8000/generate/ \
-H "Content-Type: application/json" \
-d '{"prompt":"Hola, ¿quién eres?"}'
10. Desplegar en Ubuntu con Nginx
Instalar dependencias
sudo apt update
sudo apt install python3 python3-venv nginx -y
Copiar proyecto y crear entorno
sudo mkdir /opt/normalizador
sudo chown $USER:$USER /opt/normalizador
cd /opt/normalizador
# copiar aquí los archivos del proyecto (app/, venv o crear uno)
python3 -m venv venv
source venv/bin/activate
pip install sanic google-generativeai python-dateutil
Probar localmente
sanic app.app:app --host=0.0.0.0 --port=8000
11. Crear servicio systemd
Crear archivo:
sudo nano /etc/systemd/system/normalizador.service
Contenido:
[Unit]
Description=Normalizador Sanic Service
After=network.target
[Service]
User=tu_usuario
WorkingDirectory=/opt/normalizador
Environment="PATH=/opt/normalizador/venv/bin"
ExecStart=/opt/normalizador/venv/bin/sanic app.app:app --workers=2 --host=127.0.0.1 --port=8000
Restart=always
[Install]
WantedBy=multi-user.target
Activar servicio:
sudo systemctl daemon-reload
sudo systemctl enable normalizador
sudo systemctl start normalizador
sudo systemctl status normalizador
12. Configurar Nginx
Crear archivo de sitio:
sudo nano /etc/nginx/sites-available/normalizador
Contenido:
server {
listen 80;
server_name _;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
Activar y reiniciar:
sudo ln -s /etc/nginx/sites-available/normalizador /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl restart nginx
13. (Opcional) HTTPS con Let's Encrypt
sudo apt install certbot python3-certbot-nginx -y
sudo certbot --nginx -d tu-dominio.com
14. Endpoint público
Tu API debería estar disponible en:
http://TU_IP/ o https://TU_DOMINIO/
15. Extras recomendados (breve)
- Agregar
LOGy rotación de logs (systemd/journald + archivo separado si hace falta). - Monitoreo: exponer métricas o usar Prometheus (puedes agregar un endpoint /metrics si lo necesitás).
- Seguridad: limitar CORS en producción, configurar firewall (ufw), y asegurar las keys en variables de entorno o un vault.
- Pruebas: agregar healthchecks más completas en
/healthy readiness/liveness para orquestadores.
Comentarios
Publicar un comentario