Redis

pip install netra-sdk redis

from netra import Netra
import redis
from redis.commands.search.field import VectorField
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Create Redis client - automatically traced
client = redis.Redis.from_url(os.environ.get('REDIS_URL'))

# Create index for vector search
client.ft("idx:vectors").create_index([
    VectorField("vector",
        "HNSW", {
            "TYPE": "FLOAT32",
            "DIM": 384,
            "DISTANCE_METRIC": "COSINE"
        }
    )
])

from netra import task, SpanWrapper
from redis.commands.search.field import VectorField

@task()
def create_vector_index(client, index_name: str, dimension: int):
    span = SpanWrapper("redis-create-index", {
        "index.name": index_name,
        "vector.dimension": dimension
    }).start()
    
    client.ft(index_name).create_index([
        VectorField("vector", "HNSW", {
            "TYPE": "FLOAT32",
            "DIM": dimension,
            "DISTANCE_METRIC": "COSINE"
        })
    ])
    
    span.end()

from netra import task, SpanWrapper, ActionModel
import numpy as np

@task()
def store_vector(client, key: str, vector: list[float], metadata: dict):
    span = SpanWrapper("redis-store-vector", {
        "key": key,
        "vector.size": len(vector)
    }).start()
    
    client.hset(key, mapping={
        "vector": np.array(vector, dtype=np.float32).tobytes(),
        **metadata
    })
    
    span.set_action([ActionModel(
        action="set",
        action_type="database.set",
        success=True,
        affected_records=[{"id": key}],
        metadata={"vector_size": len(vector)}
    )])
    span.set_attribute("status", "success")
    span.end()

from netra import workflow, SpanWrapper
import numpy as np

@workflow()
def search_vectors(client, index_name: str, query: list[float], limit: int = 5):
    span = SpanWrapper("redis-search", {
        "index": index_name,
        "query.size": len(query),
        "limit": limit
    }).start()
    
    query_vec = np.array(query, dtype=np.float32).tobytes()
    
    results = client.ft(index_name).search(
        f"*=>[KNN {limit} @vector $query_vec]",
        query_params={"query_vec": query_vec}
    )
    
    span.set_attribute("results.count", results.total)
    span.end()
    
    return results

from netra import task, SpanWrapper
import numpy as np

@task()
def hybrid_search(client, index_name: str, query: list[float], filter: str):
    span = SpanWrapper("redis-hybrid-search", {
        "index": index_name,
        "filter": filter
    }).start()
    
    query_vec = np.array(query, dtype=np.float32).tobytes()
    
    results = client.ft(index_name).search(
        f"{filter}=>[KNN 10 @vector $query_vec]",
        query_params={"query_vec": query_vec}
    )
    
    span.set_attribute("results.count", results.total)
    span.end()
    
    return results