Stream Text Embeddings with Redpanda, OpenAI, and MongoDB

Clone this repository:

git clone https://github.com/redpanda-data/redpanda-labs.git

Change into the redpanda-labs/connect-plugins/processor/embeddings/openai/ directory:

cd redpanda-labs/connect-plugins/processor/embeddings/openai

Log in to your Redpanda Cloud account and create a new Serverless cluster.

Make a note of the bootstrap server URL.

Create a topic called documents with the default settings.

Create a new user with permissions (ACLs) to access a topic named documents and a consumer group named connect.

Add the cluster connection information to a local .env file:

cat > .env<< EOF
REDPANDA_SERVERS="<bootstrap-server-url>"
REDPANDA_USER="<username>"
REDPANDA_PASS="<password>"
REDPANDA_TOPICS="documents"

EOF

Log in to your OpenAI developer platform account and create a new Project API key.

Add the secret key to the local .env file:

cat >> .env<< EOF
OPENAI_API_KEY="<secret_key>"
OPENAI_EMBEDDING_MODEL="text-embedding-3-small"
OPENAI_MODEL="gpt-4o"

EOF

Log in to your MongoDB Atlas account and deploy a new free cluster for development purposes.

Create a new database named VectorStore, a new collection in that database named Embeddings, and an Atlas Vector Search index with the following JSON configuration:

{
  "fields": [
    {
      "numDimensions": 1536,
      "path": "embedding",
      "similarity": "euclidean",
      "type": "vector"
    }
  ]
}

Add the Atlas connection information to the local .env file:

cat >> .env<< EOF
# Connection string for MongoDB Driver for Go:
ATLAS_CONNECTION_STRING="<connection-string>"
ATLAS_DB="VectorStore"
ATLAS_COLLECTION="Embeddings"
ATLAS_INDEX="vector_index"

EOF

Use LangChain’s WebBaseLoader and RecursiveCharacterTextSplitter to generate chunks of text from the BBC Sport website and send each chunk to a Redpanda topic named documents.

Use Redpanda Connect to consume the messages from the documents topic and pass each message through a processor that calls OpenAI’s embeddings API to retrieve the vector embeddings for the text. The enriched messages are then inserted into a MongoDB Atlas database collection that has a vector search index.

Complete the RAG pipeline by using LangChain to retrieve similar texts from the MongoDB Atlas database and add that context alongside a user question to a prompt that is sent to OpenAI’s new gpt-4o model.