​
What Are Storages in CAMEL-AI?

The Storage module in CAMEL-AI gives you a unified interface for saving, searching, and managing your data from simple key-value records to high-performance vector databases and modern graph engines. It’s your plug-and-play toolkit for building robust, AI-ready storage layers.

​
Types of Storages

​
Get Started

Here are practical usage patterns for each storage type—pick the ones you need and mix them as you like.

In-Memory Key-Value Storage

Use for: Fast, temporary storage. Data is lost when your program exits.
Perfect for: Prototyping, testing, in-memory caching.

from camel.storages.key_value_storages import InMemoryKeyValueStorage

memory_storage = InMemoryKeyValueStorage()
memory_storage.save([{'key1': 'value1'}, {'key2': 'value2'}])
records = memory_storage.load()
print(records)
memory_storage.clear()

JSON File Storage

Use for: Persistent, human-readable storage on disk.
Perfect for: Logs, local settings, configs, or sharing small data sets.

from camel.storages.key_value_storages import JsonStorage
from pathlib import Path

json_storage = JsonStorage(Path("my_data.json"))
json_storage.save([{'key1': 'value1'}, {'key2': 'value2'}])
records = json_storage.load()
print(records)
json_storage.clear()

Milvus Vector Storage

Use for: Scalable, high-performance vector search (RAG, embeddings).
Perfect for: Semantic search and production AI retrieval.

from camel.storages import MilvusStorage, VectorDBQuery, VectorRecord

milvus_storage = MilvusStorage(
    url_and_api_key=("Your Milvus URI","Your Milvus Token"),
    vector_dim=4,
    collection_name="my_collection"
)
milvus_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])
milvus_storage.load()
query_results = milvus_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)
milvus_storage.clear()

TiDB Vector Storage

Use for: Hybrid cloud-native storage, vectors + SQL in one.
Perfect for: Combining AI retrieval with your business database.

import os
from camel.storages import TiDBStorage, VectorDBQuery, VectorRecord

os.environ["TIDB_DATABASE_URL"] = "The database url of your TiDB cluster."
tidb_storage = TiDBStorage(
    url_and_api_key=(os.getenv("DATABASE_URL"), ''),
    vector_dim=4,
    collection_name="my_collection"
)
tidb_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])
tidb_storage.load()
query_results = tidb_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)
tidb_storage.clear()

Qdrant Vector Storage

Use for: Fast, scalable open-source vector search.
Perfect for: RAG, document search, and high-scale retrieval tasks.

from camel.storages import QdrantStorage, VectorDBQuery, VectorRecord

# Create an instance of QdrantStorage with dimension = 4
qdrant_storage = QdrantStorage(vector_dim=4, collection_name="my_collection")

# Add two vector records
qdrant_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])

# Query similar vectors
query_results = qdrant_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)

# Clear all vectors
qdrant_storage.clear()

ChromaDB Vector Storage

Use for: Fastest way to build LLM apps with memory and embeddings.
Perfect for: From prototyping in notebooks to production clusters with the same simple API.

from camel.storages import ChromaStorage, VectorDBQuery, VectorRecord
from camel.types import VectorDistance

# Create ChromaStorage instance with ephemeral (in-memory) client
chroma_storage = ChromaStorage(
    vector_dim=4,
    collection_name="camel_example_vectors",
    client_type="ephemeral",  # or "persistent", "http", "cloud"
    distance=VectorDistance.COSINE,
)

# Add vector records
chroma_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])

# Query similar vectors
query_results = chroma_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)

# Clear all vectors
chroma_storage.clear()

OceanBase Vector Storage

Use for: Massive vector storage with advanced analytics.
Perfect for: Batch operations, cloud or on-prem setups, and high-throughput search.

import random

from camel.storages.vectordb_storages import (
    OceanBaseStorage,
    VectorDBQuery,
    VectorRecord,
)

# Replace these with your OceanBase connection parameters
OB_URI = "127.0.0.1:2881"
OB_USER = "root@sys"
OB_PASSWORD = ""
OB_DB_NAME = "oceanbase"

