Community

Kanin Kearpimy

Large language models (LLMs) are the renowned solution for most business domains in 2024. While there is an estimation that [750 million applications will be integrated with LLMs in 2025](https://arc.net/l/quote/ssrfxyrx), training LLMs does consume significant monetary resources. So, the cost of LLM platforms such as OpenAI’s GPT REST API reflects the above cost in their pricing. The problem is how we can reduce the operational cost of the AI applications in production; the obvious answer is calling the API less, and then another problem arises: the maintenance of response quality to users. 

Caching has been a fundamental solution in software engineering for many years. The application creates a cache by extracting a key from the request and storing the corresponding result. Next time the same key is called, the server can immediately respond to user queries on the network without extra computation. 

However, the character of the LLM query is not a fixed key but a flexible format like the meaning of the human’s question. Consequently, the traditional cache that stores fixed keys is inefficient enough to handle LLM queries.

## Semantic cache

Unlike traditional cache, semantic cache **characteristics** of data and semantic **representation** are simply described as meaning-based representations. We call this process embedding. In LLM systems, the model converts text into numerical vectors that represent its semantic meaning. 

We will store the embeddings in the cache system. When a new request comes in, the system extracts its semantic representation by creating an embedding. It then searches for similarities between this new embedding and the stored embeddings in the cache system. If a high similarity match is found, the corresponding cached response will be returned. This process allows for semantic-based retrieval of previously computed responses, potentially reducing the need for repeated API calls to the LLM service.

![High level system and logic.][1]

## How to implement semantic cache with MongoDB Atlas and Vector Search

All of the code is in[ GitHub](https://github.com/kanin-kearpimy/llm-cache-with-mongodb).

### Prerequisites
•      Python (3.12.3 or newer)
•      FastAPI (0.11 or newer)
•      PyMongo (4.7.2 or newer)
•      uvicorn (0.29.0 or newer)

### 1) Set up dependency in Python
We need to install the dependencies as mentioned above. You can utilize pip for the package manager. The necessary dependencies are in requirements.txt. After cloning the project and entering the project’s directory, you can run the below command to install them.

```
pip install fastapi pymongo openai uvicorn
```
In case you are creating an isolated project, you can enable[ python virtualenv](https://virtualenv.pypa.io/en/latest/) for this specific environment.)

### 2) Create FastAPI server

To simulate the caching server, the request shall come from an HTTP request. We thus set up a web server in Python by FastAPI.1.2.1) Create app.py in the root directory.1.2.2) Import FastAPI and initiate / and /ask routes.

**app.py**

```python
from fastapi import FastAPI
server = FastAPI()
# root route
@server.get("/")
async def home():
  return { "message": "This is home server" }

# search route
@server.get("/ask")
async def search(query: str):
  return { "message": "The query is: {}".format(query) }
```
Next, run the application for testing our route. (--reload for hot refresh if application code is edited.)

```
uvicorn app:server --reload
```
Your server must be running at http://127.0.0.1:8000. Now, we can test our search route using the command below.

```
curl -X GET "http://127.0.0.1:8000/ask?query=hello+this+is+search+query"
```
The server must respond as below:

```
{ "message": "The query is: hello this is search query" }
```


### 3) Connect OpenAI
We previously set up a basic FastAPI server and ask route. Next, the LLM functionality will be integrated.

1.3.1) Create llm.py in the same directory as app.py.
1.3.2) Set up OpenAI as an LLM service.
**llm.py**

```python
from openai import OpenAI

open_api_key = "..."  # OpenAI API Key
openai_client = OpenAI(api_key=open_api_key)
language_model = "gpt-3.5-turbo"

# getTextResponse receive text and ask LLM for answer
def getTextResponse(text):
	chat_completion = openai_client.chat.completions.create(
    	messages=[{"role": "user", "content": text}], model=language_model
	)
	return chat_completion.choices[0].message.content
```
We have to modify app.py with a few lines of code.
**app.py**

```python
# ... other import dependencies
from llm import getTextResponse
# ... other routes
# search route
@server.get("/ask")
async def search(query):
	llm_response = getTextResponse(query)
	return {"message": "Your AI response is: {}".format(llm_response)}
```

Then, we can invoke the ask route with a new query.

```
curl -X GET "http://127.0.0.1:8000/ask?query=what+is+llm?"
Response
{
  "message": "Your AI response is: LLM stands for Master of Laws, which is an advanced law degree typically pursued by individuals who have already received a law degree (such as a JD) and want to further specialize or advance their knowledge in a specific area of law. LLM programs are typically one year in length and often focus on areas such as international law, human rights, or commercial law."
}
```


### 4) Embedding LLM response
Now, we can receive a response from the OpenAI LLM. However, the system is always relying on the OpenAI service. Our goal is to reduce the load from the AI service to the cache system.
To cache the LLM response, we must transform our text (or any type of data) into vector data.
A vector can be thought of as an N-dimensional (depending on the embedding model) array in which each number represents the meaning of the original data.
Example:

