Skip to main content

How to Set Up Vector Stores

Difficulty: ⭐ Beginner | Time: 45 minutes

🎯 The Problem​

You need to store document embeddings for semantic search, but you're overwhelmed by vector store options (OpenSearch, MongoDB, Azure, Vertex AI). You need to quickly choose one, set it up, and connect it to RecoAgent without getting bogged down in configuration details.

This guide solves: Choosing the right vector store for your needs and getting it running in under an hour.

⚡ TL;DR - Quick Start (OpenSearch)​

# 1. Start local OpenSearch
docker run -d -p 9200:9200 -e "discovery.type=single-node" -e "plugins.security.disabled=true" opensearchproject/opensearch:latest

# 2. Use in Python
from packages.rag.stores import OpenSearchStore

store = OpenSearchStore(endpoint="http://localhost:9200", index_name="my_docs")
store.add_document("doc_1", "Your content")
results = store.search("query", k=5)
print(f"✅ Found {len(results)} results!")

Expected: Search returns results within seconds!

Full Guide​

This guide shows you how to configure and use different vector store backends with RecoAgent, including OpenSearch, Azure AI Search, and Vertex AI Vector Search.

Prerequisites​

  • Python 3.8+ installed
  • RecoAgent installed: pip install recoagent
  • Access to at least one vector store service
  • API keys and credentials for your chosen service

Option 1: OpenSearch Setup​

OpenSearch is a popular open-source vector database that's easy to set up locally or in the cloud.

Local OpenSearch Setup​

# Using Docker
docker run -d \
  --name opensearch \
  -p 9200:9200 \
  -e "discovery.type=single-node" \
  -e "plugins.security.disabled=true" \
  opensearchproject/opensearch:2.11.0

# Wait for OpenSearch to start
sleep 30

Configure RecoAgent for OpenSearch​

from packages.rag.stores import OpenSearchStore
from packages.rag.chunkers import SemanticChunker
import os

# Initialize OpenSearch store
opensearch_store = OpenSearchStore(
    endpoint="http://localhost:9200",
    index_name="knowledge_base",
    embedding_dimension=1536,  # OpenAI ada-002
    username=None,  # No auth for local setup
    password=None
)

# Test connection
try:
    health = opensearch_store.health_check()
    print(f"OpenSearch health: {health}")
except Exception as e:
    print(f"Connection failed: {e}")

Add Documents to OpenSearch​

# Prepare sample documents
documents = [
    "RecoAgent is an enterprise RAG platform built with LangGraph.",
    "Hybrid retrieval combines BM25 and vector search for better results.",
    "OpenSearch provides scalable vector storage and search capabilities."
]

# Chunk documents
chunker = SemanticChunker(max_chunk_size=200, overlap=50)
chunks = []

for i, doc in enumerate(documents):
    doc_chunks = chunker.chunk_text(doc)
    chunks.extend(doc_chunks)

# Add chunks to OpenSearch
for i, chunk in enumerate(chunks):
    success = opensearch_store.add_document(
        document_id=f"doc_{i}",
        content=chunk.content,
        metadata={
            "source": f"document_{i//2}",
            "chunk_id": i,
            "type": "knowledge_base"
        }
    )
    print(f"Added chunk {i}: {success}")

print(f"Successfully added {len(chunks)} chunks to OpenSearch")

Option 2: Azure AI Search Setup​

Azure AI Search provides managed vector search capabilities with enterprise features.

  1. Azure subscription
  2. Azure AI Search service created
  3. API key from the Azure portal
from packages.rag.stores import AzureAISearchStore

# Initialize Azure AI Search store
azure_store = AzureAISearchStore(
    endpoint="https://your-search-service.search.windows.net",
    api_key=os.getenv("AZURE_SEARCH_API_KEY"),
    index_name="knowledge-base",
    embedding_dimension=1536
)

# Test connection
try:
    health = azure_store.health_check()
    print(f"Azure AI Search health: {health}")
except Exception as e:
    print(f"Connection failed: {e}")

Create Search Index​

# Define index schema
index_schema = {
    "name": "knowledge-base",
    "fields": [
        {"name": "id", "type": "Edm.String", "key": True},
        {"name": "content", "type": "Edm.String", "searchable": True},
        {"name": "content_vector", "type": "Collection(Edm.Single)", "searchable": True, "dimensions": 1536},
        {"name": "source", "type": "Edm.String", "filterable": True},
        {"name": "chunk_id", "type": "Edm.Int32", "filterable": True},
        {"name": "type", "type": "Edm.String", "filterable": True}
    ],
    "vectorSearch": {
        "algorithms": [
            {
                "name": "hnsw-config",
                "kind": "hnsw",
                "parameters": {
                    "metric": "cosine"
                }
            }
        ],
        "profiles": [
            {
                "name": "vector-profile",
                "algorithm": "hnsw-config"
            }
        ]
    }
}

# Create index
success = azure_store.create_index(index_schema)
print(f"Index created: {success}")

Option 3: Vertex AI Vector Search Setup​

Google's Vertex AI Vector Search provides enterprise-grade vector search with AutoML capabilities.

Prerequisites for Vertex AI​

  1. Google Cloud Platform account
  2. Vertex AI API enabled
  3. Service account with appropriate permissions
  4. Authentication credentials (JSON key file or default credentials)

Configure RecoAgent for Vertex AI​

from packages.rag.stores import VertexAIVectorStore
import json

