Sustituir la memoria de CrewAI en producción por VectorAI DB

Si has implementado una aplicación de CrewAI con memory=True y estás viendo "database is locked" Los errores bajo carga simultánea, la pérdida de memoria tras el reinicio de un contenedor o la fuga de memoria entre usuarios en un entorno multitenant se deben, en los tres casos, a la misma causa: el backend de memoria predeterminado de CrewAI no resiste en condiciones de producción.

Este tutorial te muestra cómo sustituirlo por VectorAI DB. El cambio requiere un nuevo archivo y dos líneas en la instanciación de Crew. Tus agentes, tareas y lógica de Crew se mantienen exactamente igual.

Requisitos previos

Antes de empezar, necesitarás:

• CrewAI 1.14.6 instalado
• Docker está instalado y en funcionamiento
• VectorAI DB Community Edition ejecutándose localmente
• Python 3.10 o superior
• Una clave de API de OpenAI

Por qué falla el backend de memoria predeterminado en producción

El backend de memoria predeterminado de CrewAI funciona bien en el entorno de desarrollo. En condiciones de producción, presenta tres fallos concretos.

Bloqueo concurrente

Las versiones actuales de CrewAI utilizan LanceDB con un mecanismo de reintento. Esto reduce el problema, pero no lo elimina. La guía de configuración de la memoria de producción de mem0 indica que la ejecución de varios equipos en paralelo con almacenamiento compartido puede seguir generando errores del tipo «la base de datos está bloqueada». Un número excesivo de escritores simultáneos también puede agotar el límite de reintentos de LanceDB y provocar fallos en las operaciones de escritura. Las versiones anteriores de CrewAI utilizaban ChromaDB como backend vectorial predeterminado, que tiene sus propias limitaciones de un solo subproceso bajo carga simultánea. Para conocer con más detalle cómo afectan estos límites de concurrencia a las implementaciones de agentes en producción, consulta nuestra comparación de bases de datos vectoriales integradas.

Almacenamiento temporal en contenedores

La ubicación de almacenamiento predeterminada está vinculada al equipo. Sin un montaje explícito del volumen, el directorio local de LanceDB desaparece cuando se reinicia el contenedor, y toda la memoria guardada se pierde con él. Tal y como señala TechJack Solutions en su guía de producción de CrewAI, «el almacenamiento local predeterminado es efímero en los contenedores». Lo hemos comprobado directamente, y el script de prueba está disponible en GitHub. Al borrar el directorio de almacenamiento, se eliminó toda la memoria guardada sin posibilidad de recuperación.

No hay aislamiento por usuario

Como el equipo de mem0 señala que «no existe aislamiento por usuario para los tipos de memoria de CrewAI». En nuestra prueba, un... sin ámbito recall() La consulta con dos usuarios de la misma colección devolvió los registros privados de ambos usuarios en el mismo conjunto de resultados.

Los tres fallos tienen su origen en la misma causa: el backend de almacenamiento predeterminado. A continuación te explicamos cómo sustituirlo.

Cómo funciona la configuración de la memoria externa de CrewAI

Hay dos aspectos que dejan claro el intercambio: cómo inicializa CrewAI la memoria y dónde se produce la sustitución.

Cuando configuras memory=True En un Crew, CrewAI inicializa automáticamente un Memory instancia respaldada por LanceDB. Guarda los registros tras la ejecución de las tareas, recupera el contexto relevante antes de cada turno del agente y utiliza una cola de escritura en segundo plano para que los guardados no bloqueen la ejecución del agente.

La clase `Memory` admite un parámetro de almacenamiento que puede ser cualquier objeto que implemente el protocolo `StorageBackend`. Este protocolo define la interfaz a la que recurre CrewAI al leer y escribir en la memoria. El fragmento de código que se muestra a continuación implementa save() y search(). Puedes encontrar el resto de métodos del protocolo en el Repositorio de GitHub. Cuando pasas un objeto de almacenamiento personalizado, CrewAI canaliza todas las operaciones de memoria a través de él, en lugar de utilizar el backend predeterminado de LanceDB.

El aislamiento por usuario funciona a través de la scope parámetro. Cada registro de memoria lleva asociada una ruta de ámbito, y cada operación de recuperación filtra según dicha ruta. Al pasar una ruta de ámbito como /user/alice, donde alice es el identificador único del usuario; en cada operación de escritura y lectura, los recuerdos de Alice nunca aparecen en los resultados de Bob, y viceversa.