def main():
    ob_storage = OceanBaseStorage(
        vector_dim=4,
        table_name="my_ob_vector_table",
        uri=OB_URI,
        user=OB_USER,
        password=OB_PASSWORD,
        db_name=OB_DB_NAME,
        distance="cosine",
    )

    status = ob_storage.status()
    print(f"Vector dimension: {status.vector_dim}")
    print(f"Initial vector count: {status.vector_count}")

    random.seed(20241023)
    large_batch = []
    for i in range(1000):
        large_batch.append(
            VectorRecord(
                vector=[random.uniform(-1, 1) for _ in range(4)],
                payload={'idx': i, 'batch': 'example'},
            )
        )
    ob_storage.add(large_batch, batch_size=100)
    status = ob_storage.status()
    print(f"Vector count after adding batch: {status.vector_count}")

    query_vector = [random.uniform(-1, 1) for _ in range(4)]
    query_results = ob_storage.query(
        VectorDBQuery(query_vector=query_vector, top_k=5)
    )
    for i, result in enumerate(query_results):
        print(f"Result {i+1}:")
        print(f"  ID: {result.record.id}")
        print(f"  Payload: {result.record.payload}")
        print(f"  Similarity: {result.similarity}")

    ob_storage.clear()
    status = ob_storage.status()
    print(f"Vector count after clearing: {status.vector_count}")

if __name__ == "__main__":
    main()

Weaviate Vector Storage

Use for: Vector search with hybrid (vector + keyword) capabilities.
Perfect for: Document retrieval and multimodal AI apps.

from camel.storages import WeaviateStorage, VectorDBQuery, VectorRecord

# Create WeaviateStorage instance with dimension = 4 using Weaviate Cloud
weaviate_storage = WeaviateStorage(
    vector_dim=4,
    collection_name="camel_example_vectors",
    connection_type="cloud",
    wcd_cluster_url="your-weaviate-cloud-url",
    wcd_api_key="your-weaviate-api-key",
    vector_index_type="hnsw",
    distance_metric="cosine",
)

weaviate_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])

query_results = weaviate_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)

weaviate_storage.clear()

NebulaGraph Storage

Use for: Open-source, distributed graph storage and querying.
Perfect for: Knowledge graphs, relationships, and fast distributed queries.

from camel.storages.graph_storages import NebulaGraph

nebula_graph = NebulaGraph("your_host", "your_username", "your_password", "your_space")

# Show existing tags
query = 'SHOW TAGS;'
print(nebula_graph.query(query))

Neo4j Graph Storage

Use for: Industry-standard graph database for large-scale relationships.
Perfect for: Enterprise graph workloads, Cypher queries, analytics.

from camel.storages import Neo4jGraph

neo4j_graph = Neo4jGraph(
    url="your_url",
    username="your_username",
    password="your_password",
)

query = "MATCH (n) DETACH DELETE n"
print(neo4j_graph.query(query))

PgVectorStorage Vector Storage

Use for: Storing and querying vectors in PostgreSQL. Perfect for: Leveraging an existing PostgreSQL database for vector search.

from camel.storages import PgVectorStorage, VectorDBQuery, VectorRecord

# Replace with your PostgreSQL connection details
PG_CONN_INFO = {
    "host": "127.0.0.1",
    "port": 5432,
    "user": "postgres",
    "password": "postgres",
    "dbname": "postgres",
}

# Create PgVectorStorage instance
pg_storage = PgVectorStorage(
    vector_dim=4,
    conn_info=PG_CONN_INFO,
    table_name="camel_example_vectors",
)

# Add vector records
pg_storage.add([
    VectorRecord(vector=[-0.1, 0.1, -0.1, 0.1], payload={'key1': 'value1'}),
    VectorRecord(vector=[-0.1, 0.1, 0.1, 0.1], payload={'key2': 'value2'}),
])

# Query similar vectors
query_results = pg_storage.query(VectorDBQuery(query_vector=[0.1, 0.2, 0.1, 0.1], top_k=1))
for result in query_results:
    print(result.record.payload, result.similarity)

# Clear all vectors
pg_storage.clear()