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.
Use for: Hybrid cloud-native storage, vectors + SQL in one. Perfect for: Combining AI retrieval with your business database.
Copy
import osfrom camel.storages import TiDBStorage, VectorDBQuery, VectorRecordos.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.
Copy
from camel.storages import QdrantStorage, VectorDBQuery, VectorRecord# Create an instance of QdrantStorage with dimension = 4qdrant_storage = QdrantStorage(vector_dim=4, collection_name="my_collection")# Add two vector recordsqdrant_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 vectorsquery_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 vectorsqdrant_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.
Copy
from camel.storages import ChromaStorage, VectorDBQuery, VectorRecordfrom camel.types import VectorDistance# Create ChromaStorage instance with ephemeral (in-memory) clientchroma_storage = ChromaStorage( vector_dim=4, collection_name="camel_example_vectors", client_type="ephemeral", # or "persistent", "http", "cloud" distance=VectorDistance.COSINE,)# Add vector recordschroma_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 vectorsquery_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 vectorschroma_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.
Copy
import randomfrom camel.storages.vectordb_storages import ( OceanBaseStorage, VectorDBQuery, VectorRecord,)# Replace these with your OceanBase connection parametersOB_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.
Copy
from camel.storages import WeaviateStorage, VectorDBQuery, VectorRecord# Create WeaviateStorage instance with dimension = 4 using Weaviate Cloudweaviate_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.
Copy
from camel.storages.graph_storages import NebulaGraphnebula_graph = NebulaGraph("your_host", "your_username", "your_password", "your_space")# Show existing tagsquery = '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.