Langtrace is an official ecosystem partner of MongoDB.

MongoDB Vector Search is a feature that allows you to store and search for vectors in MongoDB.

​
Setup

  1. Register for an Atlas account using your Google Account or an email address.

  2. Create and deploy a Free cluster. You can use Atlas Free clusters as a small-scale development environment to host your data. Free clusters never expire, and provide access to a subset of Atlas features.

  3. Connect to your cluster using the mongosh, the Node.js driver, the PyMongo driver, or Compass.

  4. Load the mflix dataset into your cluster.

  5. Install Langtrace, PyMongo, and OpenAI

pip install langtrace-python-sdk pymongo openai
You’ll need API keys from Langtrace and OpenAI. Sign up for Langtrace and OpenAI if you haven’t done so already.

​
Usage

here’s an example of how to use Langtrace with MongoDB:

from langtrace_python_sdk import langtrace, with_langtrace_root_span
import pymongo
import os
from openai import OpenAI

langtrace.init(api_key=os.environ["LANGTRACE_API_KEY"])
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
client = pymongo.MongoClient(os.environ["MONGO_URI"])
MODEL = "text-embedding-ada-002"


# Define a function to generate embeddings
def get_embedding(text):
    """Generates vector embeddings for the given text."""
    embedding = (
        openai_client.embeddings.create(input=[text], model=MODEL).data[0].embedding
    )
    return embedding


@with_langtrace_root_span("mongo-vector-search")
def vector_query():
    db = client["sample_mflix"]

    embedded_movies_collection = db["embedded_movies"]
    # define pipeline
    pipeline = [
        {
            "$vectorSearch": {
                "index": "vector_index",
                "path": "plot_embedding",
                "queryVector": get_embedding("time travel"),
                "numCandidates": 150,
                "limit": 10,
            }
        },
        {
            "$project": {
                "_id": 0,
                "plot": 1,
                "title": 1,
                "score": {"$meta": "vectorSearchScore"},
            }
        },
    ]

    result = embedded_movies_collection.aggregate(pipeline)
    for doc in result:
        # print(doc)
        pass


if __name__ == "__main__":
    try:
        vector_query()
    except Exception as e:
        print("error", e)
    finally:
        client.close()