```python
text = "Large Language Model"
# embedding_function(text: str) -> vector<number>([...N])
vector = embedding_function(text)

# vector = [12, 23, 0.11, 22, 85, 43, ..., 90]
```

We can embed our data using a language model. In our case, we utilize OpenAI's text-embedding model. Therefore, we modify llm.py with a few lines.

**llm.py**

```python
# ... other code
text_embedding_model = "text-embedding-3-small"
# getEmbedding receives text and response embedding (vector) of its original data
def getEmbedding(text):
	embedding = openai_client.embeddings.create(input=text, model=text_embedding_model)
	return embedding.data[0].embedding
```

So, we shall modify app.py to use the new functionality of llm.py.
**app.py**

```python
# ... other import dependencies
from llm import getTextResponse, getEmbedding

# ... other routes

# search route
@server.get("/ask")
async def search(query):
	llm_response = getTextResponse(query)
	query_vector = getEmbedding(query)
	print("embedding : ", query_vector)
	return {"message": "Your AI response is: {}".format(llm_response)}
```

If we run the `curl` command to invoke `ask` again, in the server shell, it must print similar data like below.
```
embedding :  [-0.02254991, 0.031336114, 0.019013261, 0.00017081834, -0.0202526, -0.0020466715, 0.0111036645, -0.0111036645, 0.036172554, 0.04038429, -0.027043771, 0.0046273666, -0.039820038, -0.011456322, 0.0048339227, 0.021824444, 0.0048666694, -0.017501874, 0.03915503, 0.03895351, 0.041311275, ..., 0.046349235]
```

### 5) Store vector information in MongoDB Atlas

We already have vector data for its **semantic**. Let's see how we will store them for our cache system.MongoDB Atlas Vector Search is a key feature that lets us enable AI-powered **semantic search in vector data**. To do so, we must store documents in the MongoDB database first.First,[ register for a MongoDB Atlas account](https://mongodb.prakticum-team.ru/cloud/atlas/register). Existing users can sign into MongoDB Atlas.[Follow the instructions](https://colab.research.google.com/drive/1OAdeM2pdmp_3vt8HAyjdRQ6zNuvtKoNI?authuser=1#scrollTo=UoM9h9JUruSu\&uniqifier=2:~:text=Follow%20the%20instructions). Select Atlas UI as the procedure to deploy your first cluster.

4.1) Connect MongoDB with Python.
4.1.1) Create db.py in same directory of app.py.
4.1.2) Implement document saving in MongoDB.

**db.py**

```
import pymongo

MONGO_URI = ""  # MongoDB connection string
mongo_client = pymongo.MongoClient(MONGO_URI)

db = mongo_client.get_database("logging")  # database name in mongodb
collection = db.get_collection("test")  # collection name

# document {
#   query: string,
#   response: string,
#   embeddings: vector<number<1408>>
# }

def save_cache(document):
	collection.insert_one(document)
```
4.1.3) Save response from AI to database.
Modify app.py to save the AI response and its vector information in the database.
**app.py**

```
# ... other import dependencies
from db import save_cache

# ... other code

# search route
@server.get("/ask")
async def search(query):
	llm_response = getTextResponse(query)
	query_vector = getEmbedding(query)

	document = {
  	"response": llm_response,
  	"embeddings": query_vector,
  	"query": query
  	}
	save_cache(document)

	return {"message": "Your AI response is: {}".format(llm_response)}
```


