Tutorial: Chatbot con memoria usando OpenAI + Redis

Chatbot con Memoria Persistente — OpenAI + Redis

Los chatbots sin memoria repiten preguntas y pierden contexto entre sesiones. Con Redis, cada usuario tiene su historial guardado y el bot recuerda conversaciones anteriores.

Instalación

pip install openai redis

Chatbot completo con memoria

import json
import redis
from openai import OpenAI
from datetime import datetime

client = OpenAI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

SYSTEM_PROMPT = """Eres un asistente amable de la comunidad de Erik Taveras.
Ayudas con Python, Django, FastAPI, automatización con N8N, e IA.
Recuerdas las conversaciones anteriores del usuario.
Si el usuario ya te dijo su nombre o proyecto, úsalo en las respuestas."""

MAX_HISTORY = 20  # Últimos 20 mensajes por usuario

def get_history(user_id: str) -> list[dict]:
    key = f"chat:{user_id}"
    raw = r.lrange(key, 0, -1)
    return [json.loads(m) for m in raw]

def save_message(user_id: str, role: str, content: str):
    key = f"chat:{user_id}"
    message = {
        "role": role,
        "content": content,
        "timestamp": datetime.now().isoformat(),
    }
    r.rpush(key, json.dumps(message))
    r.ltrim(key, -MAX_HISTORY, -1)  # Mantener solo los últimos N
    r.expire(key, 86400 * 30)       # Expirar después de 30 días

def chat(user_id: str, message: str) -> str:
    save_message(user_id, "user", message)

    history = get_history(user_id)
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend([
        {"role": m["role"], "content": m["content"]}
        for m in history
    ])

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        max_tokens=500,
        temperature=0.7,
    )

    reply = response.choices[0].message.content
    save_message(user_id, "assistant", reply)
    return reply

Integración con FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    user_id: str
    message: str

@app.post("/chat")
async def chat_endpoint(req: ChatRequest):
    reply = chat(req.user_id, req.message)
    return {"reply": reply}

@app.get("/chat/{user_id}/history")
async def get_chat_history(user_id: str):
    history = get_history(user_id)
    return {"messages": history, "count": len(history)}

@app.delete("/chat/{user_id}")
async def clear_history(user_id: str):
    r.delete(f"chat:{user_id}")
    return {"status": "cleared"}

Integración con Django

# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_POST
import json

@require_POST
def chat_view(request):
    data = json.loads(request.body)
    user_id = str(request.user.id)
    reply = chat(user_id, data['message'])
    return JsonResponse({'reply': reply})

¿Por qué Redis y no una base de datos normal?

• Velocidad: Redis responde en <1ms vs 5-50ms de PostgreSQL
• TTL nativo: Las conversaciones expiran automáticamente (30 días)
• Listas: RPUSH + LTRIM mantienen el historial limitado sin queries complejos
• Concurrencia: Múltiples usuarios chatean simultáneamente sin problemas

Tips

• Para producción, usa redis.ConnectionPool en vez de crear conexiones por request
• Si la memoria crece mucho, resúmela con otro call a GPT: "Resume esta conversación en 3 puntos"
• Costo estimado: GPT-4o-mini ~$0.0003 por mensaje con historial de 20 mensajes