Memoria & RAG
Hay varios casos de uso donde resulta valioso mantener un almacén de hechos útiles que puedan añadirse inteligentemente al contexto del agente justo antes de un paso específico. El caso típico aquí es un patrón RAG, donde una consulta se utiliza para recuperar información relevante de una base de datos que luego se añade al contexto del agente.
Saptiva-Agents proporciona un protocolo de memoria llamado Memory que se puede extender para proporcionar esta funcionalidad. Los métodos clave son query, update_context, add, clear y close.
add: agrega nuevas entradas al almacén de memoriaquery: recupera información relevante del almacén de memoriaupdate_context: modifica elmodel_contextinterno del agente agregando la información recuperada (utilizado en la claseAssistantAgent)clear: borra todas las entradas del almacén de memoriaclose: limpia cualquier recurso utilizado por el almacén de memoria
Ejemplo de ListMemory
ListMemoryListMemory es una implementación de ejemplo del protocolo Memory. Es una implementación simple basada en una lista que mantiene los recuerdos en orden cronológico, agregando los más recientes al contexto del modelo. Esta implementación es directa y predecible, lo que facilita su comprensión y depuración.
from saptiva_agents import SAPTIVA_OPS
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.ui import Console
from saptiva_agents.memory import ListMemory, MemoryContent, MemoryMimeType
from saptiva_agents.base import SaptivaAIChatCompletionClient# Inicializar la memoria del usuario
user_memory = ListMemory()
# Agregar preferencias del usuario a la memoria
await user_memory.add(MemoryContent(content="The weather should be in metric units", mime_type=MemoryMimeType.TEXT))
await user_memory.add(MemoryContent(content="Meal recipe must be vegan", mime_type=MemoryMimeType.TEXT))
async def get_weather(city: str, units: str = "imperial") -> str:
if units == "imperial":
return f"The weather in {city} is 73 °F and Sunny."
elif units == "metric":
return f"The weather in {city} is 23 °C and Sunny."
else:
return f"Sorry, I don't know the weather in {city}."
assistant_agent = AssistantAgent(
name="assistant_agent",
model_client=SaptivaAIChatCompletionClient(
model=SAPTIVA_OPS,
api_key="TU_SAPTIVA_API_KEY",
),
tools=[get_weather],
memory=[user_memory],
)# Ejecutar el agente con una tarea
stream = assistant_agent.run_stream(task="What is the weather in New York?")
await Console(stream)---------- user ----------
What is the weather in New York?
---------- assistant_agent ----------
[MemoryContent(content='The weather should be in metric units', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None), MemoryContent(content='Meal recipe must be vegan', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None)]
---------- assistant_agent ----------
[FunctionCall(id='call_GpimUGED5zUbfxORaGo2JD6F', arguments='{"city":"New York","units":"metric"}', name='get_weather')]
---------- assistant_agent ----------
[FunctionExecutionResult(content='The weather in New York is 23 °C and Sunny.', call_id='call_GpimUGED5zUbfxORaGo2JD6F', is_error=False)]
---------- assistant_agent ----------
The weather in New York is 23 °C and Sunny.TaskResult(messages=[TextMessage(source='user', models_usage=None, metadata={}, content='What is the weather in New York?', type='TextMessage'), MemoryQueryEvent(source='assistant_agent', models_usage=None, metadata={}, content=[MemoryContent(content='The weather should be in metric units', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None), MemoryContent(content='Meal recipe must be vegan', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None)], type='MemoryQueryEvent'), ToolCallRequestEvent(source='assistant_agent', models_usage=RequestUsage(prompt_tokens=123, completion_tokens=20), metadata={}, content=[FunctionCall(id='call_GpimUGED5zUbfxORaGo2JD6F', arguments='{"city":"New York","units":"metric"}', name='get_weather')], type='ToolCallRequestEvent'), ToolCallExecutionEvent(source='assistant_agent', models_usage=None, metadata={}, content=[FunctionExecutionResult(content='The weather in New York is 23 °C and Sunny.', call_id='call_GpimUGED5zUbfxORaGo2JD6F', is_error=False)], type='ToolCallExecutionEvent'), ToolCallSummaryMessage(source='assistant_agent', models_usage=None, metadata={}, content='The weather in New York is 23 °C and Sunny.', type='ToolCallSummaryMessage')], stop_reason=None)Podemos inspeccionar que el model_context del assistant_agent se actualiza realmente con las entradas de memoria recuperadas. El método transform se utiliza para formatear dichas entradas en una cadena de texto que pueda ser utilizada por el agente. En este caso, simplemente concatenamos el contenido de cada entrada de memoria en una única cadena.
await assistant_agent._model_context.get_messages()[UserMessage(content='What is the weather in New York?', source='user', type='UserMessage'),
SystemMessage(content='\nRelevant memory content (in chronological order):\n1. The weather should be in metric units\n2. Meal recipe must be vegan\n', type='SystemMessage'),
AssistantMessage(content=[FunctionCall(id='call_GpimUGED5zUbfxORaGo2JD6F', arguments='{"city":"New York","units":"metric"}', name='get_weather')], thought=None, source='assistant_agent', type='AssistantMessage'),
FunctionExecutionResultMessage(content=[FunctionExecutionResult(content='The weather in New York is 23 °C and Sunny.', call_id='call_GpimUGED5zUbfxORaGo2JD6F', is_error=False)], type='FunctionExecutionResultMessage')]Vemos arriba que el clima se devuelve en grados centígrados, tal como se indica en las preferencias del usuario.
De manera similar, si asumimos que hacemos una pregunta distinta sobre generar un plan de comidas, el agente es capaz de recuperar información relevante de la memoria y proporcionar una respuesta personalizada (vegana).
stream = assistant_agent.run_stream(task="Write brief meal recipe with broth")
await Console(stream)---------- user ----------
Write brief meal recipe with broth
---------- assistant_agent ----------
[MemoryContent(content='The weather should be in metric units', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None), MemoryContent(content='Meal recipe must be vegan', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None)]
---------- assistant_agent ----------
Here's a brief vegan broth recipe:
**Vegan Vegetable Broth**
**Ingredients:**
- 2 tablespoons olive oil
- 1 large onion, chopped
- 2 carrots, sliced
- 2 celery stalks, sliced
- 4 cloves garlic, minced
- 1 teaspoon salt
- 1/2 teaspoon pepper
- 1 bay leaf
- 1 teaspoon thyme
- 1 teaspoon rosemary
- 8 cups water
- 1 cup mushrooms, sliced
- 1 cup chopped leafy greens (e.g., kale, spinach)
- 1 tablespoon soy sauce (optional)
**Instructions:**
1. **Sauté Vegetables:** In a large pot, heat olive oil over medium heat. Add the onion, carrots, and celery, and sauté until they begin to soften, about 5-7 minutes.
2. **Add Garlic & Seasonings:** Stir in the garlic, salt, pepper, bay leaf, thyme, and rosemary. Cook for another 2 minutes until fragrant.
3. **Simmer Broth:** Pour in the water, add mushrooms and soy sauce (if using). Increase heat and bring to a boil. Reduce heat and let it simmer for 30-45 minutes.
4. **Add Greens:** In the last 5 minutes of cooking, add the chopped leafy greens.
5. **Strain & Serve:** Remove from heat, strain out the vegetables (or leave them in for a chunkier texture), and adjust seasoning if needed. Serve hot as a base for soups or enjoy as is!
Enjoy your flavorful, nourishing vegan broth!TaskResult(messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Write brief meal recipe with broth', type='TextMessage'), MemoryQueryEvent(source='assistant_agent', models_usage=None, metadata={}, content=[MemoryContent(content='The weather should be in metric units', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None), MemoryContent(content='Meal recipe must be vegan', mime_type=<MemoryMimeType.TEXT: 'text/plain'>, metadata=None)], type='MemoryQueryEvent'), TextMessage(source='assistant_agent', models_usage=RequestUsage(prompt_tokens=208, completion_tokens=331), metadata={}, content="Here's a brief vegan broth recipe:\n\n**Vegan Vegetable Broth**\n\n**Ingredients:**\n- 2 tablespoons olive oil\n- 1 large onion, chopped\n- 2 carrots, sliced\n- 2 celery stalks, sliced\n- 4 cloves garlic, minced\n- 1 teaspoon salt\n- 1/2 teaspoon pepper\n- 1 bay leaf\n- 1 teaspoon thyme\n- 1 teaspoon rosemary\n- 8 cups water\n- 1 cup mushrooms, sliced\n- 1 cup chopped leafy greens (e.g., kale, spinach)\n- 1 tablespoon soy sauce (optional)\n\n**Instructions:**\n\n1. **Sauté Vegetables:** In a large pot, heat olive oil over medium heat. Add the onion, carrots, and celery, and sauté until they begin to soften, about 5-7 minutes.\n\n2. **Add Garlic & Seasonings:** Stir in the garlic, salt, pepper, bay leaf, thyme, and rosemary. Cook for another 2 minutes until fragrant.\n\n3. **Simmer Broth:** Pour in the water, add mushrooms and soy sauce (if using). Increase heat and bring to a boil. Reduce heat and let it simmer for 30-45 minutes.\n\n4. **Add Greens:** In the last 5 minutes of cooking, add the chopped leafy greens.\n\n5. **Strain & Serve:** Remove from heat, strain out the vegetables (or leave them in for a chunkier texture), and adjust seasoning if needed. Serve hot as a base for soups or enjoy as is!\n\nEnjoy your flavorful, nourishing vegan broth! TERMINATE", type='TextMessage')], stop_reason=None)Almacenes de Memoria Personalizados (Bases de Datos Vectoriales, etc.)
Puedes construir tu propia implementación del protocolo Memory para incorporar almacenes más complejos. Por ejemplo, un almacén que use una base de datos vectorial para recuperar información similar, o incluso un modelo de aprendizaje automático para personalizar respuestas.
Debes sobrecargar los métodos add, query y update_context, y luego pasar ese almacén personalizado al agente.
Actualmente, los siguientes almacenes de memoria están disponibles como parte del paquete de Saptiva-Agents:
saptiva_agents.memory.chromadb.ChromaDBVectorMemory: Un almacenamiento de memoria que utiliza una base de datos vectorial para guardar y recuperar información.saptiva_agents.memory.chromadb.SentenceTransformerEmbeddingFunctionConfig: Una clase de configuración para la función de embedding de SentenceTransformer utilizada por el almacenamientoChromaDBVectorMemory.
import tempfile
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.ui import Console
from saptiva_agents.memory import MemoryContent, MemoryMimeType
from saptiva_agents.memory.chromadb import (
ChromaDBVectorMemory,
PersistentChromaDBVectorMemoryConfig,
SentenceTransformerEmbeddingFunctionConfig,
)
from saptiva_agents.base import SaptivaAIChatCompletionClient
# Usa un directorio temporal para la persistencia de ChromaDB
with tempfile.TemporaryDirectory() as tmpdir:
chroma_user_memory = ChromaDBVectorMemory(
config=PersistentChromaDBVectorMemoryConfig(
collection_name="preferences",
persistence_path=tmpdir, # Usa aquí el directorio temporal
k=2, # Devuelve los 2 resultados más relevantes
score_threshold=0.4, # Puntaje mínimo de similitud
embedding_function_config=SentenceTransformerEmbeddingFunctionConfig(
model_name="all-MiniLM-L6-v2" # Usa un modelo predeterminado para pruebas
),
)
)
# Agrega preferencias del usuario a la memoria
await chroma_user_memory.add(
MemoryContent(
content="El clima debe estar en unidades métricas",
mime_type=MemoryMimeType.TEXT,
metadata={"category": "preferences", "type": "units"},
)
)
await chroma_user_memory.add(
MemoryContent(
content="La receta de comida debe ser vegana",
mime_type=MemoryMimeType.TEXT,
metadata={"category": "preferences", "type": "dietary"},
)
)
model_client = SaptivaAIChatCompletionClient(
model="Saptiva Ops",
api_key="TU_SAPTIVA_API_KEY"
)
# Crea un agente asistente con memoria ChromaDB
assistant_agent = AssistantAgent(
name="assistant_agent",
model_client=model_client,
tools=[get_weather],
memory=[chroma_user_memory],
)
stream = assistant_agent.run_stream(task="¿Cuál es el clima en Nueva York?")
await Console(stream)
await model_client.close()
await chroma_user_memory.close()
---------- TextMessage (user) ----------
¿Cuál es el clima en Nueva York?
---------- MemoryQueryEvent (assistant_agent) ----------
[MemoryContent(content='El clima debe estar en unidades métricas', mime_type='MemoryMimeType.TEXT', metadata={'type': 'units', 'mime_type': 'MemoryMimeType.TEXT', 'category': 'preferences', 'score': 0.4342840313911438, 'id': 'd7ed6e42-0bf5-4ee8-b5b5-fbe06f583477'})]
---------- ToolCallRequestEvent (assistant_agent) ----------
[FunctionCall(id='call_ufpz7LGcn19ZroowyEraj9bd', arguments='{"city":"New York","units":"metric"}', name='get_weather')]
---------- ToolCallExecutionEvent (assistant_agent) ----------
[FunctionExecutionResult(content='El clima en Nueva York es de 23 °C y soleado.', name='get_weather', call_id='call_ufpz7LGcn19ZroowyEraj9bd', is_error=False)]
---------- ToolCallSummaryMessage (assistant_agent) ----------
El clima en Nueva York es de 23 °C y soleado.También puedes serializar la memoria ChromaDBVectorMemory y guardarla en disco para reutilizarla más adelante:
chroma_user_memory.dump_component().model_dump_json()'{"provider":"autogen_ext.memory.chromadb.ChromaDBVectorMemory","component_type":"memory","version":1,"component_version":1,"description":"ChromaDB-based vector memory implementation with similarity search.","label":"ChromaDBVectorMemory","config":{"client_type":"persistent","collection_name":"preferences","distance_metric":"cosine","k":2,"score_threshold":0.4,"allow_reset":false,"tenant":"default_tenant","database":"default_database","persistence_path":"/Users/victordibia/.chromadb_saptiva_agents"}}'Agente RAG: Unificándolo Todo
El patrón RAG (Retrieval Augmented Generation), común en la construcción de sistemas de IA, abarca dos fases distintas:
Indexación: Cargar documentos, dividirlos en fragmentos (chunking) y almacenarlos en una base de datos vectorial.
Recuperación: Buscar y utilizar fragmentos relevantes durante el tiempo de ejecución de una conversación.
En nuestros ejemplos anteriores, agregamos elementos manualmente a la memoria y los pasamos a nuestros agentes. En la práctica, el proceso de indexación suele estar automatizado y se basa en fuentes documentales mucho más grandes como documentación de productos, archivos internos o bases de conocimiento.
Creación de un agente RAG sencillo
Primero, creamos un indexador simple de documentos para cargarlos, segmentarlos en partes (chunks) y almacenarlos en una memoria ChromaDBVectorMemory.
import re
from typing import List
import aiofiles
import aiohttp
from saptiva_agents.memory import Memory, MemoryContent, MemoryMimeType
class SimpleDocumentIndexer:
"""Indexador básico de documentos para la memoria de Saptiva-Agents."""
def __init__(self, memory: Memory, chunk_size: int = 1500) -> None:
self.memory = memory
self.chunk_size = chunk_size
async def _fetch_content(self, source: str) -> str:
"""Obtiene contenido desde URL o archivo local."""
if source.startswith(("http://", "https://")):
async with aiohttp.ClientSession() as session:
async with session.get(source) as response:
return await response.text()
else:
async with aiofiles.open(source, "r", encoding="utf-8") as f:
return await f.read()
def _strip_html(self, text: str) -> str:
"""Elimina etiquetas HTML y normaliza espacios."""
text = re.sub(r"<[^>]*>", " ", text)
text = re.sub(r"\s+", " ", text)
return text.strip()
def _split_text(self, text: str) -> List[str]:
"""Divide el texto en partes de tamaño fijo (chunks)."""
chunks: list[str] = []
for i in range(0, len(text), self.chunk_size):
chunk = text[i : i + self.chunk_size]
chunks.append(chunk.strip())
return chunks
async def index_documents(self, sources: List[str]) -> int:
"""Indexa documentos en la memoria."""
total_chunks = 0
for source in sources:
try:
content = await self._fetch_content(source)
# Limpia HTML si el contenido lo contiene
if "<" in content and ">" in content:
content = self._strip_html(content)
chunks = self._split_text(content)
for i, chunk in enumerate(chunks):
await self.memory.add(
MemoryContent(
content=chunk,
mime_type=MemoryMimeType.TEXT,
metadata={"source": source, "chunk_index": i}
)
)
total_chunks += len(chunks)
except Exception as e:
print(f"Error al indexar {source}: {str(e)}")
return total_chunksAhora usemos nuestro indexador junto con ChromaDBVectorMemory para crear un agente RAG completo:
import os
from pathlib import Path
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.ui import Console
from saptiva_agents.memory.chromadb import (
ChromaDBVectorMemory,
PersistentChromaDBVectorMemoryConfig
)
from saptiva_agents.base import SaptivaAIChatCompletionClient
# Inicializar la memoria vectorial con ChromaDB
rag_memory = ChromaDBVectorMemory(
config=PersistentChromaDBVectorMemoryConfig(
collection_name="autogen_docs",
persistence_path=os.path.join(str(Path.home()), ".chromadb_saptiva"),
k=3, # Número máximo de resultados recuperados
score_threshold=0.4, # Umbral mínimo de similitud
)
)
await rag_memory.clear() # Limpia la memoria existente
# Indexar documentación de AutoGen en la memoria
async def index_autogen_docs() -> None:
indexer = SimpleDocumentIndexer(memory=rag_memory)
sources = [
"https://raw.githubusercontent.com/microsoft/autogen/main/README.md",
"https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/agents.html",
"https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/teams.html",
"https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/termination.html",
]
chunks: int = await indexer.index_documents(sources)
print(f"Se indexaron {chunks} fragmentos desde {len(sources)} documentos de AutoGen")
# Ejecutar indexación
await index_autogen_docs()
Se indexaron 72 fragmentos desde 4 documentos de AutoGen# Crear nuestro agente asistente RAG con la memoria indexada previamente
rag_assistant = AssistantAgent(
name="rag_assistant",
model_client=SaptivaAIChatCompletionClient(model="Saptiva Ops"),
memory=[rag_memory]
)
# Realizar una pregunta sobre AutoGen usando el agente RAG
stream = rag_assistant.run_stream(task="¿Qué es AgentChat?")
await Console(stream)
# No olvides cerrar la memoria cuando termines
await rag_memory.close()
Esta implementación proporciona un agente RAG capaz de responder preguntas basadas en la documentación de AutoGen. Cuando se realiza una consulta, el sistema de memoria recupera fragmentos relevantes y los agrega al contexto, lo que permite al asistente generar respuestas fundamentadas.
Para entornos de producción, podrías considerar:
Implementar estrategias de segmentación (chunking) más sofisticadas.
Añadir capacidades de filtrado por metadatos.
Personalizar el sistema de puntuación de recuperación.
Optimizar los modelos de embedding para tu dominio específico.
Ejemplo con Mem0Memory
Mem0Memorysaptiva_agents.memory.mem0.Mem0Memory proporciona integración con el sistema de memoria de Mem0.ai, compatible tanto con entornos en la nube como locales. Ofrece capacidades avanzadas de memoria para agentes, incluyendo recuperación eficiente y actualización automática del contexto, lo que lo hace ideal para entornos de producción.
En el siguiente ejemplo se demuestra cómo usar Mem0Memory para mantener memorias persistentes entre conversaciones:
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.ui import Console
from saptiva_agents.memory import MemoryContent, MemoryMimeType
from saptiva_agents.memory.mem0 import Mem0Memory
from saptiva_agents.base import SaptivaAIChatCompletionClient
# Inicializar memoria en la nube de Mem0 (requiere API key)
# Para uso local, usar is_cloud=False con la configuración adecuada
mem0_memory = Mem0Memory(
is_cloud=True,
limit=5, # Máximo número de recuerdos a recuperar
)
# Agregar preferencias del usuario a la memoria
await mem0_memory.add(
MemoryContent(
content="El clima debe estar en unidades métricas",
mime_type=MemoryMimeType.TEXT,
metadata={"category": "preferences", "type": "units"},
)
)
await mem0_memory.add(
MemoryContent(
content="La receta de comida debe ser vegana",
mime_type=MemoryMimeType.TEXT,
metadata={"category": "preferences", "type": "dietary"},
)
)
# Crear un agente asistente con memoria Mem0
assistant_agent = AssistantAgent(
name="assistant_agent",
model_client=SaptivaAIChatCompletionClient(
model="Saptiva Ops",
),
tools=[get_weather],
memory=[mem0_memory],
)
# Consultar sobre las preferencias alimenticias
stream = assistant_agent.run_stream(task="¿Cuáles son mis preferencias alimenticias?")
await Console(stream)El ejemplo anterior demuestra cómo puede usarse Mem0Memory junto con un agente asistente. La integración de memoria garantiza que:
Todas las interacciones del agente se almacenen en Mem0 para referencia futura.
Las memorias relevantes (como las preferencias del usuario) se recuperen automáticamente y se agreguen al contexto.
El agente mantenga un comportamiento consistente basado en las memorias almacenadas.
Mem0Memory es especialmente útil para:
Despliegues de agentes de larga duración que requieren memoria persistente.
Aplicaciones que necesitan controles de privacidad más avanzados.
Equipos que buscan una gestión de memoria unificada entre múltiples agentes.
Casos de uso que requieren filtrado avanzado de memoria y análisis.
Al igual que con ChromaDBVectorMemory, también puedes serializar las configuraciones de Mem0Memory:
# Serializar la configuración de memoria
config_json = mem0_memory.dump_component().model_dump_json()
print(f"Configuración de memoria (JSON): {config_json[:100]}...")Última actualización