Qdrant Agent

A multi-agent similarity search pipeline using Qdrant as the vector store. A parent orchestrator coordinates 3 child agents -- an indexer that upserts points with payloads, a searcher that runs both vector similarity search and filter-only queries then merges results, and a synthesizer that generates answers with reasoning and quality scoring. The pipeline demonstrates SDK primitives across Qdrant-specific operations including upsert, search with score threshold, and query_points by filter.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to skip real API calls.

Architecture

Key Code

Qdrant Upsert and Search with @tool

The indexer uses @tool(tool_type="vector_db") to upsert points. The searcher combines vector similarity search (with score threshold and filters) and filter-only query_points.

@waxell.tool(tool_type="vector_db")
def qdrant_upsert(points: list, collection_name: str = "ai-safety-papers",
                  wait: bool = True) -> dict:
    """Upsert points into a Qdrant collection."""
    point_ids = [p["id"] for p in points]
    return {"upserted_count": len(points), "point_ids": point_ids}

@waxell.tool(tool_type="vector_db")
def qdrant_search(query_vector: list, collection_name: str = "ai-safety-papers",
                  limit: int = 4, score_threshold: float = 0.7,
                  query_filter: dict | None = None) -> dict:
    """Search for similar vectors in a Qdrant collection."""
    return {"results": results, "total": len(results)}

@waxell.tool(tool_type="vector_db")
def qdrant_query_points(collection_name: str = "ai-safety-papers",
                        query_filter: dict | None = None, limit: int = 2) -> dict:
    """Query points by filter (no vector) in a Qdrant collection."""
    return {"points": points, "total": len(points)}

Retrieval Ranking with @retrieval

The searcher merges vector search and filter query results, deduplicating by point ID and sorting by score.

@waxell.retrieval(source="qdrant")
def rank_results(query: str, search_results: list, query_results: list) -> list[dict]:
    """Rank and merge results from vector search and filter queries."""
    seen = set()
    ranked = []
    for r in search_results:
        if r["id"] not in seen:
            seen.add(r["id"])
            ranked.append({
                "id": r["id"], "title": r["payload"]["title"],
                "topic": r["payload"]["topic"], "score": r["score"],
                "source": "vector_search",
            })
    for r in query_results:
        if r["id"] not in seen:
            seen.add(r["id"])
            ranked.append({...})
    return sorted(ranked, key=lambda d: d["score"], reverse=True)

Quality Assessment with @reasoning and score()

@waxell.reasoning_dec(step="quality_assessment")
async def assess_quality(answer: str, documents: list) -> dict:
    doc_titles = [d.get("title", "unknown") for d in documents]
    coverage = len([t for t in doc_titles if t.lower() in answer.lower()])
    return {
        "thought": f"Generated answer references {coverage}/{len(documents)} source documents.",
        "evidence": [f"Source: {t}" for t in doc_titles],
        "conclusion": "Answer adequately covers source material",
    }

waxell.score("answer_quality", 0.91, comment="auto-scored based on doc coverage")
waxell.score("source_coverage", len(documents) / len(MOCK_POINTS))

What this demonstrates

• @waxell.observe -- parent-child agent hierarchy (orchestrator + 3 child agents) with automatic lineage via WaxellContext
• @waxell.tool(tool_type="vector_db") -- Qdrant operations (upsert, search, query_points) recorded as tool spans
• @waxell.retrieval(source="qdrant") -- merged search-and-filter ranking recorded with Qdrant as the source
• @waxell.decision -- topic classification via OpenAI and output format selection
• waxell.decide() -- manual search strategy decision (vector_search, filter_search, hybrid_search)
• @waxell.reasoning_dec -- chain-of-thought quality assessment of synthesized answers
• @waxell.step_dec -- query preprocessing recorded as an execution step
• waxell.score() -- answer quality and source coverage scores attached to the trace
• waxell.tag() / waxell.metadata() -- vector DB type, agent role, collection name, vector size, and distance metric
• Auto-instrumented LLM calls -- OpenAI calls in topic classification and synthesis captured without extra code
• Dual search modes -- vector similarity search with score threshold and filter-only query_points combined in one pipeline

Run it

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

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

# Custom query
python -m app.demos.qdrant_agent --dry-run --query "Find vectors about ML safety"

Source

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