Marqo

pip install netra-sdk marqo

from netra import Netra
import marqo
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Create Marqo client - automatically traced
mq = marqo.Client(
    url=os.environ.get('MARQO_URL'),
    api_key=os.environ.get('MARQO_API_KEY')
)

# Create index
mq.create_index("my-index")

# Add documents
mq.index("my-index").add_documents([
    {"title": "Document 1", "content": "Content here"}
])

from netra.decorators import task
from netra import SpanWrapper

@task()
def create_index(mq, index_name: str):
    span = SpanWrapper("marqo-create-index", {
        "index.name": index_name
    }).start()
    
    mq.create_index(
        index_name,
        model="hf/e5-base-v2"
    )
    
    span.end()

from netra.decorators import task
from netra import SpanWrapper, ActionModel

@task()
def add_documents(mq, index_name: str, documents: list):
    span = SpanWrapper("marqo-add-documents", {
        "index.name": index_name,
        "documents.count": len(documents)
    }).start()
    
    result = mq.index(index_name).add_documents(documents)
    
    span.set_action([ActionModel(
        action="insert",
        action_type="database.insert",
        success=result["status"] == "succeeded",
        affected_records=[{"id": d.get("id") or d.get("_id")} for d in documents],
        metadata={"index": index_name}
    )])
    span.set_attribute("status", result["status"])
    span.end()
    
    return result

from netra.decorators import workflow
from netra import SpanWrapper

@workflow()
def search_documents(mq, index_name: str, query: str, limit: int = 5):
    span = SpanWrapper("marqo-search", {
        "index.name": index_name,
        "query": query,
        "limit": limit
    }).start()
    
    results = mq.index(index_name).search(
        q=query,
        limit=limit
    )
    
    span.set_attribute("results.count", len(results["hits"]))
    span.end()
    
    return results

from netra.decorators import task
from netra import SpanWrapper

@task()
def multimodal_search(mq, index_name: str, query: dict):
    span = SpanWrapper("marqo-multimodal-search", {
        "index.name": index_name
    }).start()
    
    results = mq.index(index_name).search(
        q=query,
        searchable_attributes=["title", "content", "image_url"]
    )
    
    span.set_attribute("results.count", len(results["hits"]))
    span.end()
    
    return results

from netra.decorators import task
from netra import SpanWrapper

@task()
def filter_search(mq, index_name: str, query: str, filter: str):
    span = SpanWrapper("marqo-filter-search", {
        "index.name": index_name,
        "filter": filter
    }).start()
    
    results = mq.index(index_name).search(
        q=query,
        filter_string=filter
    )
    
    span.set_attribute("results.count", len(results["hits"]))
    span.end()
    
    return results