How to Implement Real-Time Vector Search with Redis Vector
We’re building a real-time vector search using Redis Vector to handle textual data intelligently and efficiently.
Prerequisites
- Redis 7.0+
- Python 3.11+
- Install Redis-py:
pip install redis - Install numpy:
pip install numpy
Step 1: Setting Up Redis
First off, install Redis. If you’re on macOS, you can easily install it using Homebrew. If you’re on another platform, check the official Redis installation guides. The reason to choose a specific version here is about feature compatibility; you’re aiming for at least Redis 7.0 to access the new vector capabilities.
brew install redis
Run Redis in your terminal:
redis-server
If you encounter the error could not open config file, it likely means working directories are missing or you made a typo in the command. Ensure that Redis is properly installed and try again.
Step 2: Connecting Python to Redis
Now, let’s connect our Python application to Redis. Below is the code to establish a connection.
import redis
# Create a Redis client
client = redis.Redis(host='localhost', port=6379, db=0)
# Check if the connection is successful
print("Connected to Redis:", client.ping())
When you run this code, you should see True printed on the console. If not, the connection issue could stem from a firewall blocking the port or Redis not running.
Step 3: Indexing Textual Data
For real-time vector search, we need to transform our data into vectors. Let’s create a simple vector for our textual data. We will use NumPy to generate synthetic data as vectors. Here’s how you can create an index.
import numpy as np
def create_vector(text):
# Example: creating a basic synthetic vector from the length of the text
return np.random.rand(128).astype('float32')
# Suppose we have these texts
texts = ["Hello, world!", "Redis is great for caching.", "Vector search changes the game."]
# Indexing process
for i, text in enumerate(texts):
vector = create_vector(text)
# Store the vector in Redis
client.hset(f'vector:{i}', mapping={'text': text, 'vector': vector.tobytes()})
When using the hset command, if you hit a “key already exists” error, it means you’ve already indexed that vector. Use hset with the NX option to avoid overwriting existing keys.
Step 4: Performing Real-Time Vector Searches
Now it’s time to perform real-time searches using Redis Vector. First, we’ll define a function to calculate cosine similarity.
from scipy.spatial.distance import cosine
def search_vectors(query_vector):
# Fetch all indices from Redis
keys = client.keys('vector:*')
scores = []
for key in keys:
stored_vector = np.frombuffer(client.hget(key, 'vector'), dtype='float32')
# Calculate cosine similarity
score = 1 - cosine(query_vector, stored_vector)
scores.append((client.hget(key, 'text').decode(), score))
# Sort by score
scores.sort(key=lambda x: x[1], reverse=True)
return scores
# Example search
query = create_vector("Tell me about Redis.")
result = search_vectors(query)
print(result)
This should give you a sorted list of closest vectors. If you receive a ValueError related to vector dimensions when calculating cosine similarity, double-check your vector creation function to ensure consistent sizes.
Step 5: Testing the Implementation
Before we call it done, let’s test everything works beautifully. Run the full script that connects, indexes the data, and searches. Here’s how this test setup looks:
def test_search():
# Add texts to Redis
texts = ["First vector data.", "Second vector to index.", "Third vector for testing."]
for i, text in enumerate(texts):
vector = create_vector(text)
client.hset(f'vector:{i}', mapping={'text': text, 'vector': vector.tobytes()})
# Search for a similar vector
query_vector = create_vector("Give me info on vector databases.")
results = search_vectors(query_vector)
print("Search Results:")
for text, score in results:
print(f'Text: {text}, Score: {score}')
test_search()
Run this and see if you get meaningful search results. If something feels off, go through the indexing to ensure data has made it to Redis properly.
The Gotchas
- Data Size Limits: Keep your text length within reasonable limits. Redis hashes have maximum sizes, and large vectors can create performance bottlenecks.
- Vector Format: Ensure consistent types (e.g., all vectors should be float32). Redis takes a burrito-sized bite out of your sanity if you mix types.
- Memory Issues: Redis keeps everything in memory. For massive datasets, watch your memory usage like a hawk.
- Network Latency: In a production environment, network delays can slow down search times significantly. Minimize cross-region requests.
- Version Compatibility: Not upgrading Redis could mean missing critical bug fixes or features. Check the release notes.
Full Code
import redis
import numpy as np
from scipy.spatial.distance import cosine
# Create a Redis client
client = redis.Redis(host='localhost', port=6379, db=0)
def create_vector(text):
return np.random.rand(128).astype('float32')
def search_vectors(query_vector):
keys = client.keys('vector:*')
scores = []
for key in keys:
stored_vector = np.frombuffer(client.hget(key, 'vector'), dtype='float32')
score = 1 - cosine(query_vector, stored_vector)
scores.append((client.hget(key, 'text').decode(), score))
scores.sort(key=lambda x: x[1], reverse=True)
return scores
def test_search():
texts = ["First vector data.", "Second vector to index.", "Third vector for testing."]
for i, text in enumerate(texts):
vector = create_vector(text)
client.hset(f'vector:{i}', mapping={'text': text, 'vector': vector.tobytes()})
query_vector = create_vector("Give me info on vector databases.")
results = search_vectors(query_vector)
print("Search Results:")
for text, score in results:
print(f'Text: {text}, Score: {score}')
test_search()
What’s Next
Next, experiment with integrating a more complex dataset. Adding real documents will give you insights into optimization and indexing techniques. Also, consider implementing batch processing for handling large amounts of data input at once.
FAQ
- Can I use Redis for non-textual vector search? Absolutely! You can index any numerical data as vectors; just make sure to define how you create them.
- How do I scale Redis for production? Horizontal scaling is your friend. Use Redis Cluster mode to distribute your data across multiple nodes.
- What’s the best way to visualize search results? Consider using libraries like Plotly or Matplotlib to plot vectors and visually interpret search outputs.
Data Sources
Last updated April 08, 2026. Data sourced from official docs and community benchmarks.
🕒 Published: