File size: 5,838 Bytes
f51bb92 9a7da99 b409192 f51bb92 fc2cb23 f51bb92 fc2cb23 f51bb92 e029e22 aaaac46 1e2550f f51bb92 e029e22 f51bb92 fc2cb23 f51bb92 e029e22 fc2cb23 e029e22 fc2cb23 f51bb92 e029e22 fc2cb23 e029e22 1e2550f f51bb92 e029e22 aaaac46 fc2cb23 f51bb92 fc2cb23 e029e22 fc2cb23 b409192 fc2cb23 a2ac5f7 f51bb92 e029e22 e165ea5 f51bb92 fc2cb23 e165ea5 b409192 f51bb92 b409192 e029e22 aaaac46 f51bb92 fc2cb23 f51bb92 b409192 fc2cb23 8f6647c e029e22 8f6647c f51bb92 e029e22 b409192 e029e22 fc2cb23 e029e22 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
from modules.chat.helpers import get_prompt
from modules.chat.chat_model_loader import ChatModelLoader
from modules.vectorstore.store_manager import VectorStoreManager
from modules.retriever.retriever import Retriever
from modules.chat.langchain.langchain_rag import (
Langchain_RAG_V2,
QuestionGenerator,
)
class LLMTutor:
def __init__(self, config, user, logger=None):
"""
Initialize the LLMTutor class.
Args:
config (dict): Configuration dictionary.
user (str): User identifier.
logger (Logger, optional): Logger instance. Defaults to None.
"""
self.config = config
self.llm = self.load_llm()
self.user = user
self.logger = logger
self.vector_db = VectorStoreManager(config, logger=self.logger).load_database()
self.qa_prompt = get_prompt(config, "qa") # Initialize qa_prompt
self.rephrase_prompt = get_prompt(
config, "rephrase"
) # Initialize rephrase_prompt
# TODO: Removed this functionality for now, don't know if we need it
# if self.config["vectorstore"]["embedd_files"]:
# self.vector_db.create_database()
# self.vector_db.save_database()
def update_llm(self, old_config, new_config):
"""
Update the LLM and VectorStoreManager based on new configuration.
Args:
new_config (dict): New configuration dictionary.
"""
changes = self.get_config_changes(old_config, new_config)
if "llm_params.llm_loader" in changes:
self.llm = self.load_llm() # Reinitialize LLM if chat_model changes
if "vectorstore.db_option" in changes:
self.vector_db = VectorStoreManager(
self.config, logger=self.logger
).load_database() # Reinitialize VectorStoreManager if vectorstore changes
# TODO: Removed this functionality for now, don't know if we need it
# if self.config["vectorstore"]["embedd_files"]:
# self.vector_db.create_database()
# self.vector_db.save_database()
if "llm_params.llm_style" in changes:
self.qa_prompt = get_prompt(
self.config, "qa"
) # Update qa_prompt if ELI5 changes
def get_config_changes(self, old_config, new_config):
"""
Get the changes between the old and new configuration.
Args:
old_config (dict): Old configuration dictionary.
new_config (dict): New configuration dictionary.
Returns:
dict: Dictionary containing the changes.
"""
changes = {}
def compare_dicts(old, new, parent_key=""):
for key in new:
full_key = f"{parent_key}.{key}" if parent_key else key
if isinstance(new[key], dict) and isinstance(old.get(key), dict):
compare_dicts(old.get(key, {}), new[key], full_key)
elif old.get(key) != new[key]:
changes[full_key] = (old.get(key), new[key])
# Include keys that are in old but not in new
for key in old:
if key not in new:
full_key = f"{parent_key}.{key}" if parent_key else key
changes[full_key] = (old[key], None)
compare_dicts(old_config, new_config)
return changes
def retrieval_qa_chain(
self, llm, qa_prompt, rephrase_prompt, db, memory=None, callbacks=None
):
"""
Create a Retrieval QA Chain.
Args:
llm (LLM): The language model instance.
qa_prompt (str): The QA prompt string.
rephrase_prompt (str): The rephrase prompt string.
db (VectorStore): The vector store instance.
memory (Memory, optional): Memory instance. Defaults to None.
Returns:
Chain: The retrieval QA chain instance.
"""
retriever = Retriever(self.config)._return_retriever(db)
if self.config["llm_params"]["llm_arch"] == "langchain":
self.qa_chain = Langchain_RAG_V2(
llm=llm,
memory=memory,
retriever=retriever,
qa_prompt=qa_prompt,
rephrase_prompt=rephrase_prompt,
config=self.config,
callbacks=callbacks,
)
self.question_generator = QuestionGenerator()
else:
raise ValueError(
f"Invalid LLM Architecture: {self.config['llm_params']['llm_arch']}"
)
return self.qa_chain
def load_llm(self):
"""
Load the language model.
Returns:
LLM: The loaded language model instance.
"""
chat_model_loader = ChatModelLoader(self.config)
llm = chat_model_loader.load_chat_model()
return llm
def qa_bot(self, memory=None, callbacks=None):
"""
Create a QA bot instance.
Args:
memory (Memory, optional): Memory instance. Defaults to None.
qa_prompt (str, optional): QA prompt string. Defaults to None.
rephrase_prompt (str, optional): Rephrase prompt string. Defaults to None.
Returns:
Chain: The QA bot chain instance.
"""
# sanity check to see if there are any documents in the database
if len(self.vector_db) == 0:
raise ValueError(
"No documents in the database. Populate the database first."
)
qa = self.retrieval_qa_chain(
self.llm,
self.qa_prompt,
self.rephrase_prompt,
self.vector_db,
memory,
callbacks=callbacks,
)
return qa
|