Los pasos que se indican a continuación solo sustituyen el backend de memoria vectorial.

Paso 1: Instalar las dependencias

Instala los dos paquetes a la vez:

pip install "crewai==1.14.6" actian-vectorai-client

En nuestro entorno de pruebas, estos dos paquetes declararon requisitos de Protobuf que entraban en conflicto. La cadena de dependencias de CrewAI fija protobuf<6.0 a través de opentelemetry-proto==1.34.1, mientras que actian-vectorai-client Requiere el entorno de ejecución de Gencode 6.33. El esquema de versiones de Protobuf implica que un 7.x El motor de ejecución de Python cumple con un requisito de gencode 6.33, ya que los motores de ejecución más recientes son compatibles con versiones anteriores de gencode. Fija Protobuf a la versión que hemos validado:

pip install "protobuf==7.35.1"

Verás un pip check advertencia sobre opentelemetry-proto A continuación. Se trata únicamente de una advertencia sobre una restricción de metadatos. Hemos comprobado que CrewAI y los exportadores de OpenTelemetry se importan y se ejecutan correctamente con protobuf 7.35.1.

Paso 2: Iniciar VectorAI DB

Carga la imagen e inicia el contenedor con un volumen montado para que la memoria se conserve tras cada reinicio:

docker pull actian/vectorai:latest
docker run -d --name vectorai \
  -v ./local_data:/var/lib/actian-vectorai \
  -p 6573-6575:6573-6575 \
  -e ACTIAN_VECTORAI_ACCEPT_EULA=YES \
  actian/vectorai:latest

El montaje del volumen (-v ./local_data:/var/lib/actian-vectorai) es lo que hace que la memoria se mantenga tras cada reinicio del contenedor. Sin ello, VectorAI DB pierde todos los datos almacenados cuando el contenedor se detiene, lo que vuelve a plantear el mismo problema de almacenamiento efímero que se pretende sustituir.

Una vez que se inicia el contenedor, el puerto gRPC en 6574 es lo que VectorAIStorage a la que se conecta. La interfaz de usuario local está disponible en http://localhost:6575.

Paso 3: Crear el proveedor de memoria personalizado

VectorAIStorage implementa el StorageBackend protocolo que CrewAI’s Memory lo que espera la clase. El archivo que aparece a continuación recoge los métodos que CrewAI invoca al guardar y recuperar registros de memoria. Crea un archivo llamado vectorai_storage.py en el directorio raíz de tu proyecto. El archivo completo está disponible en GitHub, y los métodos que se exponen a continuación abarcan las decisiones fundamentales de diseño.

Las rutas de Scope contienen identificadores de usuario que pueden proceder de datos introducidos por el usuario. Elimina los caracteres de barra vertical de cualquier identificador de usuario antes de pasarlo a VectorAIStorage para evitar que se eluda el filtro de ámbito.

import threading
import uuid
from actian_vectorai import VectorAIClient, VectorParams, Distance, PointStruct, Filter, Field
from actian_vectorai.exceptions import CollectionExistsError
from crewai.memory.types import MemoryRecord, ScopeInfo
# json, datetime, and Any are used in _record_to_payload and _payload_to_record
# in the full file on GitHub

VECTOR_DIM = 1536  # matches OpenAI text-embedding-3-small, CrewAI's default embedder
COLLECTION_NAME = "crewai_memories"
_collection_lock = threading.Lock()

# _record_to_payload and _payload_to_record are defined in the full file on GitHub
# https://github.com/Tiioluwani/crewai-vectorai-memory

def _build_scope_ancestors(scope: str) -> list[str]:
    parts = scope.strip("/").split("/")
    ancestors: list[str] = ["/"]
    current = ""
    for part in parts:
        if part:
            current = f"{current}/{part}"
            ancestors.append(current)
    return ancestors

