Agente de IA con memoria multisesión

Introducción

En la informática, al igual que en la cognición humana, hay diferentes niveles de memoria:

• Memoria primaria (como RAM) es la memoria temporal activa utilizada para las tareas actuales, el razonamiento y la toma de decisiones en las tareas actuales. Sostiene la información con la que está trabajando actualmente. Es rápido pero volátillo que significa que pierde datos cuando la potencia está apagada.
• Memoria secundaria (como el almacenamiento físico) se refiere al almacenamiento a largo plazo del conocimiento aprendido que no está inmediatamente activo en la memoria de trabajo. No siempre se accede durante la toma de decisiones en tiempo real, pero se puede recuperar cuando sea necesario. Por lo tanto, es más lento pero más persistente.
• Memoria terciaria (como la copia de seguridad de los datos históricos) se refiere a la memoria de archivo, donde la información se almacena para fines de copia de seguridad y la recuperación de desastres. Se caracteriza por alta capacidad y bajo costo, pero con más lento tiempo de acceso. En consecuencia, rara vez se usa.

Los agentes de IA pueden aprovechar todos los tipos de memoria. Primero, pueden usar la memoria primaria para manejar su pregunta actual. Luego, accederían a la memoria secundaria para traer conocimiento de las conversaciones recientes. Y, si es necesario, incluso podrían recuperar información anterior de la memoria terciaria.

En este tutorial, voy a mostrar cómo Construya un agente de IA con memoria en múltiples sesiones. Presentaré un código de Python útil que se puede aplicar fácilmente en otros casos similares (solo copie, pegue, ejecute) y camine por cada línea de código con comentarios para que pueda replicar este ejemplo (enlace al código completo al final del artículo).

Configuración

Comencemos por configurar Ollama (pip install ollama==0.5.1), una biblioteca que permite a los usuarios ejecutar LLM de código abierto localmente, sin necesidad de servicios basados ​​en la nube, lo que brinda más control sobre la privacidad y el rendimiento de los datos. Dado que se ejecuta localmente, cualquier datos de conversación no sale de su máquina.
En primer lugar, necesitas descargar Ollama del sitio web.

Luego, en el shell de inmediato de su computadora portátil, use el comando para descargar el LLM seleccionado. Voy con Alibaba’s Qwencomo es inteligente y ligero.

Después de completar la descarga, puede pasar a Python y comenzar a escribir código.

import ollama
llm = "qwen2.5"

Probemos el LLM:

stream = ollama.generate(model=llm, prompt='''what time is it?''', stream=True)
for chunk in stream:
    print(chunk['response'], end='', flush=True)

Base de datos

Un agente con memoria multisesión es un sistema de inteligencia artificial que puede Recuerde la información de una interacción a la siguienteincluso si esas interacciones ocurren en diferentes momentos o durante sesiones separadas. Por ejemplo, una IA asistente personal que recuerda su horario y preferencias diarias, o un bot de atención al cliente que conoce el historial de su problema sin necesitar que vuelva a explicarle cada vez.

Básicamente, el agente necesita acceder al historial de chat. En base a la antigüedad de las conversaciones pasadas, esto podría clasificarse como memoria secundaria o terciaria.

Vamos a trabajar. Podemos almacenar datos de conversación en un base de datos vectorialque es la mejor solución para almacenar, indexar y buscar datos no estructurados de manera eficiente. Actualmente, el Vector DB más utilizado es Microsoft’s Aisearchmientras que el mejor de código abierto es Cromadbque es útil, fácil y gratuito.

Después de un rápido pip install chromadb==0.5.23 Puedes interactuar con el DB usando Python en tres formas diferentes:

• chromadb.Client() Para crear un DB que permanezca temporalmente en la memoria sin ocupar espacio físico en el disco.
• chromadb.PersistentClient(path) Para guardar y cargar el DB desde su máquina local.
• chromadb.HttpClient(host='localhost', port=8000) tener un modo cliente-servidor en su navegador.

Al almacenar documentos en Cromadblos datos se guardan como vectores para que uno pueda buscar con un vector de consulta para recuperar los registros coincidentes más cercanos. Tenga en cuenta que, si no se especifica lo contrario, la función de incrustación predeterminada es una transformador de oraciones modelo (All-Minilm-L6-V2).

import chromadb

## connect to db
db = chromadb.PersistentClient()

## check existing collections
db.list_collections()

## select a collection
collection_name = "chat_history"
collection = db.get_or_create_collection(name=collection_name, 
    embedding_function=chromadb.utils.embedding_functions.DefaultEmbeddingFunction())

