Pinecone Agent

A multi-agent pipeline for managed vector search using Pinecone. A parent orchestrator coordinates 3 child agents -- an indexer that upserts vectors, a searcher that queries by similarity and fetches by ID, and a synthesizer that generates answers with reasoning and dual decision decorators. The pipeline demonstrates SDK primitives across Pinecone-specific operations including namespace routing, upsert, query, and fetch.

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

Pinecone Upsert with @tool

The indexer child agent uses @tool(tool_type="vector_db") to upsert vectors into a Pinecone namespace with full metadata.

@waxell.tool(tool_type="vector_db")
def pinecone_upsert(vectors: list, namespace: str = "default") -> dict:
    """Upsert vectors into a Pinecone index."""
    upsert_ids = [v["id"] for v in vectors]
    return {"upserted_count": len(vectors), "ids": upsert_ids, "namespace": namespace}

@waxell.observe(agent_name="pinecone-indexer", workflow_name="vector-indexing")
async def run_indexer(vectors: list, namespace: str = "default", waxell_ctx=None):
    waxell.tag("vector_db", "pinecone")
    waxell.tag("agent_role", "indexer")
    waxell.metadata("index_name", "ai-knowledge-base")
    waxell.metadata("dimension", 1536)
    waxell.metadata("metric", "cosine")

    result = pinecone_upsert(vectors=vectors, namespace=namespace)
    waxell.metadata("upserted_count", result["upserted_count"])
    return result

Pinecone Query and Fetch with @tool and @retrieval

The searcher child agent demonstrates two Pinecone read operations -- similarity query and ID fetch -- followed by @retrieval for ranking.

@waxell.tool(tool_type="vector_db")
def pinecone_query(query_vector: list, top_k: int = 3, namespace: str = "default",
                   include_metadata: bool = True, filter_dict: dict | None = None) -> dict:
    """Query vectors by similarity in a Pinecone index."""
    return {"matches": [...], "namespace": namespace}

@waxell.tool(tool_type="vector_db")
def pinecone_fetch(ids: list, namespace: str = "default") -> dict:
    """Fetch specific vectors by ID from a Pinecone index."""
    fetched = {vid: {"id": vid, "metadata": v["metadata"]} for vid in ids ...}
    return {"vectors": fetched, "namespace": namespace}

@waxell.retrieval(source="pinecone")
def rank_matches(query: str, matches: list) -> list[dict]:
    """Rank and return matched documents from Pinecone query results."""
    ranked = []
    for m in matches:
        ranked.append({
            "id": m["id"], "title": m["metadata"]["title"],
            "source": m["metadata"]["source"], "score": m["score"],
        })
    return sorted(ranked, key=lambda d: d["score"], reverse=True)

Synthesis with @reasoning, @decision, and score()

The synthesizer uses @reasoning for quality assessment, @decision for output format selection, and score() for final metrics.

@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" if coverage > 0 else "Answer may need more grounding",
    }

@waxell.decision(name="output_format", options=["brief", "detailed", "bullet_points"])
def choose_output_format(num_docs: int, context: str) -> dict:
    format_choice = "detailed" if num_docs > 2 else "brief"
    return {"chosen": format_choice, "reasoning": f"{context} -- {format_choice} format"}

# Scores attached to the synthesizer span
waxell.score("answer_quality", 0.92, comment="auto-scored based on doc coverage")
waxell.score("source_coverage", 3 / 5, comment="fraction of corpus matched")

What this demonstrates

• @waxell.observe -- parent-child agent hierarchy (orchestrator + 3 child agents) with automatic lineage via WaxellContext
• @waxell.tool(tool_type="vector_db") -- Pinecone operations (upsert, query, fetch) recorded as tool spans
• @waxell.retrieval(source="pinecone") -- match ranking recorded with Pinecone as the source
• @waxell.decision -- both namespace selection (via LLM) and output format selection (rule-based)
• waxell.decide() -- manual indexing strategy decision (single vs batch vs streaming upsert)
• @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, index name, dimension, and metric
• Auto-instrumented LLM calls -- OpenAI calls in namespace selection and synthesis captured without extra code
• Namespace routing -- dynamic namespace selection based on query classification

Run it

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

# Live mode
export OPENAI_API_KEY="sk-..."
export WAXELL_API_KEY="your-waxell-api-key"
export WAXELL_API_URL="https://api.waxell.ai"
python -m app.demos.pinecone_agent

# Custom query
python -m app.demos.pinecone_agent --dry-run --query "Find docs about AI governance"

Source

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