class VectorAIStorage:
    def __init__(self, host: str = "localhost:6574", collection: str = COLLECTION_NAME) -> None:
        self._host = host
        self._collection = collection
        self._client = VectorAIClient(host)
        self._client.connect()
        self._ensure_collection()

    def close(self) -> None:
        self._client.shutdown()

    def _ensure_collection(self) -> None:
        with _collection_lock:
            try:
                self._client.collections.create(
                    self._collection,
                    vectors_config=VectorParams(size=VECTOR_DIM, distance=Distance.Cosine),
                )
            except CollectionExistsError:
                pass

    def _scope_filter(self, scope_prefix: str | None) -> Filter | None:
        if not scope_prefix or not scope_prefix.strip("/"):
            return None
        prefix = scope_prefix.rstrip("/")
        if not prefix.startswith("/"):
            prefix = "/" + prefix
        return Filter(must=[Field("scope_ancestors_str").text(f"|{prefix}|")])

    def save(self, records: list[MemoryRecord]) -> None:
        if not records:
            return
        points = []
        for record in records:
            vector = record.embedding if record.embedding else [0.0] * VECTOR_DIM
            points.append(
                PointStruct(
                    id=str(uuid.uuid4()),
                    vector=vector,
                    payload=self._record_to_payload(record),
                )
            )
        self._client.points.upsert(self._collection, points)

    def search(
        self,
        query_embedding: list[float],
        scope_prefix: str | None = None,
        categories: list[str] | None = None,
        metadata_filter: dict[str, Any] | None = None,
        limit: int = 10,
        min_score: float = 0.0,
    ) -> list[tuple[MemoryRecord, float]]:
        fetch_limit = max(limit * 20, 200) if (scope_prefix or categories or metadata_filter) else limit
        results = self._client.points.search(
            self._collection,
            vector=query_embedding,
            limit=fetch_limit,
            filter=self._scope_filter(scope_prefix),
        )
        out: list[tuple[MemoryRecord, float]] = []
        for hit in results:
            score = float(hit.score)
            if score < min_score:
                continue
            record = self._payload_to_record(hit.payload)
            if categories and not any(c in record.categories for c in categories):
                continue
            if metadata_filter and not all(
                record.metadata.get(k) == v for k, v in metadata_filter.items()
            ):
                continue
            out.append((record, score))
            if len(out) >= limit:
                break
        return out

El scope_ancestors_str El campo almacena cada ruta de antepasados como una cadena delimitada por barras verticales, de modo que VectorAI DB pueda filtrar por ámbito en el lado del servidor sin necesidad de un operador de prefijo nativo. search() La cantidad de resultados recuperados se multiplica por 20 cuando hay un filtro, ya que VectorAI DB aplica el filtro tras clasificar una ventana de candidatos interna, y no antes. Una búsqueda filtrada por ámbito puede devolver menos resultados de los que existen cuando los registros del usuario quedan fuera de esa ventana, pero nunca devuelve registros de un usuario equivocado. Llama a close() cuando tu aplicación se cierre para liberar la conexión gRPC.

Paso 4: Configurar Crew para que utilice el proveedor personalizado

Con VectorAIStorage Una vez hecho esto, crea un archivo llamado main.py en el directorio raíz de tu proyecto y añade lo siguiente:

from crewai import Crew, Agent, Task, Process
from crewai.memory import Memory
from vectorai_storage import VectorAIStorage

# Replace with your actual user identifier
user_id = "alice"

storage = VectorAIStorage(host="localhost:6574")

crew = Crew(
    agents=[
        Agent(
            role="Research Analyst",
            goal="Research and summarize topics accurately",
            backstory="You are an experienced research analyst.",
            llm="gpt-4o-mini",
        )
    ],
    tasks=[
        Task(
            description="Summarize the latest developments in vector databases.",
            expected_output="A concise summary of key developments.",
        )
    ],
    memory=Memory(
        storage=storage,
        root_scope=f"/user/{user_id}",
    ),
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)
storage.close()

El root_scope El parámetro delimita cada operación de memoria a /user/alice. En una implementación multitenant, crea uno VectorAIStorage instancia por solicitud y pasar el identificador del usuario actual como el root_scope. Dos instancias de Crew ejecutándose simultáneamente con diferentes root_scope Los valores almacenan y recuperan la memoria en espacios de nombres totalmente independientes.

Paso 5: Comprueba que se hayan resuelto los tres modos de fallo

Antes de cambiar el backend, este es el resultado de ejecutar el conjunto de pruebas con el almacenamiento LanceDB predeterminado de CrewAI:

Con VectorAI DB como backend, crea un archivo llamado test_failure_modes.py en el directorio raíz de tu proyecto. A continuación, añade lo siguiente:

import threading
from vectorai_storage import VectorAIStorage
from crewai.memory.types import MemoryRecord

def make_record(content, scope, n):
    return MemoryRecord(
        content=content,
        scope=scope,
        categories=["test"],
        importance=0.5,
        # Embeddings are synthetic. Isolation in Test 3 depends on the
        # scope filter, not vector similarity.
        embedding=[float(n % 10) / 10.0] * 1536,
    )