Para almacenar sus datos, primero necesita extraer el chat y guárdelo como un documento de texto. En Ollamahay 3 roles En la interacción con un LLM:

• sistema – Se utiliza para pasar instrucciones centrales al modelo sobre cómo debe continuar la conversación (es decir, el aviso principal)
• usuario – Se utiliza para las preguntas del usuario, y también para el refuerzo de la memoria (es decir, “recuerde que la respuesta debe tener un formato específico”)
• asistente – Es la respuesta del modelo (es decir, la respuesta final)

Asegúrese de que cada documento tenga una identificación única, que puede generar manualmente o permitir Croma para generar automáticamente. Una cosa importante a mencionar es que puede agregar información adicional como metadatos (es decir, título, etiquetas, enlaces). Es opcional pero muy útil, como enriquecimiento de metadatos puede mejorar significativamente la recuperación de documentos. Por ejemplo, aquí, voy a usar el LLM para resumir cada documento en algunas palabras clave.

from datetime import datetime

def save_chat(lst_msg, collection):
    print("--- Saving Chat ---")
    ## extract chat
    chat = ""
    for m in lst_msg:
        chat += f'{m["role"]}: <<{m["content"]}>>' +'\n\n'
    ## get idx
    idx = str(collection.count() +1)
    ## generate info
    p = "Describe the following conversation using only 3 keywords separated by a comma (for example: 'finance, volatility, stocks')."
    tags = ollama.generate(model=llm, prompt=p+"\n"+chat)["response"]
    dic_info = {"tags":tags,
                "date": datetime.today().strftime("%Y-%m-%d"),
                "time": datetime.today().strftime("%H:%M")}
    ## write db
    collection.add(documents=[chat], ids=[idx], metadatas=[dic_info])
    print(f"--- Chat num {idx} saved ---","\n")
    print(dic_info,"\n")
    print(chat)
    print("------------------------")

Necesitamos comenzar y guardar una charla para verlo en acción.

Ejecutar agente básico

Para comenzar, ejecutaré un chat LLM muy básico (no se necesitan herramientas) para guardar la primera conversación en la base de datos. Durante la interacción, voy a mencionar información importante, no incluida en la base de conocimiento de LLM, que quiero que el agente recuerde en la próxima sesión.

indic = "Usted es un asistente inteligente, proporcione la mejor respuesta posible a la solicitud del usuario". Mensajes = [{"role":"system", "content":prompt}]

while True: ## User q = input('🙂 >') if q == "quit": ### save chat before quitting save_chat(lst_msg=messages, collection=collection) break messages.append( {"role":"user", "content":q} ) ## Model agent_res = ollama.chat(model=llm, messages=messages, tools=[]) res = agente_res["message"]["content"]
   
    
   
    ## Respuesta imprime ("👽>", f "\ x1b[1;30m{res}\x1b[0m")
    messages.append( {"role":"assistant", "content":res} )

At the end, the conversation was saved with enriched metadata.

Tools

I want the Agent to be able to retrieve information from previous conversations. Therefore, I need to provide it with a Tool to do so. To put it in another way, the Agent must do a Retrieval-Augmented Generation (RAG) from the history. It’s a technique that combines retrieval and generative models by adding to LLMs knowledge facts fetched from external sources (in this case, ChromaDB).

