Managed Vector Services Agent

A multi-agent comparison pipeline across 3 managed vector database services (Supabase/pgvector, SingleStore, Vectara). A parent orchestrator coordinates 2 child agents -- a runner that executes insert + search operations across all services and ranks results, and an evaluator that reasons about service trade-offs, recommends a service, and synthesizes a comparison with an LLM.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to skip real API calls. All service clients are mocked for portability.

Architecture

Key Code

Managed Service Operations with @tool

The runner child agent executes insert + search across 3 managed services, each recorded as separate tool spans.

@waxell.tool(tool_type="vector_db")
def supabase_insert(sb_client, documents: list, dim: int):
    """Insert documents into Supabase pgvector table."""
    result = sb_client.table("documents").insert([...]).execute()
    return {"inserted": len(result.data)}

@waxell.tool(tool_type="vector_db")
def supabase_match(sb_client, dim: int, match_count: int = 3):
    """Run Supabase RPC match_documents for similarity search."""
    result = sb_client.rpc("match_documents", {"query_embedding": [...], "match_count": match_count})
    return {"matches": len(result.data), "top_similarity": result.data[0]["similarity"]}

@waxell.tool(tool_type="vector_db")
def singlestore_search(ss_conn, dim: int, limit: int = 3):
    """Run SingleStore DOT_PRODUCT vector similarity search."""
    cursor = ss_conn.cursor()
    cursor.execute("SELECT id, title, DOT_PRODUCT(embedding, ...) AS score FROM documents ORDER BY score DESC")
    return {"rows": len(rows), "top_score": rows[0]["score"]}

@waxell.tool(tool_type="vector_db")
def vectara_query(vectara_client, query: str, corpus_key: str = "demo-corpus"):
    """Run Vectara RAG query with summary generation."""
    result = vectara_client.query(query=query, corpus_key=corpus_key)
    return {"results": len(result.search_results), "has_summary": bool(result.summary)}

Service Ranking and Evaluation

@waxell.retrieval(source="managed-vector-services")
def rank_service_results(service_results: dict) -> list[dict]:
    """Rank results across managed vector services by latency."""
    ranked = [{"service": name, "results": r["results"], "latency_ms": r["latency_ms"],
               "top_score": r["top_score"], "backend": r["backend"]}
              for name, r in service_results.items()]
    ranked.sort(key=lambda x: x["latency_ms"])
    return ranked

@waxell.reasoning_dec(step="service_evaluation")
async def evaluate_services(comparison: list, query: str) -> dict:
    fastest = comparison[0]["service"]
    best_quality = max(comparison, key=lambda x: x["top_score"])["service"]
    return {
        "thought": f"Evaluated {len(comparison)} managed services. Fastest: {fastest}.",
        "evidence": [f"{c['service']}: {c['latency_ms']}ms, backend={c['backend']}" for c in comparison],
        "conclusion": f"{fastest} lowest latency; {best_quality} highest relevance.",
    }

What this demonstrates

• @waxell.observe -- parent-child agent hierarchy (orchestrator + 2 child agents) with automatic lineage via WaxellContext
• @waxell.tool(tool_type="vector_db") -- 6 tool spans across 3 services (insert + search each for Supabase, SingleStore, Vectara)
• @waxell.retrieval(source="managed-vector-services") -- cross-service result ranking by latency
• @waxell.decision -- service selection via OpenAI (all, supabase_only, singlestore_only, vectara_only)
• waxell.decide() -- ranking strategy and service recommendation decisions
• @waxell.reasoning_dec -- service trade-off evaluation across latency, relevance, and backend type
• waxell.score() -- query coverage and recommendation confidence scores
• Auto-instrumented LLM calls -- OpenAI synthesis captured without extra code
• 3 managed services -- Supabase (pgvector), SingleStore (DOT_PRODUCT), Vectara (RAG-as-a-Service) compared

Run it

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

# Live mode
export OPENAI_API_KEY="sk-..."
python -m app.demos.managed_vector_agent

# Custom query
python -m app.demos.managed_vector_agent --dry-run --query "Best managed vector DB for RAG"

Source

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