Arquitectura de Memoria Persistente
para Chatbots en Telegram

José Miguel Amaya — construyendo software con IA

• CTO y socio fundador de Xprende Tech. Full Stack Engineer en Efilm Online.
• Fundador de Piura AI, Tallanix, Python Piura y VICUX.
• Especialista en Python, Django/FastAPI e IA (LangChain, LangGraph, RAG).
• Construí este bot de Telegram con memoria persistente que verán hoy.

comunidad

empresa

“¿Quién ha usado un chatbot que olvida todo lo que le acabas de decir?”

Hoy veremos cómo darle a un bot de Telegram memoria de corto y largo plazo.

Chatbots y la importancia de la memoria

Los chatbots deben dar respuestas coherentes y útiles. Sin memoria, un chatbot “olvida” lo que el usuario le dijo antes y lo obliga a repetir datos en cada mensaje.

De un mensaje en Telegram a una respuesta con memoria

La memoria vive en MongoDB: tanto la de corto como la de largo plazo.

¿Por qué Telegram?

Telegram es una app de mensajería que permite crear bots conversacionales fácilmente.

Usamos un webhook en FastAPI que recibe las actualizaciones de Telegram en tiempo real.

Usuario → Telegram → Webhook (POST /webhook/{token}) │ ▼ Bot (procesa) │ ▼ Respuesta al usuario por Telegram

¿Qué es LangChain?

LangChain es una herramienta para construir aplicaciones que usan LLMs (grandes modelos de lenguaje). Hace fácil conectar el modelo con otras cosas:

¿Qué es un Prompt?

Un prompt es el texto que le das al modelo para guiar su respuesta. Piensa en él como la personalidad + instrucciones del bot.

CHARACTER_PROMPT = """
Estás a punto de jugar a un juego de prueba de Turing como parte de un experimento.
Te emparejarán con un humano en una conversación de chat.
Tu tarea es interpretar al siguiente personaje, convenciendo al humano de que eres un humano real.

## Biografía de Lucía

## Personalidad de Lucía

## Antecedentes del usuario

Esto es lo que sabes sobre el usuario gracias a conversaciones anteriores:

{memory_context}

Proporciona respuestas de texto simple, sin indicadores de formato ni metacomentarios.
"""

Prompt de memoria

Usamos otro prompt para extraer los datos relevantes del usuario y guardarlos en la memoria de largo plazo.

MEMORY_ANALYSIS_PROMPT = """Extraiga y formatee datos personales importantes del usuario a partir de su mensaje.
Céntrese en la información real, no en metacomentarios ni solicitudes.

Los datos importantes incluyen:
- Datos personales (nombre, edad, ubicación)
- Información profesional (trabajo, formación, habilidades)
- Preferencias (gustos, disgustos, favoritos)
- Circunstancias vitales (familia, relaciones)
- Experiencias o logros significativos
- Metas o aspiraciones personales

Examples:
Input: "Oye, ¿podrías recordar que me encanta Star Wars?"
Output: {{
    "is_important": true,
    "formatted_memory": "Le encanta Star Wars"
}}
"""

¿Qué es una cadena (chain)?

Las chains son secuencias de operaciones que procesan y transforman entradas para producir salidas. Son como tuberías que conectan componentes.

¿Qué es LangGraph?

LangGraph es una extensión de LangChain para crear agentes con lógica compleja. Usa grafos para definir flujos de decisión entre pasos.

Flujo definido por nodos y aristas

def create_workflow_graph():
    g = StateGraph(StateBot, context_schema=Context)
    g.add_node("memory_extraction_node", memory_extraction_node)
    g.add_node("memory_injection_node", memory_injection_node)
    g.add_node("generate_response", generate_response)

    g.add_edge(START, "memory_extraction_node")
    g.add_edge("memory_extraction_node", "memory_injection_node")
    g.add_edge("memory_injection_node", "generate_response")
    g.add_edge("generate_response", END)
    return g

El estado de la conversación

from dataclasses import dataclass
from langgraph.graph import MessagesState


@dataclass
class Context:
    user_id: str


class StateBot(MessagesState):
    memory_context: str

Corto plazo vs. largo plazo

Memoria a corto plazo

Es el contexto y los mensajes recientes de una sola conversación.

LangGraph la trata como parte del estado interno del agente, persistido con un checkpointer por cada hilo de conversación.

En desarrollo: InMemorySaver (antes MemorySaver). En producción: MongoDBSaver de langgraph.checkpoint.mongodb.

La ventana de contexto no es infinita

Con memoria de corto plazo activada, las conversaciones largas superan la ventana del LLM. La doc oficial recomienda gestionarla:

Memoria a largo plazo

Son conocimientos que se comparten entre conversaciones distintas: datos permanentes como el perfil del usuario o hechos guardados.

