WebSockets y FastAPI: Construyendo un Chatbot de IA en Tiempo Real
En el mundo del desarrollo de aplicaciones, la interactividad y la respuesta en tiempo real son cada vez más cruciales. Cuando pensamos en chatbots de IA, sistemas de notificaciones en vivo o paneles de control dinámicos, la comunicación tradicional de solicitud-respuesta HTTP a menudo se queda corta. Aquí es donde entran en juego los WebSockets, ofreciendo una solución robusta para la comunicación bidireccional y persistente. [1, 4, 14, 15, 26]
Este artículo te guiará a través de la construcción de un chatbot de IA simple utilizando WebSockets con FastAPI, el framework web moderno y de alto rendimiento para construir APIs con Python. Aprenderás los conceptos clave, implementarás un servidor WebSocket, crearás un cliente web básico y entenderás cómo manejar las conexiones y los mensajes en tiempo real. [1, 2, 3, 4, 5, 15, 16]
Contexto del Problema: La Necesidad de la Comunicación en Tiempo Real
Imagina un chatbot de IA. Si cada mensaje del usuario y cada respuesta de la IA requirieran una nueva solicitud HTTP, la experiencia sería lenta y fragmentada. HTTP es un protocolo sin estado, diseñado para transacciones de una sola vez. Para una conversación fluida y continua, necesitamos un canal de comunicación persistente donde tanto el cliente como el servidor puedan enviar datos en cualquier momento, sin necesidad de iniciar una nueva conexión para cada intercambio. [14, 26]
Aquí es donde los WebSockets brillan. Permiten una conexión dúplex completa sobre un único socket TCP, lo que significa que los datos pueden fluir en ambas direcciones simultáneamente, de forma eficiente y con baja latencia. Esto es ideal para aplicaciones que demandan actualizaciones instantáneas, como aplicaciones de chat, juegos multijugador o herramientas de colaboración en tiempo real. [4, 8, 14, 15, 26]
Conceptos Clave
¿Qué son los WebSockets?
Los WebSockets son un protocolo de comunicación que proporciona canales de comunicación bidireccionales y persistentes sobre una única conexión TCP. A diferencia de HTTP, que es unidireccional y se cierra después de cada respuesta, un WebSocket, una vez establecido, permanece abierto, permitiendo que el cliente y el servidor se envíen mensajes mutuamente en cualquier momento. [1, 4, 14, 26]
Ventajas para Aplicaciones de IA
- Baja Latencia: La conexión persistente elimina la sobrecarga de establecer nuevas conexiones para cada mensaje, lo que resulta en respuestas casi instantáneas, crucial para la interactividad de la IA.
- Comunicación Bidireccional: Tanto el cliente (navegador, aplicación móvil) como el servidor pueden iniciar el envío de datos, lo que es perfecto para chatbots donde la IA puede enviar mensajes proactivamente o el usuario puede interrumpir.
- Eficiencia: Menos sobrecarga de encabezados en comparación con HTTP, lo que reduce el uso de ancho de banda una vez que la conexión está establecida.
FastAPI y WebSockets
FastAPI, construido sobre Starlette, ofrece soporte nativo y elegante para WebSockets. Utiliza el decorador @app.websocket() para definir rutas WebSocket, y la clase WebSocket para manejar la conexión. [1, 2, 5]
Manejo de Conexiones y Mensajes
Para un chatbot, es fundamental poder gestionar múltiples conexiones de clientes. Esto implica:
- Aceptar Conexiones: Cuando un cliente intenta conectarse, el servidor debe aceptar la conexión. [1, 2]
- Enviar y Recibir Mensajes: La capacidad de enviar texto (
send_text) y recibir texto (receive_text) es el núcleo de la comunicación. [1, 5] - Gestionar Múltiples Clientes: Para un chat, el servidor necesita mantener un registro de todas las conexiones activas para poder enviar mensajes a clientes específicos o a todos los conectados (broadcast). [1, 2, 11, 16]
- Manejo de Desconexiones: Es vital detectar cuándo un cliente se desconecta para limpiar los recursos y evitar errores. [2, 9, 13]
Implementación Paso a Paso
Paso 1: Configuración del Entorno
Primero, asegúrate de tener Python 3.7+ instalado. Luego, instala FastAPI y Uvicorn (un servidor ASGI) junto con la librería websockets:
pip install fastapi uvicorn websocketsPaso 2: Creación del Servidor FastAPI con WebSocket
Crearemos un archivo main.py. Este archivo contendrá nuestra aplicación FastAPI y el endpoint WebSocket. Implementaremos una clase ConnectionManager para gestionar las conexiones activas, lo cual es una práctica común para aplicaciones de chat. [1, 11, 16]
# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import List
import os
import time
app = FastAPI()
# Clase para gestionar las conexiones WebSocket activas
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
# Simulación de una "IA" que responde a mensajes
async def mock_ai_response(message: str) -> str:
# Simula un procesamiento de IA con un pequeño retraso
await asyncio.sleep(0.5)
message_lower = message.lower()
if "hola" in message_lower:
return "¡Hola! ¿En qué puedo ayudarte hoy?"
elif "cómo estás" in message_lower:
return "Estoy bien, gracias por preguntar. Soy un programa, así que no tengo sentimientos, ¡pero estoy listo para conversar!"
elif "python" in message_lower:
return "Python es un lenguaje increíble para la IA. ¿Tienes alguna pregunta específica sobre él?"
elif "fastapi" in message_lower:
return "FastAPI es excelente para construir APIs rápidas y modernas, ¡especialmente con WebSockets!"
elif "gracias" in message_lower:
return "De nada. ¡Estoy aquí para servirte!"
else:
return "No estoy seguro de cómo responder a eso. ¿Puedes reformular tu pregunta?"
# Endpoint HTTP para servir el cliente HTML
@app.get("/", response_class=HTMLResponse)
async def get():
# El cliente HTML se servirá desde aquí. En un proyecto real, usarías un servidor de archivos estáticos.
html_content = """
<!DOCTYPE html>
<html lang="es">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Chatbot de IA en Tiempo Real</title>
<style>
body { font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; margin: 0; padding: 20px; background-color: #f4f7f6; color: #333; display: flex; flex-direction: column; align-items: center; min-height: 100vh; }
.chat-container { background-color: #fff; border-radius: 8px; box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1); width: 100%; max-width: 600px; display: flex; flex-direction: column; overflow: hidden; }
.chat-header { background-color: #007bff; color: white; padding: 15px 20px; text-align: center; font-size: 1.2em; border-bottom: 1px solid #0056b3; }
.chat-box { flex-grow: 1; padding: 20px; overflow-y: auto; max-height: 70vh; background-color: #e9ecef; border-bottom: 1px solid #dee2e6; }
.message { margin-bottom: 15px; display: flex; }
.message.user { justify-content: flex-end; }
.message.ai { justify-content: flex-start; }
.message-bubble { padding: 10px 15px; border-radius: 20px; max-width: 70%; line-height: 1.4; word-wrap: break-word; }
.message.user .message-bubble { background-color: #007bff; color: white; border-bottom-right-radius: 5px; }
.message.ai .message-bubble { background-color: #f8f9fa; color: #333; border: 1px solid #ced4da; border-bottom-left-radius: 5px; }
.chat-input { display: flex; padding: 15px 20px; border-top: 1px solid #dee2e6; background-color: #fff; }
.chat-input input { flex-grow: 1; padding: 10px 15px; border: 1px solid #ced4da; border-radius: 20px; margin-right: 10px; font-size: 1em; }
.chat-input button { background-color: #28a745; color: white; border: none; border-radius: 20px; padding: 10px 20px; cursor: pointer; font-size: 1em; transition: background-color 0.2s ease; }
.chat-input button:hover { background-color: #218838; }
#status { text-align: center; margin-top: 10px; font-size: 0.9em; color: #6c757d; }
</style>
</head>
<body>
<div class="chat-container">
<div class="chat-header">Chatbot de IA en Tiempo Real</div>
<div class="chat-box" id="chatBox"></div>
<div class="chat-input">
<input type="text" id="messageInput" placeholder="Escribe tu mensaje..." autocomplete="off">
<button id="sendButton">Enviar</button>
</div>
</div>
<div id="status">Conectando...</div>
<script>
const chatBox = document.getElementById('chatBox');
const messageInput = document.getElementById('messageInput');
const sendButton = document.getElementById('sendButton');
const statusDiv = document.getElementById('status');
let ws;
function connectWebSocket() {
// Usar 'ws://' para HTTP y 'wss://' para HTTPS
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
const host = window.location.host;
ws = new WebSocket(`${protocol}//${host}/ws`);
ws.onopen = (event) => {
statusDiv.textContent = 'Conectado al chatbot.';
console.log('WebSocket conectado:', event);
};
ws.onmessage = (event) => {
const messageData = JSON.parse(event.data);
addMessage(messageData.sender, messageData.message);
};
ws.onclose = (event) => {
statusDiv.textContent = 'Desconectado. Intentando reconectar...';
console.log('WebSocket desconectado:', event);
// Intentar reconectar después de un breve retraso
setTimeout(connectWebSocket, 3000);
};
ws.onerror = (error) => {
statusDiv.textContent = 'Error en la conexión. Intentando reconectar...';
console.error('WebSocket error:', error);
ws.close(); // Forzar cierre para activar onclose y reconexión
};
}
function addMessage(sender, message) {
const messageDiv = document.createElement('div');
messageDiv.classList.add('message', sender);
const bubbleDiv = document.createElement('div');
bubbleDiv.classList.add('message-bubble');
bubbleDiv.textContent = message;
messageDiv.appendChild(bubbleDiv);
chatBox.appendChild(messageDiv);
chatBox.scrollTop = chatBox.scrollHeight; // Auto-scroll al final
}
sendButton.addEventListener('click', () => {
sendMessage();
});
messageInput.addEventListener('keypress', (event) => {
if (event.key === 'Enter') {
sendMessage();
}
});
function sendMessage() {
const message = messageInput.value.trim();
if (message && ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ message: message }));
addMessage('user', message);
messageInput.value = '';
} else if (ws.readyState !== WebSocket.OPEN) {
statusDiv.textContent = 'No conectado. Espera la reconexión.';
}
}
// Iniciar la conexión WebSocket al cargar la página
connectWebSocket();
</script>
</body>
</html>
"""
return HTMLResponse(content=html_content)
# Endpoint WebSocket para el chatbot
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_json() # Espera un JSON del cliente
user_message = data.get("message")
if user_message:
print(f"Mensaje recibido del cliente: {user_message}")
# Obtener respuesta de la IA simulada
ai_response = await mock_ai_response(user_message)
# Enviar la respuesta de la IA de vuelta al cliente
await manager.send_personal_message(json.dumps({"sender": "ai", "message": ai_response}), websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)
print(f"Cliente desconectado: {websocket.client}")
except Exception as e:
print(f"Error en WebSocket: {e}")
# Opcional: enviar un mensaje de error al cliente antes de cerrar
await manager.send_personal_message(json.dumps({"sender": "ai", "message": "Lo siento, ha ocurrido un error interno."}), websocket)
manager.disconnect(websocket)
# Para ejecutar la aplicación:
# uvicorn main:app --reload
Paso 3: Ejecución de la Aplicación
Guarda el código anterior como main.py. Abre tu terminal en el mismo directorio y ejecuta:
uvicorn main:app --reloadLuego, abre tu navegador y ve a http://127.0.0.1:8000. Verás la interfaz del chatbot. Abre varias pestañas o ventanas para simular múltiples usuarios y observa cómo los mensajes de la IA se envían de vuelta a cada cliente individualmente. [5, 15]
Mini Proyecto / Aplicación Sencilla: Un Chatbot de IA Básico
El código proporcionado ya es un mini-proyecto funcional. El servidor FastAPI maneja las conexiones WebSocket y utiliza una función mock_ai_response para simular la lógica de un chatbot. El cliente HTML/JavaScript se conecta al servidor, envía mensajes y muestra las respuestas de la IA en tiempo real. [1, 2, 5, 7, 15, 16, 21, 24, 26, 33]
La función mock_ai_response es un ejemplo simple de cómo podrías integrar la lógica de tu IA. En un escenario real, esta función interactuaría con un modelo de lenguaje grande (LLM) o un sistema de procesamiento de lenguaje natural (NLP).
Errores Comunes y Depuración
-
WebSocket connection to 'ws://...' failed: Error during WebSocket handshake: Unexpected response code: 404: Esto suele indicar que la URL del WebSocket es incorrecta o que el endpoint no está definido correctamente en FastAPI. Verifica la ruta/ws. [23] -
Cierre inesperado de conexiones (
WebSocketDisconnect): Es normal que las conexiones se cierren cuando el cliente cierra la pestaña o pierde la red. La excepciónWebSocketDisconnectde FastAPI (que viene de Starlette) maneja esto elegantemente. Asegúrate de que tu lógica de desconexión (comomanager.disconnect(websocket)) se ejecute en el bloqueexcept WebSocketDisconnect. [2, 9, 13] -
Problemas de CORS (Cross-Origin Resource Sharing): Si tu cliente HTML se sirve desde un origen diferente al de tu API FastAPI (por ejemplo, dominios o puertos diferentes), el navegador podría bloquear la conexión WebSocket por políticas de seguridad. Puedes configurar
CORSMiddlewareen FastAPI para permitir conexiones desde orígenes específicos. [28, 31]from fastapi.middleware.cors import CORSMiddleware app = FastAPI() origins = [ "http://localhost:8000", # Permite tu origen de desarrollo "http://127.0.0.1:8000", # Agrega aquí otros orígenes si tu frontend está en otro dominio/puerto ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) -
Bloqueo de la UI por operaciones de IA lentas: Si tu función de IA es síncrona o muy lenta, podría bloquear el bucle de eventos de FastAPI, afectando la capacidad de respuesta de otras conexiones. Asegúrate de que las operaciones de E/S o las llamadas a modelos de IA externos sean
await-ables o se ejecuten en tareas en segundo plano. [10, 25, 29, 30, 32] -
Errores de JSON: Asegúrate de que tanto el cliente como el servidor envíen y reciban datos en el formato JSON esperado. Utiliza
JSON.stringify()en JavaScript ywebsocket.receive_json()/json.dumps()en Python. [33]
Aprendizaje Futuro
Este es solo el comienzo. Aquí hay algunas ideas para llevar tu chatbot de IA en tiempo real al siguiente nivel:
-
Integración con LLMs Reales: Reemplaza
mock_ai_responsecon llamadas a APIs de modelos de lenguaje grandes como OpenAI GPT, Anthropic Claude o Google Vertex AI. Necesitarás manejar las claves API de forma segura (usando variables de entorno). [10] - Persistencia de la Conversación (Memoria): Para que el chatbot recuerde el contexto de la conversación, implementa un sistema de memoria. Esto podría implicar almacenar el historial de chat en una base de datos (SQL, NoSQL o vectorial) y pasarlo al LLM en cada turno.
- Autenticación y Autorización: Protege tus endpoints WebSocket implementando autenticación (por ejemplo, JWT) y autorización para asegurar que solo los usuarios permitidos puedan interactuar con el chatbot. [11]
- Escalabilidad: Para manejar un gran número de usuarios, considera usar un broker de mensajes como Redis Pub/Sub para distribuir los mensajes entre múltiples instancias de tu aplicación FastAPI. [11]
- Streaming de Respuestas: Los LLMs pueden generar respuestas palabra por palabra. Implementa el streaming de respuestas a través de WebSockets para una experiencia de usuario más dinámica, donde el texto aparece a medida que se genera.
-
Manejo de Errores Avanzado: Utiliza
WebSocketExceptionpara enviar códigos de cierre y razones específicas al cliente en caso de errores controlados. [9, 13, 17] - Despliegue en Producción: Aprende a desplegar tu aplicación FastAPI con WebSockets en servicios en la nube como Railway, Render, o plataformas de contenedores como Docker y Kubernetes.
Los WebSockets abren un mundo de posibilidades para aplicaciones interactivas y en tiempo real con IA. Con FastAPI, tienes una herramienta poderosa y fácil de usar para construir estas soluciones de manera eficiente.