chat_utils.py

from dataclasses import dataclass
from typing import Optional, List
from langchain.pydantic_v1 import BaseModel, Field

SYSTEM_PROMPT = (
    "You are an assistant  specialized in the legal and compliance field who must answer and converse with the user using the context provided. " +
    "When you answer the user, if it is relevant, cite the laws and articles you are referring to. NEVER mention the use of context in your answers. "
    "If you believe the question cannot be answered from the given context, do not make up an answer. Answer in the same language the user is speaking.\n\n ### Context:\n {context}"
)

SYSTEM_PROMPT_LOOP = (
    "You are an assistant who must inform the user that you do not have enough information to answer and ask if the user can provide you with additional information. " +
    "This answer, must be adapted to the conversation that occurred with the user that is provided to you. Just write down the answer "
)

@dataclass
class Answer():
    answer: str
    new_documents: Optional[List] = None
    status: Optional[int] = 1

class ContextInput(BaseModel):
    text: str = Field(
        title="Text",
        description="Self-explanatory summary describing what the user is asking for"
        )

def get_instance_dynamic_class(lib_path:str, class_name:str, **kwargs):
    """
    Instantiate a dynamically imported class from a given library path and class name.

    Args:
        lib_path (str): The path to the library/module containing the class.
        class_name (str): The name of the class to instantiate.
        **kwargs: Additional keyword arguments to pass to the class constructor.

    Returns:
        An instance of the dynamically imported class initialized with the provided arguments.
    """

    mod = __import__(lib_path, fromlist=[class_name])
    dynamic_class = getattr(mod, class_name)
    return dynamic_class(**kwargs)


def get_init_modules(config):
    embedder = get_instance_dynamic_class(
        lib_path='langchain_community.embeddings',
        class_name=config["embeddings"]["class"],
        **config["embeddings"]["kwargs"]
    )

    llm = get_instance_dynamic_class(
        lib_path='langchain_community.chat_models',
        class_name=config["llm"]["class"],
        **config["llm"]["kwargs"]
    )

    mod_chat = __import__("langchain_community.chat_message_histories",
                          fromlist=[config["chatDB"]["class"]])
    chatDB_class = getattr(mod_chat, config["chatDB"]["class"])
    retriever = get_vectorDB_module(config['vectorDB'], embedder)

    return embedder, llm, chatDB_class, retriever


def get_vectorDB_module(db_config, embedder):
    mod_chat = __import__("langchain_community.vectorstores",
                          fromlist=[db_config["class"]])
    vectorDB_class = getattr(mod_chat, db_config["class"])

    if db_config["class"] == 'Qdrant':
        from qdrant_client import QdrantClient
        import inspect

        # Get QdrantClient init parameters name from signature
        signature_params = inspect.signature(QdrantClient.__init__).parameters.values()
        params_to_exclude = ['self', 'kwargs']
        client_args = [el.name for el in list(signature_params) if el.name not in params_to_exclude]

        client_kwargs = {k: v for k,
                         v in db_config['kwargs'].items() if k in client_args}
        db_kwargs = {
            k: v for k, v in db_config['kwargs'].items() if k not in client_kwargs}

        client = QdrantClient(**client_kwargs)

        retriever = vectorDB_class(
            client, embeddings=embedder, **db_kwargs).as_retriever(
                search_type=db_config["retriever_args"]["search_type"],
                search_kwargs=db_config["retriever_args"]["search_kwargs"]
        )

    else:
        retriever = vectorDB_class(embeddings=embedder, **db_config["kwargs"]).as_retriever(
            search_type=db_config["retriever_args"]["search_type"],
            search_kwargs=db_config["retriever_args"]["search_kwargs"]
        )

    return retriever