LangGraph provee la API de Store (put, get, search), organizada por namespace. Con index_config, el search es semántico.

from langgraph.store.mongodb import MongoDBStore, create_vector_index_config

store = MongoDBStore.from_conn_string(
    settings.MONGO_DB_URL, collection_name="memories",
    index_config=create_vector_index_config(embed=embeddings, dims=1536),
)
ns = (user_id, "memories")
store.put(ns, memory_id, {"data": "Le encanta Star Wars"})
store.search(ns, query="¿qué le gusta?", limit=5)

¿Cómo modela LangGraph la memoria?

Memoria corta: checkpointer que guarda el estado al ejecutar el grafo.

Memoria larga: Store de larga duración que el agente consulta cuando hace falta.

Implementaciones: InMemorySaver/MongoDBSaver para el hilo y MongoDBStore para los hechos.

MongoDB, la base de datos

Soporta búsqueda de texto (Search) y similitud vectorial con Atlas Vector Search.

Compilar el grafo una sola vez

El checkpointer y el store se crean al arrancar la app. El grafo se compila una vez con ambos y se reutiliza en cada mensaje.

from langgraph.checkpoint.mongodb import MongoDBSaver

with MongoDBSaver.from_conn_string(
    settings.MONGO_DB_URL, db_name=settings.MONGO_DB_NAME,
) as checkpointer:
    graph = create_workflow_graph().compile(
        checkpointer=checkpointer,
        store=store,
    )

async def reply(self, chat_id, user_id, text):
    config = {"configurable": {"thread_id": str(chat_id)}}
    await self.graph.ainvoke(
        {"messages": [HumanMessage(content=text)]},
        config,
        context=Context(user_id=str(user_id)),
    )
    state = await self.graph.aget_state(config)
    return state.values["messages"][-1].content

¿Qué se guarda?

Atlas Vector Search

Es el motor de búsqueda vectorial de MongoDB Atlas: indexa embeddings y hace KNN por similitud con el stage $vectorSearch. No confundir con Atlas Search, que es búsqueda de texto (BM25).

Vector Search como memoria de largo plazo

Configurar MongoDBStore

El store oficial crea el índice vectorial al abrir la conexión (no hay setup() aparte). Definimos colección, index_config y, opcionalmente, TTL.

from langgraph.store.mongodb import MongoDBStore, create_vector_index_config

store = MongoDBStore.from_conn_string(
    settings.MONGO_DB_URL,
    collection_name="memories",
    index_config=create_vector_index_config(embed=embeddings, dims=1536),
)

ttl_config = {"default_ttl": 43200, "refresh_on_read": True}

Nodo de extracción de memoria

Guardamos el hecho en el store con aput. El store calcula el embedding solo; nosotros damos namespace, key y valor.

async def memory_extraction_node(state, runtime: Runtime[Context]):
    response = await get_memory_chain().ainvoke(
        {"message": state["messages"][-1]}
    )
    if response.is_important:

        ns = (runtime.context.user_id, "memories")
        try:
            await runtime.store.aput(
                ns, str(uuid.uuid4()),
                {"data": response.formatted_memory},
            )
        except Exception as e:
            print(f"Error almacenando memoria: {e}")
    return {}

Nodo de inyección de memoria

Buscamos por similitud semántica con asearch; el namespace por usuario garantiza que cada quien recupere solo sus recuerdos.

async def memory_injection_node(state, runtime: Runtime[Context]):
    ns = (runtime.context.user_id, "memories")
    last_message = state["messages"][-1].content

    try:
        items = await runtime.store.asearch(
            ns, query=last_message, limit=5,
        )
        memory_context = "\n".join(i.value["data"] for i in items)
        return {"memory_context": memory_context}
    except Exception as e:
        print(f"Error recuperando memorias: {e}")
        return {"memory_context": ""}

La colección memories

{
  "namespace": ["68412097", "memories"],
  "key": "b1f2c3d4-...",
  "value": { "data": "Le encanta Star Wars" },
  "created_at": "2026-09-30T...",
  "updated_at": "2026-09-30T..."
}

Nodo de generación de respuesta

La memoria de largo plazo se inserta junto con los mensajes del estado (memoria de corto plazo) para entregar una respuesta coherente.

async def generate_response(state: StateBot, config: RunnableConfig):
    memory_context = state.get('memory_context')
    chain = get_character_chain()

    response = await chain.ainvoke(
        {
            "messages": state['messages'],
            "memory_context": memory_context,
        },
        config,
    )
    return {"messages": response}

El bot recuerda tu nombre

Le pregunté mi nombre sin habérselo dado en esa conversación; cuando se lo di, lo recordó perfectamente.

Sobrevive al borrado de la memoria corta

Borré las tablas que guardan la memoria de corto plazo y dejé intactas las de largo plazo. El bot siguió recordando.

Gotchas de la memoria persistente