Notes

import json
import os
import json
import numpy as np
import redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from langchain_openai import AzureOpenAIEmbeddings

class dataVectorization():

def __init__(self):
self.config_details = self.load_config()
self.embedding_model = self.config_details['EMBEDDING_MODEL']

def load_config(self):
'''
Function to load the config file(json) from a local path

return:
- config_details (dict): dictionary of configuration details

'''
with open(r"C:UsersBN117KHOneDrive - EYDesktopWorkGenAITokio Marine2 NotebooksRAGconfig.json",'r') as f:
config_details = json.load(f)

os.environ['AZURE_OPENAI_API_KEY'] = config_details["OPENAI_API_KEY"]
os.environ['AZURE_OPENAI_ENDPOINT'] = config_details["azure_endpoint_gpt3"]

return config_details

def getRedisConnection(self,strHost, port=6379, ssl=False):
'''
Function to establish connection with Redis

Parameters:
- strHost (str): redis host
- port (int): redis port

return:
- az_redis (redis.client.Redis): Redis client
each page

'''
try:
az_redis = redis.Redis(host=strHost, port=port, ssl=ssl, health_check_interval=30)
return az_redis
except Exception as e:
print(f'Error getRedisConnection(): {e}')
return None

# Store data in Redis
def generateEmbeddings(self,conn,listPageData, indexFilename):
'''
Function to generate embeddings

Parameters:
- conn : Redis client
- listPageData (list): page level data from document
- indexFilename (str): index name for the document

'''
connectionPipeline = conn.pipeline(transaction=False)

for i in range(len(listPageData)):
embeddings = AzureOpenAIEmbeddings(deployment=self.embedding_model,chunk_size=1,
openai_api_version = self.config_details["OPENAI_API_VERSION"])

vector = embeddings.embed_query(listPageData[i]["pageText"])
vector = np.array(vector).astype(np.float32).tobytes()


# Create a new hash with Answer and embedding
dictPostHash = {
"pageText": listPageData[i]["pageText"],
"embedding": vector,
"pageNnumber":listPageData[i]["pageNumber"]
}
# create hash
conn.hset(name=f"{indexFilename}:{i}", mapping=dictPostHash)

connectionPipeline.execute()

def createIndex(self,conn,indexFilename):
'''
Function to create index for the document in Redis

Parameters:
- conn : Redis client
- indexFilename (str): index name for the document

'''
SCHEMA = [
TextField("pageText"),
VectorField("embedding", "HNSW", {"TYPE": "FLOAT32", "DIM": 1536, "DISTANCE_METRIC": "COSINE"}),
TextField("pageNumber")
]

try:
conn.ft(f"{indexFilename}").info()
print("Index already exists")
except:
conn.ft(f"{indexFilename}").create_index(fields=SCHEMA, definition=IndexDefinition(prefix=[f"{indexFilename}:"], index_type=IndexType.HASH))