def retrieve_chat(query:str) -> str:
    res_db = collection.query(query_texts=[query])["documents"][0][0:10]
    
    History = '' .Join (res_db) .replace ("\ n", ") return History Tool_retrieve_chat = {'type': 'function', 'function': {'name': 'Remieve_chat', 'Descripción': 'cuando su conocimiento no es suficiente para responder al usuario, puede usar esta herramienta para recuperar la historia de chat de chat. ['query']'Propiedades': {'Query': {'type': 'str', 'descripción': 'Ingrese la pregunta del usuario o el tema del chat actual'},}}}}

Después de obtener datos, la IA debe procesar toda la información y dar la respuesta final al usuario. A veces, puede ser más efectivo tratar el “Respuesta final” como herramienta. Por ejemplo, si el agente realiza múltiples acciones para generar resultados intermedios, la respuesta final puede considerarse como la herramienta que integra toda esta información en una respuesta cohesiva. Al diseñarlo de esta manera, tiene más personalización y control sobre los resultados.

def final_answer(text:str) -> str:
    return text

tool_final_answer = {'type':'function', 'function':{
  'name': 'final_answer',
  'description': 'Returns a natural language response to the user',
  'parameters': {'type': 'object', 
                 'required': ['text'],
                 'properties': {'text': {'type':'str', 'description':'natural language response'}}
}}}

Finalmente estamos listos para probar el agente y su memoria.

dic_tools = {'retrieve_chat':retrieve_chat, 
             'final_answer':final_answer}

Ejecutar agente con memoria

Agregaré un par de Utils Funciones para el uso de la herramienta y para ejecutar el agente.

def use_tool(agent_res:dict, dic_tools:dict) -> dict:
    ## use tool
    if agent_res["message"].tool_calls is not None:
        for tool in agent_res["message"].tool_calls:
            t_name, t_inputs = tool["function"]["name"], tool["function"]["arguments"]
            if f := dic_tools.get(t_name):
                ### calling tool
                print('🔧 >', f"\x1b[1;31m{t_name} -> Inputs: {t_inputs}\x1b[0m")
                ### tool output
                t_output = f(**tool["function"]["arguments"])
                print(t_output)
                ### final res
                res = t_output
            else:
                print('🤬 >', f"\x1b[1;31m{t_name} -> NotFound\x1b[0m")      
    ## don't use tool
    else:
        res = agent_res["message"].content
        t_name, t_inputs = '', ''
    return {'res':res, 'tool_used':t_name, 'inputs_used':t_inputs}

Cuando el agente intenta resolver una tarea, quiero realizar un seguimiento de las herramientas que se han utilizado y los resultados que obtiene. El modelo debe probar cada herramienta solo una vez, y la iteración se detendrá solo cuando el agente esté listo para dar la respuesta final.

def run_agent(llm, messages, available_tools):
    ## use tools until final answer
    tool_used, local_memory = '', ''
    while tool_used != 'final_answer':
        ### use tool
        try:
            agent_res = ollama.chat(model=llm, messages=messages, tools=[v for v in available_tools.values()])
            dic_res = use_tool(agent_res, dic_tools)
            res, tool_used, inputs_used = dic_res["res"], dic_res["tool_used"], dic_res["inputs_used"]
        ### error
        except Exception as e:
            print("⚠️ >", e)
            res = f"I tried to use {tool_used} but didn't work. I will try something else."
            print("👽 >", f"\x1b[1;30m{res}\x1b[0m")
            messages.append( {"role":"assistant", "content":res} )       
        ### update memory
        if tool_used not in ['','final_answer']:
            local_memory += f"\n{res}"
            messages.append( {"role":"user", "content":local_memory} )
            available_tools.pop(tool_used)
            if len(available_tools) == 1:
                messages.append( {"role":"user", "content":"now activate the tool final_answer."} ) 
        ### tools not used
        if tool_used == '':
            break
    return res

Comencemos una nueva interacción, y esta vez quiero que el agente active todas las herramientas, para recuperar y procesar información anterior.

prompt = '''
You are an intelligent assistant, provide the best possible answer to user's request. 
You must return natural language response.
When interacting with a user, first you must use the tool 'retrieve_chat' to remember previous chats history.  
'''
messages = [{"role":"system", "content":prompt}]

while True:
    ## User
    q = input('🙂 >')
    if q == "quit":
        ### save chat before quitting
        save_chat(lst_msg=messages, collection=collection)
        break
    messages.append( {"role":"user", "content":q} )
   
    ## Model
    available_tools = {"retrieve_chat":tool_retrieve_chat, "final_answer":tool_final_answer}
    res = run_agent(llm, messages, available_tools)
   
    ## Response
    print("👽 >", f"\x1b[1;30m{res}\x1b[0m")
    messages.append( {"role":"assistant", "content":res} )

Le di al agente una tarea que no estaba directamente correlacionada con el tema de la última sesión. Como se esperaba, el agente activó la herramienta y buscó chats anteriores. Ahora, usará la “respuesta final” para procesar la información y responderme.

Conclusión

Este artículo ha sido un tutorial para demostrar Cómo construir agentes de IA con memoria de múltiples sesiones desde cero utilizando solo Ollama. Con estos bloques de construcción en su lugar, ya está equipado para comenzar a desarrollar sus propios agentes para diferentes casos de uso.

Código completo para este artículo: Github

¡Espero que lo hayas disfrutado! No dude en ponerse en contacto conmigo para obtener preguntas y comentarios o simplemente para compartir sus interesantes proyectos.

👉 Vamos a conectarnos 👈

_{^{(Todas las imágenes, a menos que se indique lo contrario, son por el autor)}}