Redis Vector Agent

A single-agent pipeline using Redis Stack for vector search. The agent creates an HNSW vector index via FT.CREATE, inserts documents with HSET, performs KNN search via FT.SEARCH, evaluates result quality, and synthesizes an answer with an LLM. Demonstrates SDK primitives across real Redis Stack vector operations.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Requires Redis Stack on localhost:6380. Use --dry-run to skip real API calls.

Architecture

Key Code

HNSW Index Creation and KNN Search with @tool

The agent creates an HNSW index with FT.CREATE and performs KNN search with FT.SEARCH dialect 2.

@waxell.tool(tool_type="vector_db", name="redis_create_index")
def create_index(r, dim: int) -> dict:
    """Create an HNSW vector index via FT.CREATE."""
    schema = (
        TextField("text"), TagField("category"),
        VectorField("embedding", "HNSW", {
            "TYPE": "FLOAT32", "DIM": dim, "DISTANCE_METRIC": "COSINE",
        }),
    )
    r.ft(INDEX_NAME).create_index(
        schema,
        definition=IndexDefinition(prefix=[PREFIX], index_type=IndexType.HASH),
    )
    return {"index": INDEX_NAME, "algorithm": "HNSW", "dimensions": dim}

@waxell.tool(tool_type="vector_db", name="redis_knn_search")
def knn_search(r, query_embedding, k: int = 3) -> dict:
    """KNN vector search via FT.SEARCH."""
    q = (
        Query(f"*=>[KNN {k} @embedding $vec AS score]")
        .sort_by("score").return_fields("text", "category", "score").dialect(2)
    )
    params = {"vec": np.array(query_embedding, dtype=np.float32).tobytes()}
    results = r.ft(INDEX_NAME).search(q, query_params=params)
    return {"hits": [...], "total": results.total}

Search Mode Decision and Quality Assessment

@waxell.decision(name="choose_search_mode", options=["knn_only", "pre_filtered"])
def choose_search_mode(query: str) -> dict:
    """Choose search mode based on query characteristics."""
    category_keywords = {"deployment", "mlops", "infrastructure", "optimization"}
    has_category = any(w in category_keywords for w in query.lower().split())
    if has_category:
        return {"chosen": "pre_filtered", "reasoning": "Query mentions category keywords"}
    return {"chosen": "knn_only", "reasoning": "No category keywords, use pure KNN"}

@waxell.reasoning_dec(step="assess_result_quality")
def assess_result_quality(results: list[dict]) -> dict:
    scores = [r["score"] for r in results]
    avg = sum(scores) / len(scores) if scores else 0
    return {
        "thought": f"Evaluating {len(results)} Redis search results",
        "evidence": [f"Average similarity: {avg:.3f}", f"Top score: {max(scores):.3f}"],
        "conclusion": "High quality results" if avg > 0.7 else "Moderate quality",
    }

What this demonstrates

• @waxell.observe -- single agent with full lifecycle tracing
• @waxell.tool(tool_type="vector_db") -- Redis Stack operations (FT.CREATE index, HSET insert, FT.SEARCH KNN) recorded as tool spans
• @waxell.retrieval(source="redis") -- search result extraction recorded with Redis as the source
• @waxell.decision -- search mode selection (knn_only vs pre_filtered)
• @waxell.reasoning_dec -- result quality assessment with average score analysis
• waxell.score() -- search quality and relevance scores attached to the trace
• waxell.tag() / waxell.metadata() -- vector DB type, index type (HNSW), and Redis host
• Auto-instrumented LLM calls -- OpenAI synthesis captured without extra code
• Real Redis Stack operations -- actual FT.CREATE, HSET, and FT.SEARCH with sentence-transformers embeddings

Run it

# Dry-run mode (no API key needed)
cd dev/waxell-dev
python -m app.demos.redis_vector_agent --dry-run

# Live mode (requires Redis Stack on localhost:6380)
export OPENAI_API_KEY="sk-..."
python -m app.demos.redis_vector_agent

# Custom query
python -m app.demos.redis_vector_agent --dry-run --query "Search for AI deployment patterns"

Source

dev/waxell-dev/app/demos/redis_vector_agent.py