Voyage AI Integration

pip install voyageai moorcheh-sdk python-dotenv

MOORCHEH_API_KEY=your_moorcheh_key
VOYAGE_API_KEY=your_voyage_key

import os
import textwrap
from typing import List

import voyageai
from dotenv import load_dotenv
from moorcheh_sdk import MoorchehClient

load_dotenv()

MOORCHEH_API_KEY = os.getenv("MOORCHEH_API_KEY", "").strip()
VOYAGE_API_KEY = os.getenv("VOYAGE_API_KEY", "").strip()
if not MOORCHEH_API_KEY or not VOYAGE_API_KEY:
    raise SystemExit("Set MOORCHEH_API_KEY and VOYAGE_API_KEY.")

MODEL = "voyage-4"
VECTOR_DIMENSION = 1024
NAMESPACE = "voyage-embed-demo"
CHUNK_SIZE = 900
CHUNK_OVERLAP = 180


def to_float_vector(values: List[float]) -> List[float]:
    return [float(x) for x in values]


def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
    chunks: List[str] = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunks.append(text[start:end].strip())
        if end == len(text):
            break
        start = max(end - overlap, 0)
    return [c for c in chunks if c]


def extract_text(result: dict) -> str:
    if result.get("text"):
        return str(result["text"])
    metadata = result.get("metadata") or {}
    if isinstance(metadata, dict):
        return str(metadata.get("text") or metadata.get("raw_text") or metadata.get("content") or "")
    return ""


def clean_text(text: str) -> str:
    return " ".join(str(text).split())


def print_result(idx: int, result: dict) -> None:
    metadata = result.get("metadata") or {}
    text_value = clean_text(extract_text(result))
    wrapped = textwrap.fill(text_value, width=100)
    print(f"[{idx}] id={result.get('id')}")
    print(f"score={result.get('score')} label={result.get('label')}")
    print(f"section={metadata.get('section')} source_doc_id={metadata.get('source_doc_id')}")
    print("text:")
    print(wrapped if wrapped else "(no text returned)")
    print("-" * 120)


vo = voyageai.Client(api_key=VOYAGE_API_KEY)
mc = MoorchehClient(api_key=MOORCHEH_API_KEY)

try:
    mc.namespaces.create(
        namespace_name=NAMESPACE,
        type="vector",
        vector_dimension=VECTOR_DIMENSION,
    )
except Exception:
    pass

source_documents = [
    {
        "id": "guide-vector-namespaces",
        "section": "vector-namespace-best-practices",
        "text": (
            "Moorcheh vector namespaces support bring-your-own-embedding workflows. "
            "Use Voyage with input_type document for chunks and query for search strings; match vector_dimension to the embedding size."
        ),
    },
    {
        "id": "guide-search-tuning",
        "section": "semantic-search-tuning",
        "text": (
            "Tune similarity_search top_k and threshold for your use case."
        ),
    },
]

documents = []
for doc in source_documents:
    parts = chunk_text(doc["text"])
    for idx, chunk in enumerate(parts):
        documents.append(
            {
                "id": f"{doc['id']}-chunk-{idx}",
                "text": chunk,
                "source_doc_id": doc["id"],
                "section": doc["section"],
                "chunk_index": idx,
                "total_chunks": len(parts),
            }
        )

texts = [d["text"] for d in documents]
doc_result = vo.embed(texts, model=MODEL, input_type="document")

mc.vectors.upload(
    namespace_name=NAMESPACE,
    vectors=[
        {
            "id": documents[i]["id"],
            "vector": to_float_vector(doc_result.embeddings[i]),
            "text": documents[i]["text"],
            "source": "voyage",
            "model": MODEL,
            "section": documents[i]["section"],
            "source_doc_id": documents[i]["source_doc_id"],
            "chunk_index": documents[i]["chunk_index"],
            "total_chunks": documents[i]["total_chunks"],
        }
        for i in range(len(documents))
    ],
)

query = "How do I use Voyage input_type with Moorcheh?"
q_result = vo.embed([query], model=MODEL, input_type="query")
query_vec = to_float_vector(q_result.embeddings[0])

results = mc.similarity_search.query(
    namespaces=[NAMESPACE],
    query=query_vec,
    top_k=5,
    kiosk_mode=True,
    threshold=0.15,
)

print(f"namespace={NAMESPACE} total_results={len(results.get('results', []))}")
print("=" * 120)
for idx, r in enumerate(results.get("results", []), start=1):
    print_result(idx, r)