# Option 1: Using service account key file
credentials_path = "path/to/service-account-key.json"
with open(credentials_path, 'r') as f:
    credentials_info = json.load(f)

vertex_store = VertexAIVectorStore(
    project_id="your-gcp-project",
    location="us-central1",
    credentials_info=credentials_info,
    index_name="knowledge-base",
    embedding_dimension=1536
)

# Option 2: Using default credentials (if running on GCP)
vertex_store = VertexAIVectorStore(
    project_id="your-gcp-project",
    location="us-central1",
    index_name="knowledge-base",
    embedding_dimension=1536
)

Create Vector Search Index​

# Define index configuration
index_config = {
    "display_name": "knowledge-base-index",
    "description": "Vector index for knowledge base documents",
    "metadata_schema_uri": "gs://your-bucket/schema.json",
    "index_update_method": "BATCH_UPDATE",
    "embedding_config": {
        "dimensions": 1536,
        "distance_measure_type": "COSINE_DISTANCE"
    }
}

# Create index
index_id = vertex_store.create_index(index_config)
print(f"Created index with ID: {index_id}")

# Create index endpoint
endpoint_config = {
    "display_name": "knowledge-base-endpoint",
    "description": "Endpoint for knowledge base vector search"
}

endpoint_id = vertex_store.create_index_endpoint(endpoint_config)
print(f"Created endpoint with ID: {endpoint_id}")

# Deploy index to endpoint
deployment_id = vertex_store.deploy_index(index_id, endpoint_id)
print(f"Deployed index with deployment ID: {deployment_id}")

Using Vector Stores with Retrievers​

Once your vector store is set up, you can use it with RecoAgent's retrievers:

from packages.rag.retrievers import VectorRetriever, HybridRetriever, BM25Retriever

# Create vector retriever
vector_retriever = VectorRetriever(
    vector_store=opensearch_store,  # or azure_store, vertex_store
    embedding_model="text-embedding-ada-002"
)

# Create BM25 retriever (if supported)
bm25_retriever = BM25Retriever(vector_store=opensearch_store)

# Create hybrid retriever
hybrid_retriever = HybridRetriever(
    vector_retriever=vector_retriever,
    bm25_retriever=bm25_retriever,
    alpha=0.7
)

# Test retrieval
query = "How does RecoAgent work?"
results = hybrid_retriever.retrieve(query, k=5)

for i, result in enumerate(results):
    print(f"{i+1}. Score: {result.score:.3f}")
    print(f"   Content: {result.chunk.content[:100]}...")
    print(f"   Source: {result.chunk.metadata.get('source', 'unknown')}")
    print()

Production Considerations​

Performance Optimization​

# Batch operations for better performance
def batch_add_documents(store, documents, batch_size=100):
    """Add documents in batches for better performance."""
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        store.batch_add_documents(batch)
        print(f"Added batch {i//batch_size + 1}")

# Use connection pooling
store = OpenSearchStore(
    endpoint="http://localhost:9200",
    index_name="knowledge_base",
    max_retries=3,
    retry_on_timeout=True,
    timeout=30
)

Monitoring and Health Checks​

import time
from datetime import datetime

def monitor_vector_store(store):
    """Monitor vector store health and performance."""
    while True:
        try:
            # Health check
            health = store.health_check()
            print(f"[{datetime.now()}] Health: {health}")
            
            # Performance test
            start_time = time.time()
            results = store.search(query_embedding=[0.1] * 1536, k=1)
            latency = (time.time() - start_time) * 1000
            print(f"[{datetime.now()}] Search latency: {latency:.2f}ms")
            
        except Exception as e:
            print(f"[{datetime.now()}] Error: {e}")
        
        time.sleep(60)  # Check every minute

# Run monitoring (in background thread)
import threading
monitor_thread = threading.Thread(target=monitor_vector_store, args=(opensearch_store,))
monitor_thread.daemon = True
monitor_thread.start()

Backup and Recovery​

def backup_vector_store(store, backup_path):
    """Backup vector store data."""
    try:
        # Export all documents
        documents = store.export_all_documents()
        
        # Save to file
        import json
        with open(backup_path, 'w') as f:
            json.dump(documents, f, indent=2)
        
        print(f"Backup saved to {backup_path}")
        return True
    except Exception as e:
        print(f"Backup failed: {e}")
        return False

def restore_vector_store(store, backup_path):
    """Restore vector store from backup."""
    try:
        import json
        with open(backup_path, 'r') as f:
            documents = json.load(f)
        
        # Re-add all documents
        for doc in documents:
            store.add_document(
                document_id=doc['id'],
                content=doc['content'],
                metadata=doc['metadata']
            )
        
        print(f"Restored {len(documents)} documents from {backup_path}")
        return True
    except Exception as e:
        print(f"Restore failed: {e}")
        return False

# Create backup
backup_vector_store(opensearch_store, "backup.json")

# Restore from backup (if needed)
# restore_vector_store(new_store, "backup.json")

Troubleshooting​

Common Issues​

Connection Errors

  • Verify endpoint URLs and ports
  • Check firewall settings
  • Ensure authentication credentials are correct

Index Creation Failures

  • Check vector dimensions match embedding model
  • Verify field mappings and types
  • Ensure sufficient permissions

Slow Performance

  • Use batch operations for bulk inserts
  • Optimize index settings for your use case
  • Consider sharding for large datasets

Memory Issues

  • Increase heap size for OpenSearch
  • Use streaming for large document sets
  • Implement pagination for search results

Getting Help​