Skip to main content

Voyage AI

A multi-agent Voyage AI embedding pipeline with cost tracking across 3 agents. The orchestrator preprocesses the corpus and selects the Voyage model, the embedder generates query and document embeddings with per-token cost tracking, and the analyzer performs similarity search and synthesizes results with OpenAI.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to run without any API keys. Voyage embeddings run in simulated mode.

Architecture​

Key Code​

Voyage embeddings with per-token cost tracking​

Each embedding call tracks tokens and cost at the Voyage pricing rate.

VOYAGE_COST_PER_TOKEN = 0.00000006  # ~$0.06 per 1M tokens for voyage-3

@waxell.tool(tool_type="embedding")
async def embed_query_voyage(query: str, model: str = "voyage-3") -> dict:
    tokens = len(query.split()) * 2
    cost = round(tokens * VOYAGE_COST_PER_TOKEN, 8)
    return {"model": model, "input_type": "query", "dimensions": 1024,
            "tokens": tokens, "cost": cost}

@waxell.tool(tool_type="embedding")
async def embed_documents_voyage(texts: list[str], model: str = "voyage-3") -> dict:
    tokens = sum(len(t.split()) * 2 for t in texts)
    cost = round(tokens * VOYAGE_COST_PER_TOKEN, 8)
    return {"model": model, "count": len(texts), "dimensions": 1024,
            "tokens": tokens, "cost": cost}

Analyzer with @retrieval, @reasoning, and cost-aware scoring​

The analyzer evaluates retrieval quality and tracks cumulative embedding costs.

@waxell.retrieval(source="voyage_embeddings")
def similarity_search(query: str, texts: list[str], top_k: int = 3) -> list[dict]:
    return [{"id": str(i), "text": text[:80], "score": round(0.92 - i * 0.05, 4)}
            for i, text in enumerate(texts[:top_k])]

@waxell.reasoning_dec(step="retrieval_quality_evaluation")
async def evaluate_retrieval_quality(query, results, total_docs) -> dict:
    avg_score = round(sum(r["score"] for r in results) / len(results), 4)
    return {
        "thought": f"Retrieved {len(results)}/{total_docs} documents. Avg: {avg_score}.",
        "evidence": [f"Top: \"{results[0]['text'][:60]}\" (score: {results[0]['score']})"],
        "conclusion": "Retrieval quality is strong" if avg_score > 0.8
                      else "May benefit from query expansion",
    }

waxell.score("embedding_cost", total_cost, comment="total Voyage cost")
waxell.score("retrieval_relevance", results[0]["score"], comment="top score")

What this demonstrates​

  • @waxell.observe -- parent-child agent hierarchy with automatic lineage
  • @waxell.step_dec -- corpus preprocessing with cost estimation
  • @waxell.tool -- Voyage embedding generation with tool_type="embedding" and cost tracking
  • @waxell.retrieval -- similarity search with source="voyage_embeddings"
  • @waxell.decision -- Voyage model selection (voyage-3, voyage-3-large, voyage-code-3)
  • waxell.decide() -- batch strategy decision (full_batch, chunked, streaming)
  • @waxell.reasoning_dec -- retrieval quality evaluation
  • waxell.score() -- embedding cost and retrieval relevance scores
  • Cost attribution -- per-token cost tracking at $0.06/1M tokens

Run it​

# Dry-run (no API keys needed)
cd dev/waxell-dev
python -m app.demos.voyage_agent --dry-run

# Live (real Voyage AI + OpenAI)
export OPENAI_API_KEY="sk-..."
python -m app.demos.voyage_agent

Source​

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