# Test 1: Concurrent writes
print("=== Test 1: Concurrent writes ===")
errors = []

def write_memories(user_id, n):
    try:
        storage = VectorAIStorage()
        for i in range(5):
            storage.save([make_record(f"Memory {i} for user {user_id}", f"/user/{user_id}", i)])
        storage.close()
    except Exception as e:
        errors.append(str(e))

threads = [threading.Thread(target=write_memories, args=(f"user{i}", i)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

if errors:
    print(f"FAIL: {len(errors)} concurrent write error(s): {errors[0]}")
else:
    print("PASS: 5 concurrent writers completed without errors")

# Test 2: Persistence across reconnect
print("\n=== Test 2: Persistence across reconnect ===")
storage1 = VectorAIStorage()
storage1.save([make_record("Persistent memory test", "/user/persist_test", 1)])
storage1.close()

storage2 = VectorAIStorage()
results = storage2.search(query_embedding=[0.1] * 1536, scope_prefix="/user/persist_test", limit=5)
if results:
    print(f"PASS: Memory persisted across reconnect: {results[0][0].content[:50]}")
else:
    print("FAIL: Memory not found after reconnect")
storage2.delete(scope_prefix="/user/persist_test")
storage2.close()

# Test 3 verifies scope-filter isolation at the storage level.
# Application-level enforcement is handled by root_scope on the Memory class in main.py,
# which automatically prepends the user scope to every save and recall operation.
print("\n=== Test 3: Per-user isolation ===")
storage3 = VectorAIStorage()
storage3.save([make_record("Alice preference: prefers dark mode", "/user/alice", 1)])
storage3.save([make_record("Bob preference: speaks Spanish", "/user/bob", 2)])

alice_results = storage3.search(query_embedding=[0.1] * 1536, scope_prefix="/user/alice", limit=5)
bob_results = storage3.search(query_embedding=[0.2] * 1536, scope_prefix="/user/bob", limit=5)

alice_contents = [r.content for r, _ in alice_results]
bob_contents = [r.content for r, _ in bob_results]

alice_leaked = any("Bob" in c for c in alice_contents)
bob_leaked = any("Alice" in c for c in bob_contents)

if not alice_leaked and not bob_leaked:
    print(f"PASS: Per-user isolation confirmed: Alice sees {len(alice_results)} record(s), Bob sees {len(bob_results)} record(s), no cross-contamination")
else:
    print(f"FAIL: Memory leaked — Alice results: {alice_contents}, Bob results: {bob_contents}")

storage3.delete(scope_prefix="/user/alice")
storage3.delete(scope_prefix="/user/bob")
storage3.close()

print("\nAll failure mode tests complete.")

Las tres pruebas confirman que VectorAI DB resuelve los tres modos de fallo en producción. VectorAI DB gestiona las escrituras simultáneas sin bloqueos bajo la carga de prueba, conserva la memoria tras las reconexiones de los clientes y limita el alcance de cada operación de recuperación al usuario al que pertenece.

Qué aborda este patrón y qué hacer con el resto

Hay tres ámbitos que quedan fuera del alcance de este intercambio.

Memoria a largo plazo: Esto sigue utilizando SQLite a través de KickoffTaskOutputsSQLiteStorage. Esta capa almacena los resultados de la ejecución de las tareas en todas las ejecuciones y se encuentra fuera de lo que VectorAIStorage toques. Si tu implementación necesita memoria a largo plazo para que los datos se mantengan tras los reinicios del contenedor, monta un volumen para el archivo SQLite o sustituye esa capa por separado.

Calidad de la extracción de recuerdos: Esto depende del proceso de análisis de LLM integrado en CrewAI, que en este tutorial no se modifica. El LLM deduce el alcance, las categorías y la importancia en cada guardado. Si tus agentes almacenan recuerdos de baja calidad o irrelevantes, se trata de un problema de extracción, no de almacenamiento.

Latencia de recuperación y presupuestos de tokens: Estos aumentan a medida que crece el tamaño de la memoria. VectorAIStorage deja los límites de recuperación sin configurar. Tal y como se indica en el paso 3, search() Se produce una sobrecarga de 20 veces cuando hay un filtro. Ajuste limit Aumentar el valor sin tener en cuenta ese multiplicador da lugar a resultados incoherentes a gran escala. En las implementaciones de gran volumen se debe ajustar el limit parámetro en las operaciones de recuperación y supervisar la cantidad de contexto recuperado que recibe cada turno de agente.

Conclusión

Al sustituir el backend de memoria predeterminado de CrewAI por VectorAI DB se solucionan los tres modos de fallo que provocan que memory=True poco fiable en producción: bloqueo concurrente, almacenamiento efímero y falta de aislamiento por usuario. El cambio principal consiste en un nuevo archivo y un argumento de configuración en tu Crew. Tus agentes, tareas y lógica de Crew permanecen exactamente igual que antes.

Puedes añadir una capa de extracción de memoria semántica con mem0 sobre el backend persistente y aislado que acabas de crear. La Edición Comunitaria es gratuita para empezar.

Preguntas frecuentes

¿Funciona este tutorial con CrewAI 1.14.7?

Este tutorial está pensado para CrewAI 1.14.6 y no se ha probado con la versión 1.14.7. Antes de continuar con la versión 1.14.7, comprueba que el storage parámetro activado Memory sigue existiendo y que las firmas de los métodos en crewai/memory/storage/backend.py coincidir con lo que VectorAIStorage herramientas.

¿Qué ocurre con la memoria a largo plazo (SQLite)?

Este tutorial sustituye únicamente el backend de memoria vectorial. La memoria a largo plazo a través de KickoffTaskOutputsSQLiteStorage permanece intacto. Monta un volumen para el archivo SQLite si necesitas que se conserve tras los reinicios del contenedor.

¿Puedo utilizar este patrón con mem0 en lugar de una conexión directa a la base de datos VectorAI?

Sí, pero se trata de una vía de integración diferente. mem0 se integra a través de la external_memory parámetro, no el StorageBackend intercambio que se trata en este tutorial. Incorporación de una capa de extracción de memoria semántica con mem0 recorre ese camino.

¿Influye el uso de un proveedor de memoria externo en el rendimiento de la tripulación?

Sí. Ahora, cada guardado de memoria implica una llamada gRPC a la base de datos de VectorAI, además del proceso de análisis del LLM de CrewAI. La sobrecarga de red de una llamada gRPC es mínima en un despliegue local, pero mide tanto la latencia de guardado como la de recuperación en tu propio entorno antes de implementar en producción. Ajusta el limit parámetro en las operaciones de recuperación si el contexto recuperado ocupa demasiado espacio.

Problemas habituales

Los errores del tipo «la base de datos está bloqueada» persisten tras cambiar a VectorAI DB

Es probable que The Crew siga recurriendo al backend predeterminado. Confirma Memory(storage=VectorAIStorage(...)) se transmita correctamente en la tripulación y que VectorAIStorage se importan sin errores.

Error al instanciar el proveedor de memoria

Es probable que no se pueda acceder a VectorAI DB. Comprueba que el contenedor se esté ejecutando con docker ps y que el puerto 6574 esté abierto. Si te encuentras en un servidor remoto, introduce la dirección correcta en VectorAIStorage(host="your-host:6574").

Fugas de memoria entre usuarios tras configurar el aislamiento por usuario

Comprueba que root_scope Se configura por usuario al crear la instancia de Crew y no se comparte entre instancias. Dos instancias de Crew que compartan la misma root_scope almacenar los valores y los recuerdos en el mismo espacio de nombres.

VectorAIStorage no cumple con el protocolo StorageBackend

Ejecuta esto para comprobarlo VectorAIStorage cumple con el protocolo:

python -c "from crewai.memory.storage.backend import StorageBackend; from vectorai_storage import VectorAIStorage; print(issubclass(VectorAIStorage, StorageBackend))"

Si devuelve False, compara las firmas de los métodos en vectorai_storage.py contra crewai/memory/storage/backend.py en el paquete que tienes instalado.

Acerca del autor

Acerca de Tahiya Chowdhury

Tahiya Chowdhury es la directora de producto de Actian Zen y VectorAI DB, donde dirige la estrategia de la plataforma de datos en el borde más sólida del sector. Gracias a su experiencia en MongoDB y Goldman Sachs, Tahiya se especializa en crear productos que se sitúan en la intersección entre las necesidades empresariales a gran escala y la agilidad de los desarrolladores modernos. Le apasiona eliminar la complejidad de la infraestructura de datos, lo que permite a los equipos de ingeniería pasar más rápidamente del prototipo a la producción.