Conversation Class
class ConversationManager:
_instance = None
@staticmethod
def get_instance():
if ConversationManager._instance is None:
ConversationManager._instance = ConversationManager()
return ConversationManager._instance
def __init__(self):
if ConversationManager._instance is not None:
raise Exception("This class is a singleton!")
else:
self.brand = None
def set_conversation(self, brand, product, table_data, models_data, service_list):
self.brand = brand
self.product = product
self.table_data = table_data
self.models_data = models_data
self.service_list = service_list
def get_conversation(self):
if self.brand is None:
raise Exception("Conversation object has not been initialized.")
return self.brand, self.product, self.table_data, self.models_data, self.service_list
RETRIEVAL CHAIN DECOMPRISATION
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import PromptTemplate
import json
from twilio.rest import Client
import requests
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages.human import HumanMessage
from langchain_core.messages.ai import AIMessage
from dotenv import load_dotenv
def transform_messages(messages):
print(messages)
transformed_messages = []
for i in range(len(messages)):
if type(messages[i]) == HumanMessage:
transformed_message = {
'message': messages[i].content,
'role':'user'
}
transformed_messages.append(transformed_message)
else:
transformed_message = {
'message': messages[i].content,
'role':'assistant'
}
transformed_messages.append(transformed_message)
return transformed_messages
convo_dict = conversation.__dict__
y = convo_dict['memory'].__dict__
x = y['chat_memory'].__dict__
print(x)
data_dict_convo = {
"memory": transform_messages(x['messages']),
}
def retrieve_history_from_json(message_list):
convo_hist = []
for i in range(len(message_list)):
if message_list[i]['role'] == 'user':
message = HumanMessage(message_list[i]['message'])
else:
message = AIMessage(message_list[i]['message'])
convo_hist.append(message)
chat_history = InMemoryChatMessageHistory(messages=convo_hist)
return ConversationBufferMemory(memory_key='chat_history', return_messages=True, output_key='answer', chat_memory=chat_history)