4.2) Create index vector search in MongoDB AtlasMongoDB’s Vector Search enables AI-powered experiences to perform semantic search of unstructured data by its embeddings with machine learning models. We have to enable [**vector search index**](https://mongodb.prakticum-team.ru/docs/atlas/atlas-vector-search/create-index/#procedure/) in the database. You can go to the database in Atlas -> Atlas Search -> CREATE SEARCH INDEX.

Below is the JSON editor version of Atlas index.

```python
{
  "type": "vectorSearch", # vector search identity
  "fields": [
	{
  	# type of data in the asking the system again with a new question (but similar meaning): Howfield (vector information)
  	"type": "vector",
  	# field of document that store vector information
  	"path": "embeddings",
  	# dimension of vector, receiving from language model specification.
  	# for `text-embedding-3-small` of OpenAI, it's 1408 dimensions of vector.
  	"numDimensions": 1408,
  	# vector search find similarity of searching document with other. So, closest documents would get high score.
  	"similarity": "cosine"
	}
  ]
}
```


### 6) Obtain cache from the system in MongoDB Atlas

Logically, when we receive a new request from the client, we’ll embed the search query and perform a vector search to find the documents that contain embeddings that are semantically similar to the query embedding.
Vector search is one of the stages of aggregation pipelines. The pipeline is constructed as shown below.

```python
# Step 1: Perform vector search
{
  # $vectorSearch operation
    "$vectorSearch": {
        # Vector Search index in Atlas
        "index": "vector_index",
        # vector information in document
        "path": "embeddings",
        # embedding information of client's query
        "queryVector": numerical_embedding,
       # number of nearest neighbors to find the closest group. This value is used for 'searchScore' ranking
        "numCandidates": 20,
           # number of document to return ranked by semantic meaning / searchScore
        "limit": 5,
    },
},
# add 'score' field to show vector search score.
{
  "$addFields": {
  "score": {
    # vectorSearchScore is attached in $meta for pipeline after vectorSearch operation
    "$meta": "vectorSearchScore"
    }
  }
},
# return document that score greater than specific ratio
# in this case, only top one that has ration > 70% is going to return.
{
  "$match": {
    "score": { "$gte": 0.70 }
  }
},
# remove `embeddings` field. Because it reduce load to retrieve data.
{
  "$unset": ["embeddings"]
}
]
```
Modify db.py and app.py to implement the PyMongo aggregation pipeline for vector search.
**db.py**

```python
# ... other code
def perform_search_cache(query):
	pipeline = [
    	{
        	"$vectorSearch": {
            	"index": "vector_index",
            	"path": "embeddings",
            	"queryVector": query,
            	"numCandidates": 20,
            	"limit": 1,
        	},
    	},
    	{"$addFields": {"score": {"$meta": "vectorSearchScore"}}},
    	{"$match": {"score": {"$gte": 0.70}}},
    	{"$unset": ["embeddings"]},
	]
	result = collection.aggregate(pipeline)
	return result
```

**app.py**

```python
# ... other import dependencies
from db import save_cache, perform_search_cache

# ... other code
# search route
@server.get("/ask")
async def search(query):
	query_vector = getEmbedding(query)
	cache_response = list(perform_search_cache(query_vector))
	response = ""
	if len(cache_response) < 1:
    	# in case no cache hit
    	# perform search and save the cache
    	llm_response = getTextResponse(query)
    	document = {"response": llm_response, "embeddings": query_vector, "query": query}
    	save_cache(document)
    	response = llm_response
	else:
    	# if cache hit, return it
    	response = cache_response[0]["response"]
    	print("cache hit")
    	print(cache_response)
	return {"message": "Your AI response is: {}".format(response)}
```

We can try to send the request to our system. Let’s ask the system, “How are things with you?”

```
curl -X GET "http://127.0.0.1:8000/ask?query=how+are+things+with+you?"
```
response For the first time, the system retrieves data from the AI service.
```
{
  "message": "Your AI response is: I'm just a computer program, so I don't have feelings or emotions. But I'm here to help you with anything you need! How can I assist you today?"
}
```
Let’s try to ask the system a new question (but with a similar meaning): How are you today?
```
curl -X GET "http://127.0.0.1:8000/ask?query=how+are+you+today?"
```
Now, the system will return cache data from MongoDB Atlas.

```
{
  "message": "Your AI response is: I'm just a computer program, so I don't have feelings or emotions. But I'm here to help you with anything you need! How can I assist you today?"
}
```

If you go to shell/terminal, you will see a log like below.

```python
cache hit
[
  {
	'_id': ObjectId('6671440cb2bf0b0eb12b75b3'),
	'response': "I'm just a computer program, so I don't have feelings or emotions. But I'm here to help you with anything you need! How can I assist you today?",
	'query': 'how are thing with you?',
	'score': 0.8066450357437134
  }
]
```
It seems that the query “How are you today?” is **80%** similar to “How are things with you?” That is what we expect.

## Summary

This article outlines the implementation of a semantic caching system for LLM responses using MongoDB Atlas and Vector Search. The solution covered in this article aims to reduce costs and latency associated with frequent LLM API calls by caching responses based on query semantics rather than exact matches. 

The solution integrates FastAPI, OpenAI, and MongoDB Atlas to create a workflow where incoming queries are embedded into vectors and compared against cached entries. Matching queries retrieve stored responses, while new queries are processed by the LLM and then cached. 

Key benefits include reduced LLM service load, lower costs, faster response times for similar queries, and scalability. The system demonstrates how combining vector search capabilities with LLMs can optimize natural language processing applications, offering a balance between efficiency and response quality. 

Learn [how to implement semantic cache](https://mongodb.prakticum-team.ru/developer/products/atlas/advanced-rag-langchain-mongodb/) with the widely adopted LLM framework library LangChain.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb0d2d3f986bea52c/66d5c52fb11d19fc679c26d7/Screenshot_2024-09-02_at_15.00.21.png

Discover how to reduce API costs and improve response times for Large Language Models (LLMs) by implementing semantic caching using MongoDB Atlas and Vector Search. Learn to efficiently handle LLM queries by storing and retrieving embeddings—numerical vectors representing the semantic meaning of text—reducing the need for repeated API calls. This guide covers setting up a FastAPI server, integrating OpenAI, embedding LLM responses, and utilizing MongoDB Atlas's advanced vector search capabilities. Perfect for developers looking to optimize AI-driven applications, lower operational costs, and enhance scalability.

Kanin Kearpimy

Content by Kanin Kearpimy

Caching LLMs Response With MongoDB Atlas and Vector Search