Merge branch 'main' of https://github.com/paisleypark3121/ibn

.gitignore

@@ -0,0 +1,11 @@
+.env
+.*
+.cer
+data
+__pycache__
+
+# Except .gitignore itself
+!.gitignore
+
+# Except .gitkeep files
+!.gitkeep

app.py

@@ -0,0 +1,530 @@
+from dotenv import load_dotenv
+import os
+import json
+import inspect
+
+from langchain_openai import ChatOpenAI
+
+#from langchain.schema import AIMessage, HumanMessage, SystemMessage
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
+
+from utilities.qdrant.QdrantLangchainManager import *
+from langchain_core.tools import tool
+import gradio as gr
+
+load_dotenv()
+
+credentials = {}
+credentials[os.getenv("USERNAME")] = os.getenv("PASSWORD")
+i = 1
+while os.getenv(f"USERNAME_{i}") and os.getenv(f"PASSWORD_{i}"):
+    credentials[os.getenv(f"USERNAME_{i}")] = os.getenv(f"PASSWORD_{i}")
+    i += 1
+
+#print(credentials)
+
+system_message = """
+You are a specialized assistant that helps system managers configure 5G networks. 
+
+🔹 **Your Main Role**:
+- Assisting in **5G SA modem configuration**.
+- Setting up modems for **Mixed 5G/4G Networks**.
+- Deploying **industrial 5G SA networks** with custom requirements.
+- Providing expert knowledge on **5G technology, network components, frequency bands, and optimization**.
+
+📌 **Context Handling**:
+- If relevant [context] is available, use it to provide **precise and accurate answers**.
+- If [context] is **not provided or does not apply**, rely on your **general knowledge** of 5G.
+- Always **distinguish between official reference information and general knowledge**, prioritizing accuracy.
+
+🔹 **How to Respond**:
+- **Prioritize clarity**: Keep answers direct and solution-oriented.
+- **Guide the user step by step**, ensuring only essential parameters are asked first.
+- **Acknowledge the source**: If using [context], clarify that the information comes from official reference material.
+
+[context]
+{context}
+"""
+
+MAX_HISTORY = 20
+messages = [SystemMessage(content=system_message)]
+
+def _truncate_messages(messages, MAX_HISTORY=20):
+    """Keeps the first system message and only retains the last MAX_HISTORY user/AI messages."""
+    
+    # ✅ Ensure system_message is always the first message
+    system_msg = messages[0]  
+    truncated_messages = messages[1:][-MAX_HISTORY:]  # Keep only the last MAX_HISTORY messages (excluding system message)
+    
+    # ✅ Rebuild messages list with system_message at the beginning
+    messages = [system_msg] + truncated_messages
+
+    return messages
+
+
+tool_system_message = """You are an assistant that generates a Bash script based on the given [parameters] and [context].
+If sufficient context is not available, return an ERROR message.
+
+Below is the [function_description] that defines the tool you are using. Use it to correctly interpret the parameters.
+
+[function_description]
+{function_description}
+
+[parameters]
+{parameters}
+
+[context]
+{context}
+"""
+
+collection_name = "5g reference"
+
+manager = QdrantLangchainManager(
+    qdrant_url=os.getenv("QDRANT_URL"),
+    qdrant_api_key=os.getenv("QDRANT_API_KEY"),
+    llm=ChatOpenAI(model="gpt-4o-mini", streaming=True),
+    crossencoder="cross-encoder/ms-marco-MiniLM-L-6-v2"
+)
+
+def _get_tool_query(tool_name: str) -> str:
+    """Generates a query to prompt the LLM for a tool-based response."""    
+        
+    if tool_name == "configure_5g_sa_modem":
+        return "I need a Bash script to configure a modem to connect to a Standalone (SA) 5G network with a specific APN."
+
+    elif tool_name == "configure_mixed_5g_4g_modem":
+        return "I need a Bash script to configure a modem for a mixed 5G/4G network."
+
+    elif tool_name == "configure_industrial_5g_sa_modem":
+        return "I need a Bash script to configure a modem for an industrial 5G SA network."
+
+    return None
+
+def _tool_helper(tool_name: str, tool_args: dict) -> str:
+    """Helper function to generate a Bash script using the LLM based on tool parameters and retrieved context."""
+
+    manager.get_collection(collection_name)
+    query = _get_tool_query(tool_name)
+    if not query:
+        return "ERROR: Invalid tool name."
+
+    context = manager.search_qdrant(query)  
+    if not context:
+        return "ERROR: No relevant reference found in the vector store."
+
+    # Get the function object from globals() based on the tool name
+    function_obj = globals().get(tool_name)
+    # Extract the function description directly
+    function_description = getattr(function_obj, "description", "No documentation available.")
+
+    # Format parameters into a readable string
+    parameters_str = "\n".join([f"- {key}: {value}" for key, value in tool_args.items()])
+
+    # **Include function documentation in the system message**
+    system_message = tool_system_message.format(
+        parameters=parameters_str, 
+        context=context, 
+        function_description=function_description
+    )
+
+    # print("*****")
+    # print(system_message)
+    # print("*****")
+
+    messages = [
+        SystemMessage(content=system_message),
+        #HumanMessage(content=query)
+    ]
+    
+    response = manager.llm.invoke(messages)
+    return response.content
+
+@tool
+def configure_5g_sa_modem(
+    modem_device: str,  
+    apn: str,  
+    pdp_type: str,  
+    bearer: str,  
+    bands: list,  
+    ip_config: str,  
+    dns: str,  
+    enable_roaming: bool  
+) -> str:
+    """Configures a modem to connect to a 5G SA (Standalone) network.
+
+    Parameters:
+    - modem_device (str): Serial port of the modem (e.g., '/dev/ttyUSB0').
+    - apn (str): APN name to configure.
+    - pdp_type (str): Data connection type (IP, IPV6, IPV4V6).
+    - bearer (str): Physical connection type (NR5G, LTE, AUTO).
+    - bands (list): List of 5G bands to enable (e.g., ['n1', 'n3', 'n78']).
+    - ip_config (str): IP configuration (Static, Dynamic).
+    - dns (str): DNS address to use.
+    - enable_roaming (bool): Enables or disables roaming.
+
+    Returns:
+    - str: Confirmation message with the applied configuration details.
+    """
+
+    function_name = inspect.currentframe().f_code.co_name
+    return _tool_helper(tool_name=function_name, tool_args=locals())
+
+@tool
+def configure_mixed_5g_4g_modem(
+    modem_device: str,  
+    lte_bands: list,  
+    nr5g_bands: list,  
+    network_priority: str,  
+    connection_timeout: int  
+) -> str:
+    """Configures a modem for a mixed 5G/4G network.
+
+    Parameters:
+    - modem_device (str): Serial port of the modem (e.g., '/dev/ttyUSB0').
+    - lte_bands (list): List of LTE bands to activate (e.g., ['1', '3', '7']).
+    - nr5g_bands (list): List of 5G NR bands to activate (e.g., ['n78', 'n79']).
+    - network_priority (str): Preference between 5G, 4G, or automatic mode ('5G', '4G', 'AUTO').
+    - connection_timeout (int): Maximum connection timeout in seconds.
+
+    Returns:
+    - str: Confirmation message with the applied configuration details.
+    """
+
+    function_name = inspect.currentframe().f_code.co_name
+    return _tool_helper(tool_name=function_name, tool_args=locals())
+
+@tool
+def configure_industrial_5g_sa_modem(
+    modem_device: str,  
+    industrial_apn: str,  
+    lte_bands: list,  
+    nr5g_bands: list,  
+    disable_wcdma: bool,  
+    verify_registration: bool  
+) -> str:
+    """Configures a modem to connect to an industrial 5G SA network.
+
+    Parameters:
+    - modem_device (str): Serial port of the modem (e.g., '/dev/ttyUSB2').
+    - industrial_apn (str): APN name specific to the industrial network.
+    - lte_bands (list): List of LTE bands to activate.
+    - nr5g_bands (list): List of 5G NR bands to activate.
+    - disable_wcdma (bool): If True, disables WCDMA bands.
+    - verify_registration (bool): If True, verifies registration status.
+
+    Returns:
+    - str: Confirmation message with the applied configuration details.
+    """
+
+    function_name = inspect.currentframe().f_code.co_name
+    return _tool_helper(tool_name=function_name, tool_args=locals())
+
+tools = [
+    configure_5g_sa_modem, 
+    configure_mixed_5g_4g_modem, 
+    configure_industrial_5g_sa_modem
+]
+
+llm = ChatOpenAI(
+    model="gpt-4o-mini", 
+    streaming=True,
+)
+llm_with_tools = llm.bind_tools(tools)
+
+tool_name = ""
+tool_args = ""
+
+tool_mapping = {
+    "configure_5g_sa_modem": configure_5g_sa_modem,
+    "configure_mixed_5g_4g_modem": configure_mixed_5g_4g_modem,
+    "configure_industrial_5g_sa_modem": configure_industrial_5g_sa_modem
+}
+
+label_buttons = ["Confirm", "Cancel"]
+
+def _check_response(ai_msg):
+
+    # if "persona fisica" in ai_msg.lower() and "società" in ai_msg.lower():
+
+    #     context_analysis_prompt = f"""
+    #     Questa è la risposta dell'LLM a un utente che vuole creare un cliente:
+
+    #     \"{ai_msg}\"
+
+    #     Devi solo rispondere con "create_client" se la risposta chiede all'utente di selezionare tra "Persona Fisica" e "Società", 
+    #     oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale dei tipi di clienti senza richiedere un'azione.
+    #     """
+    #     analysis_response = llm.invoke(context_analysis_prompt).content.strip()
+    #     print(analysis_response)
+
+    #     if analysis_response == "create_client":
+    #         return True, analysis_response
+
+    # elif "partita iva" in ai_msg.lower() and "codice fiscale" in ai_msg.lower():
+
+    #     context_analysis_prompt = f"""
+    #     Questa è la risposta dell'LLM a un utente che deve scegliere la modalità con cui vuole inserire i dati societari
+
+    #     \"{ai_msg}\"
+
+    #     Devi solo rispondere con "select_mode" se la risposta chiede all'utente di selezionare tra "Partita IVA" e "Codice Fiscale", 
+    #     oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale senza richiedere un'azione.
+    #     """
+    #     analysis_response = llm.invoke(context_analysis_prompt).content.strip()
+    #     print(analysis_response)
+
+    #     if analysis_response == "select_mode":
+    #         return True, analysis_response
+
+    return False, ""
+
+def _get_confirmation_message(tool_name: str, tool_args: dict) -> str:
+    """Generate a descriptive message to confirm the action that is about to be performed."""
+    
+    if tool_name == "configure_5g_sa_modem":
+        return f"**ACTION** You are about to configure 5G SA modem." # {tool_args.get('first_name', 'N/D')} {tool_args.get('last_name', 'N/D')}."
+
+    elif tool_name == "configure_mixed_5g_4g_modem":
+        return f"**ACTION** You are about to configure a mixed 5G/4G modem."
+
+    elif tool_name == "configure_industrial_5g_sa_modem":
+        return f"**ACTION** You are about to configure an industrial 5G SA modem."
+
+    return "**ACTION** Unknown action."
+
+def _format_tool_args(tool_name: str, tool_args: dict) -> str:
+    """Transforms technical parameters into a user-friendly description, handling optional values."""
+
+    translations = {
+        "modem_device": "Modem Device",
+        "apn": "APN",
+        "pdp_type": "PDP Type",
+        "bearer": "Bearer",
+        "bands": "5G Bands",
+        "ip_config": "IP Configuration",
+        "dns": "DNS",
+        "enable_roaming": "Roaming Enabled",
+        "lte_bands": "LTE Bands",
+        "nr5g_bands": "5G NR Bands",
+        "network_priority": "Network Priority",
+        "connection_timeout": "Connection Timeout",
+        "industrial_apn": "Industrial APN",
+        "disable_wcdma": "Disable WCDMA",
+        "verify_registration": "Verify Registration",
+    }
+
+    _formatted_args = "\n".join([
+        f"- {translations.get(key, key)}: {', '.join(value) if isinstance(value, list) else value if value else 'Not specified'}"
+        for key, value in tool_args.items()
+    ])
+
+    return _formatted_args
+
+def chatbot_response(message, history):
+    
+    global messages, tool_name, tool_args
+
+    history = history or []
+
+    if message.startswith("**OPERATION CANCELLED**"):
+        history.append({"role": "user", "content": "**OPERATION CANCELLED**"})
+    else:
+        history.append({"role": "user", "content": message})
+    
+    #messages = [SystemMessage(content=system_message)]
+    #truncated_history = history[-MAX_HISTORY:] if len(history) > MAX_HISTORY else history
+
+    #print(f"Truncated history: {truncated_history}")
+
+    # for entry in truncated_history:
+    #     if entry["role"] == "user":
+    #         messages.append(HumanMessage(content=entry["content"]))
+    #     elif entry["role"] == "assistant":
+    #         messages.append(AIMessage(content=entry["content"]))
+
+    if message is not None:
+        messages.append(HumanMessage(content=message))
+        messages=_truncate_messages(messages,MAX_HISTORY)       
+        
+        #_response=llm_with_tools.stream(history_langchain_format)
+        ai_msg=llm_with_tools.invoke(messages)
+        #print(ai_msg)
+
+        choice, analysis_response = _check_response(ai_msg.content)
+
+        # if choice and analysis_response == "create_client":
+        #     history.append({"role": "assistant", "content": f"**CREAZIONE CLIENTE**\n{ai_msg.content}"})
+
+        # elif choice and analysis_response == "select_mode":
+        #     history.append({"role": "assistant", "content": f"**MODALITA' INSERIMENTO DATI SOCIETARI**\n{ai_msg.content}"})
+
+        # elif hasattr(ai_msg, "tool_calls") and ai_msg.tool_calls:
+        if hasattr(ai_msg, "tool_calls") and ai_msg.tool_calls:
+    
+            for tool_call in ai_msg.tool_calls:
+                tool_name = tool_call["name"].lower()
+                tool_args = tool_call["args"]
+                selected_tool = tool_mapping.get(tool_name)
+
+                if selected_tool:
+                    # if selected_tool==create_company_vat_number:
+                    #     tool_args=_get_company_info(tool_args.get('vat_number', 'N/D'))
+                    #     if tool_args is None:
+                    #         history.append({"role": "user", "content": "Errore: Dati societari non trovati."})
+                    #         return history
+
+                    confirmation_message = _get_confirmation_message(tool_name, tool_args)
+                    
+                    formatted_args = _format_tool_args(tool_name, tool_args)
+                    ai_msg= f"{confirmation_message}\n\n**Here are the details entered:**\n{formatted_args}"
+                    history.append({"role": "assistant", "content": ai_msg})   
+                    messages.append(AIMessage(content=ai_msg))
+
+                    print("---")
+                    print("Tool name: ", tool_name)                         
+                    print("Tool args: ", tool_args)
+                    print("---")
+
+        else:
+            history.append({"role": "assistant", "content": ai_msg.content})
+            messages.append(AIMessage(content=ai_msg.content))
+
+        return history
+
+def reset_textbox():
+    """Clears the textbox after sending a message."""
+    return gr.update(value="")
+
+def hide_buttons():
+    """Hides buttons and shows the textbox after clicking a button."""
+    return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)
+
+def show_or_hide_buttons(history):
+    global label_buttons
+
+    if history and history[-1]["content"].startswith("**ACTION**"):
+        label_buttons = ["Confirm", "Cancel"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1]),
+            gr.update(visible=False)
+        )
+    return (
+        gr.update(visible=True), 
+        gr.update(visible=False), 
+        gr.update(visible=False),
+        gr.update(visible=True)
+    )
+
+def button_clicked(option, history):
+    """Handles button clicks and updates chat history."""
+
+    global messages, tool_name, tool_args
+
+    print(f"Button clicked: {option}")
+
+    if (option == "Confirm"):
+        history.append({"role": "user", "content": option})
+        print("***")
+        print(f"Tool name: {tool_name}")
+        print(f"Tool args: {tool_args}")
+        print("***")
+        if tool_name and tool_args:
+            selected_tool = tool_mapping.get(tool_name)
+            #print(f"Selected tool: {selected_tool}")
+            if selected_tool:
+                tool_output = selected_tool.invoke(tool_args)
+                history.append({"role": "assistant", "content": tool_output})                
+            else:
+                history.append({"role": "user", "content": "Operazione annullata"})
+    elif option == "Cancel":
+        history.append({"role": "user", "content": option})
+        llm_message = (
+            "**OPERATION CANCELLED**\n"
+        )
+        history = chatbot_response(llm_message, history)
+    
+    messages = [messages[0]]
+    tool_name = ""
+    tool_args = ""
+    
+    return history
+
+def authenticate(username, password):
+    if username in credentials and credentials[username] == password:
+        print("🔑 Login successful!")
+        return gr.update(visible=False), gr.update(visible=True), gr.update(value="", visible=False)  # Hide login, show chatbot, clear error
+    else:
+        print("❌ Incorrect username or password")
+        return gr.update(visible=True), gr.update(visible=False), gr.update(value="❌ Incorrect username or password", visible=True)  # Show error
+
+def disable_inputs():
+    return gr.update(interactive=False), gr.update(interactive=False), gr.update(interactive=False), gr.update(interactive=False)
+
+def enable_inputs():
+    return gr.update(interactive=True), gr.update(interactive=True), gr.update(interactive=True), gr.update(interactive=True)
+
+with gr.Blocks() as demo:
+
+    # 🔒 Login Section (Initially Visible)
+    with gr.Column(visible=True) as login_section:
+        gr.Markdown("### 🔒 Login Required")
+        username_input = gr.Textbox(label="Username")
+        password_input = gr.Textbox(label="Password", type="password")
+        login_button = gr.Button("Login")
+        error_message = gr.Text("", visible=False)
+
+    # 🧠 Chatbot Section (Initially Hidden)
+    with gr.Column(visible=False) as chatbot_section:
+
+        chatbot = gr.Chatbot(
+            label="System Manager Chatbot",
+            type="messages"
+        )  
+
+        user_input = gr.Textbox(label="User",placeholder="What would you like to ask your assistant?")
+
+        with gr.Row():
+            btn1 = gr.Button("", visible=False)
+            btn2 = gr.Button("", visible=False)
+
+        send_btn = gr.Button("Send")
+
+        # When user submits text
+        user_input.submit(disable_inputs, None, [user_input, send_btn, btn1, btn2]) \
+            .then(chatbot_response, [user_input, chatbot], chatbot) \
+            .then(reset_textbox, None, user_input) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn]) \
+            .then(enable_inputs, None, [user_input, send_btn, btn1, btn2])
+
+        # When user clicks send
+        send_btn.click(disable_inputs, None, [user_input, send_btn, btn1, btn2]) \
+            .then(chatbot_response, [user_input, chatbot], chatbot) \
+            .then(reset_textbox, None, user_input) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn]) \
+            .then(enable_inputs, None, [user_input, send_btn, btn1, btn2])
+
+        # Button clicks: Show textbox, hide buttons
+        btn1.click(disable_inputs, None, [user_input, send_btn, btn1, btn2]) \
+            .then(lambda h: button_clicked(label_buttons[0], h), chatbot, chatbot) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn]) \
+            .then(enable_inputs, None, [user_input, send_btn, btn1, btn2])
+        
+        btn2.click(disable_inputs, None, [user_input, send_btn, btn1, btn2]) \
+            .then(lambda h: button_clicked(label_buttons[1], h), chatbot, chatbot) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn]) \
+            .then(enable_inputs, None, [user_input, send_btn, btn1, btn2])
+
+    # 🔑 Login Button Action
+    login_button.click(
+        authenticate, 
+        [username_input, password_input], 
+        [login_section, chatbot_section, error_message]
+    )
+
+
+demo.launch(
+    debug=True, 
+    #share=True
+)

old_app.py

@@ -0,0 +1,609 @@
+from dotenv import load_dotenv
+import os
+import json
+
+from langchain_openai import ChatOpenAI
+
+#from langchain.schema import AIMessage, HumanMessage, SystemMessage
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
+
+from langchain_core.tools import tool
+import gradio as gr
+
+load_dotenv()
+
+credentials = {}
+credentials[os.getenv("USERNAME")] = os.getenv("PASSWORD")
+i = 1
+while os.getenv(f"USERNAME_{i}") and os.getenv(f"PASSWORD_{i}"):
+    credentials[os.getenv(f"USERNAME_{i}")] = os.getenv(f"PASSWORD_{i}")
+    i += 1
+
+#print(credentials)
+
+MAX_HISTORY = 20
+
+system_message = """
+Sei l'assistente virtuale di Forfè, un software di fatturazione e gestione fiscale 
+specifico per il Regime Forfettario, la Gestione Separata INPS e la Cassa Artigiani e Commercianti. 
+Il tuo compito è aiutare gli utenti nella gestione della loro attività, fornendo supporto nella creazione 
+di fatture, gestione clienti e prodotti, e altre operazioni contabili.
+
+🔹 **Le tue funzionalità principali includono:**
+- Creazione di nuovi clienti, sia persone fisiche che società.
+- Creazione di nuovi prodotti o servizi da fatturare.
+- Generazione di fatture con i dati dei clienti e dei prodotti registrati.
+
+📌 **Regole di interazione:**
+- Chiedi all'utente solo i dati strettamente necessari per l'operazione richiesta.
+- Mantieni sempre un linguaggio chiaro, professionale e amichevole.
+- Se il numero di informazioni da richiedere all'utente sono troppe, dividi l'operazione in più passaggi, chiedendo dapprima i parametri obbligatori e poi quelli opzionali.
+- Rispondi in italiano e guida l'utente passo dopo passo nel processo.
+
+🚀 **Obiettivo:** Aiutare i professionisti e le piccole imprese a gestire la loro attività in modo semplice ed efficace con Forfè.
+"""
+
+@tool
+def create_product(
+    name: str,  
+    price: float  
+) -> str:
+    """Crea un nuovo prodotto o servizio e restituisce una conferma.
+
+    Parametri:
+    - name (str): Nome del prodotto o del servizio offerto.
+    - price (float): Importo in euro (€) del prodotto o servizio.
+
+    Ritorna:
+    - str: Un messaggio di conferma con i dettagli del prodotto o servizio creato.
+    """
+
+    print("\n--- Nuovo Prodotto/Servizio Creato ---")
+    print(f"Nome: {name}")
+    print(f"Prezzo: {price:.2f} EUR")
+
+    return f"Il nuovo prodotto o servizio '{name}' con un importo di {price:.2f}€ è stato generato con successo!"
+
+@tool
+def create_customer_type(customer_type: str) -> str:
+    """Seleziona il tipo di cliente da creare.
+
+    Parametri:
+    - customer_type (str): Il tipo di cliente da creare. Deve essere 'individual' (persona fisica) o 'company' (società).
+
+    Ritorna:
+    - str: Un messaggio di conferma con il tipo di cliente selezionato.
+    """
+
+    if customer_type not in ["individual", "company"]:
+        return "Errore: Il tipo di cliente deve essere 'individual' (persona fisica) o 'company' (società)."
+
+    return f"Hai selezionato {customer_type}. Ora puoi procedere con l'inserimento dei dati."
+
+@tool
+def create_individual(
+    first_name: str, 
+    last_name: str, 
+    tax_code: str, 
+    address: str, 
+    street_number: str, 
+    postal_code: str, 
+    city: str, 
+    province: str, 
+    country: str, 
+    vat_number: str = None,  # Opzionale
+    email: str = None, 
+    pec: str = None, 
+    phone: str = None, 
+    recipient_code: str = None
+) -> str:
+    """Crea un cliente di tipo persona fisica.
+
+    Parametri:
+    - first_name (str): Nome del cliente.
+    - last_name (str): Cognome del cliente.
+    - tax_code (str): Codice fiscale.
+    - address (str): Indirizzo di residenza.
+    - street_number (str): Numero civico.
+    - postal_code (str): CAP.
+    - city (str): Città.
+    - province (str): Provincia.
+    - country (str): Nazione.
+    - vat_number (str, opzionale): Partita IVA, se presente.
+    - email (str, opzionale): Indirizzo email.
+    - pec (str, opzionale): Indirizzo PEC.
+    - phone (str, opzionale): Numero di telefono.
+    - recipient_code (str, opzionale): Codice destinatario.
+
+    Ritorna:
+    - str: Un messaggio di conferma con i dati della persona fisica creata.
+    """
+
+    print("\n--- Nuova Persona Fisica Creata ---")
+    print(f"Nome: {first_name} {last_name}")
+    print(f"Codice Fiscale: {tax_code}")
+    print(f"Indirizzo: {address}, {street_number}, {postal_code}, {city}, {province}, {country}")
+
+    if vat_number:
+        print(f"Partita IVA: {vat_number}")
+    if email:
+        print(f"Email: {email}")
+    if pec:
+        print(f"PEC: {pec}")
+    if phone:
+        print(f"Telefono: {phone}")
+    if recipient_code:
+        print(f"Codice Destinatario: {recipient_code}")
+
+    return f"La persona fisica {first_name} {last_name} è stata creata con successo!"
+
+@tool
+def create_company_mode(company_type: str) -> str:
+    """Permette di scegliere se inserire i dati societari tramite Partiva IVA o Codice Fiscale.
+
+    Parametri:
+    - company_type (str): Modalità di inserimento delle informazioni societarie. Deve essere 'vat_number' (Partiva IVA) o 'tax_code' (Codice Fiscale).
+
+    Ritorna:
+    - str: Un messaggio di conferma con il tipo di selezionato.
+    """
+
+    if customer_type not in ["vat_number", "tax_code"]:
+        return "Errore: Il tipo di modalità deve essere 'vat_number' (Partiva IVA) o 'tax_code' (Codice Fiscale)."
+
+    return f"Hai selezionato {company_type}. Ora puoi procedere con l'inserimento dei dati."
+
+@tool
+def create_company_tax_code(
+    company_name: str, 
+    tax_code: str, 
+    address: str, 
+    street_number: str, 
+    postal_code: str, 
+    city: str, 
+    province: str, 
+    country: str, 
+    vat_number: str = None,
+    email: str = None, 
+    pec: str = None, 
+    phone: str = None, 
+    recipient_code: str = None
+) -> str:
+    """Crea un cliente di tipo società tramite Codice Fiscale.
+
+    Parametri:
+    - company_name (str): Nome della società (ragione sociale).
+    - tax_code (str): Codice fiscale della società.
+    - address (str): Indirizzo della sede legale.
+    - street_number (str): Numero civico.
+    - postal_code (str): CAP.
+    - city (str): Città.
+    - province (str): Provincia.
+    - country (str): Nazione.
+    - vat_number (str, opzionale): Partita IVA.
+    - email (str, opzionale): Indirizzo email.
+    - pec (str, opzionale): Indirizzo PEC.
+    - phone (str, opzionale): Numero di telefono.
+    - recipient_code (str, opzionale): Codice destinatario.
+
+    Ritorna:
+    - str: Un messaggio di conferma con i dati della società creata.
+    """
+
+    print("\n--- Nuova Società Creata ---")
+    print(f"Ragione Sociale: {company_name}")
+    print(f"Codice Fiscale: {tax_code}")
+    print(f"Indirizzo: {address}, {street_number}, {postal_code}, {city}, {province}, {country}")
+
+    if vat_number:
+        print(f"Partita IVA: {vat_number}")
+    if email:
+        print(f"Email: {email}")
+    if pec:
+        print(f"PEC: {pec}")
+    if phone:
+        print(f"Telefono: {phone}")
+    if recipient_code:
+        print(f"Codice Destinatario: {recipient_code}")
+
+    return f"La società {company_name} è stata creata con successo!"
+
+@tool
+def create_company_vat_number(
+    vat_number: str = None,
+) -> str:
+    """Crea un cliente di tipo società tramite Partita IVA, non vanno richiesti altri dati, solo confermare la Partita IVA.
+
+    Parametri:
+    - vat_number (str): Partita IVA.
+
+    Ritorna:
+    - str: Un messaggio di conferma con i dati della Partita IVA della società creata.
+    """
+
+    print("\n--- Nuova Società Creata ---")
+    print(f"Partita IVA: {vat_number}")
+
+    return f"La società con Partita IVA {vat_number} è stata creata con successo!"
+
+
+tools = [
+    create_product, 
+    create_customer_type, 
+    create_individual, 
+    create_company_mode, 
+    create_company_tax_code, 
+    create_company_vat_number
+]
+
+llm = ChatOpenAI(
+    model="gpt-4o-mini", 
+    streaming=True,
+)
+llm_with_tools = llm.bind_tools(tools)
+
+tool_name = ""
+tool_args = ""
+
+tool_mapping = {
+    #"create_customer_type": create_customer_type,
+    "create_individual": create_individual,
+    "create_company_tax_code": create_company_tax_code,
+    "create_company_vat_number": create_company_vat_number,
+    "create_product": create_product
+}
+
+label_buttons = ["Procedi", "Annulla"]
+
+def _check_response(ai_msg):
+
+    if "persona fisica" in ai_msg.lower() and "società" in ai_msg.lower():
+
+        context_analysis_prompt = f"""
+        Questa è la risposta dell'LLM a un utente che vuole creare un cliente:
+
+        \"{ai_msg}\"
+
+        Devi solo rispondere con "create_client" se la risposta chiede all'utente di selezionare tra "Persona Fisica" e "Società", 
+        oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale dei tipi di clienti senza richiedere un'azione.
+        """
+        analysis_response = llm.invoke(context_analysis_prompt).content.strip()
+        print(analysis_response)
+
+        if analysis_response == "create_client":
+            return True, analysis_response
+
+    elif "partita iva" in ai_msg.lower() and "codice fiscale" in ai_msg.lower():
+
+        context_analysis_prompt = f"""
+        Questa è la risposta dell'LLM a un utente che deve scegliere la modalità con cui vuole inserire i dati societari
+
+        \"{ai_msg}\"
+
+        Devi solo rispondere con "select_mode" se la risposta chiede all'utente di selezionare tra "Partita IVA" e "Codice Fiscale", 
+        oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale senza richiedere un'azione.
+        """
+        analysis_response = llm.invoke(context_analysis_prompt).content.strip()
+        print(analysis_response)
+
+        if analysis_response == "select_mode":
+            return True, analysis_response
+
+    return False, ""
+
+def _get_confirmation_message(tool_name: str, tool_args: dict) -> str:
+    """Genera un messaggio descrittivo per confermare l'azione che sta per essere eseguita."""
+    
+    if tool_name == "create_individual":
+        return f"**ACTION** Stai per creare un cliente di tipo Persona Fisica: {tool_args.get('first_name', 'N/D')} {tool_args.get('last_name', 'N/D')}."
+
+    elif tool_name == "create_company_tax_code":
+        return f"**ACTION** Stai per creare una Società: {tool_args.get('company_name', 'N/D')}."
+
+    elif tool_name == "create_company_vat_number":
+        return f"**ACTION** Stai per creare una Società con Partita IVA: {tool_args.get('vat_number', 'N/D')}."
+
+    elif tool_name == "create_product":
+        return f"**ACTION** Stai per creare un nuovo prodotto o servizio: '{tool_args.get('name', 'N/D')}' al prezzo di {tool_args.get('price', 0):.2f}€."
+
+    return "**ACTION** Stai per eseguire un'operazione sconosciuta."
+
+def _format_tool_args(tool_name: str, tool_args: dict) -> str:
+    """Trasforma i parametri tecnici in una descrizione leggibile per l'utente, gestendo i valori opzionali."""
+    
+    translations = {
+        "first_name": "Nome",
+        "last_name": "Cognome",
+        "tax_code": "Codice Fiscale",
+        "address": "Indirizzo",
+        "street_number": "Numero Civico",
+        "postal_code": "CAP",
+        "city": "Città",
+        "province": "Provincia",
+        "country": "Nazione",
+        "vat_number": "Partita IVA",
+        "email": "Email",
+        "pec": "PEC",
+        "phone": "Telefono",
+        "recipient_code": "Codice Destinatario",
+        "company_name": "Ragione Sociale",
+        "name": "Nome del Prodotto",
+        "price": "Prezzo",
+        "amount": "Importo"
+    }
+    
+    _formatted_args = "\n".join([
+        f"- {translations.get(key, key)}: {value if value else 'Non specificato'}" 
+        for key, value in tool_args.items()
+    ])
+    
+    return _formatted_args
+
+def _get_company_info(vat_number: str) -> dict:
+    print(f"Getting company info for vat_number: {vat_number}")
+
+    company_info = {
+        "company_name": "NewCo",
+        "tax_code": "abcdefghilmnopqr",
+        "vat_number": vat_number,
+        "address": "via Roma",
+        "street_number": "12",
+        "postal_code": "00100",
+        "city": "Roma",
+        "province": "RM",
+        "country": "IT",
+        "email": "aaa@bbb.ccc",
+        "pec": "aaa@bbb.ccc",
+        "phone": "1234567890",
+        "recipient_code": "abcabcabc"
+    }
+
+    return company_info
+
+def chatbot_response(message, history):
+    
+    global tool_name, tool_args
+
+    history = history or []
+
+    if message.startswith("**OPERAZIONE ANNULLATA**"):
+        history.append({"role": "user", "content": "**OPERAZIONE ANNULLATA**"})
+    else:
+        history.append({"role": "user", "content": message})
+
+    messages = [SystemMessage(content=system_message)]
+    truncated_history = history[-MAX_HISTORY:] if len(history) > MAX_HISTORY else history
+
+    #print(f"Truncated history: {truncated_history}")
+
+    for entry in truncated_history:
+        if entry["role"] == "user":
+            messages.append(HumanMessage(content=entry["content"]))
+        elif entry["role"] == "assistant":
+            messages.append(AIMessage(content=entry["content"]))
+
+    if message is not None:
+        messages.append(HumanMessage(content=message))
+        
+        #_response=llm_with_tools.stream(history_langchain_format)
+        ai_msg=llm_with_tools.invoke(messages)
+        print(ai_msg)
+
+        choice, analysis_response = _check_response(ai_msg.content)
+
+        if choice and analysis_response == "create_client":
+            history.append({"role": "assistant", "content": f"**CREAZIONE CLIENTE**\n{ai_msg.content}"})
+
+        elif choice and analysis_response == "select_mode":
+            history.append({"role": "assistant", "content": f"**MODALITA' INSERIMENTO DATI SOCIETARI**\n{ai_msg.content}"})
+
+        elif hasattr(ai_msg, "tool_calls") and ai_msg.tool_calls:
+    
+            for tool_call in ai_msg.tool_calls:
+                tool_name = tool_call["name"].lower()
+                tool_args = tool_call["args"]
+                selected_tool = tool_mapping.get(tool_name)
+
+                if selected_tool:
+                    if selected_tool==create_company_vat_number:
+                        tool_args=_get_company_info(tool_args.get('vat_number', 'N/D'))
+                        if tool_args is None:
+                            history.append({"role": "user", "content": "Errore: Dati societari non trovati."})
+                            return history
+
+                    confirmation_message = _get_confirmation_message(tool_name, tool_args)
+                    formatted_args = _format_tool_args(tool_name, tool_args)
+                    ai_msg= f"{confirmation_message}\n\n**Ecco i dettagli inseriti:**\n{formatted_args}"
+                    history.append({"role": "assistant", "content": ai_msg})                            
+
+        else:
+            history.append({"role": "assistant", "content": ai_msg.content})
+
+        return history
+
+def reset_textbox():
+    """Clears the textbox after sending a message."""
+    return gr.update(value="")
+
+def show_buttons(history):
+    global label_buttons
+
+    if history and history[-1]["content"].startswith("**ACTION**"):
+        label_buttons = ["Procedi", "Annulla"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1])
+        )
+    elif history and history[-1]["content"].startswith("**CREAZIONE CLIENTE**"):
+        label_buttons = ["Persona Fisica", "Società"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1])
+        )
+    elif history and history[-1]["content"].startswith("**MODALITA' INSERIMENTO DATI SOCIETARI**"):
+        #print("SONO QUI")
+        label_buttons = ["Partita IVA", "Codice Fiscale"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1])
+        )
+    return (
+        gr.update(visible=True), 
+        gr.update(visible=False), 
+        gr.update(visible=False)
+    )
+
+def hide_buttons():
+    """Hides buttons and shows the textbox after clicking a button."""
+    return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)
+
+def show_or_hide_buttons(history):
+    global label_buttons
+
+    if history and history[-1]["content"].startswith("**ACTION**"):
+        label_buttons = ["Procedi", "Annulla"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1]),
+            gr.update(visible=False)
+        )
+    elif history and history[-1]["content"].startswith("**CREAZIONE CLIENTE**"):
+        label_buttons = ["Persona Fisica", "Società"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1]),
+            gr.update(visible=False)
+        )
+    elif history and history[-1]["content"].startswith("**MODALITA' INSERIMENTO DATI SOCIETARI**"):
+        #print("SONO QUI")
+        label_buttons = ["Partita IVA", "Codice Fiscale"]
+        return (
+            gr.update(visible=False), 
+            gr.update(visible=True, value=label_buttons[0]), 
+            gr.update(visible=True, value=label_buttons[1]),
+            gr.update(visible=False)
+        )
+    return (
+        gr.update(visible=True), 
+        gr.update(visible=False), 
+        gr.update(visible=False),
+        gr.update(visible=True)
+    )
+
+def button_clicked(option, history):
+    """Handles button clicks and updates chat history."""
+
+    global tool_name, tool_args
+
+    print(f"Button clicked: {option}")
+
+    if (option == "Procedi"):
+        history.append({"role": "user", "content": option})
+        if tool_name and tool_args:
+            selected_tool = tool_mapping.get(tool_name)
+            if selected_tool:
+                tool_output = selected_tool.invoke(tool_args)
+                history.append({"role": "assistant", "content": tool_output})
+            else:
+                history.append({"role": "user", "content": "Operazione annullata"})
+    elif option == "Annulla":
+        history.append({"role": "user", "content": option})
+        llm_message = (
+            "**OPERAZIONE ANNULLATA**\n"
+            "Ho annullato l'operazione corrente.\n"
+            "Dobbiamo modificare alcuni parametri oppure passare a una nuova operazione.\n"
+        )
+        history = chatbot_response(llm_message, history)
+    elif option == "Persona Fisica":
+        llm_message = "Persona Fisica"
+        history = chatbot_response(llm_message, history)
+    elif option == "Società":
+        llm_message = "Società"
+        history = chatbot_response(llm_message, history)
+    elif option == "Partita IVA":
+        llm_message = "Partita IVA"
+        history = chatbot_response(llm_message, history)
+    elif option == "Codice Fiscale":
+        llm_message = "Codice Fiscale"
+        history = chatbot_response(llm_message, history)
+    
+    tool_name = ""
+    tool_args = ""
+    
+    return history
+
+# Authentication function
+def authenticate(username, password):
+    if username in credentials and credentials[username] == password:
+        print("🔑 Login successful!")
+        return gr.update(visible=False), gr.update(visible=True), gr.update(value="", visible=False)  # Hide login, show chatbot, clear error
+    else:
+        print("❌ Incorrect username or password")
+        return gr.update(visible=True), gr.update(visible=False), gr.update(value="❌ Incorrect username or password", visible=True)  # Show error
+
+
+with gr.Blocks() as demo:
+
+    # 🔒 Login Section (Initially Visible)
+    with gr.Column(visible=True) as login_section:
+        gr.Markdown("### 🔒 Login Required")
+        username_input = gr.Textbox(label="Username")
+        password_input = gr.Textbox(label="Password", type="password")
+        login_button = gr.Button("Login")
+        error_message = gr.Text("", visible=False)
+
+    # 🧠 Chatbot Section (Initially Hidden)
+    with gr.Column(visible=False) as chatbot_section:
+
+        chatbot = gr.Chatbot(
+            label="Assistente Forfè",
+            type="messages"
+        )  
+
+        user_input = gr.Textbox(label="Utente",placeholder="Cosa vuoi chiedere al tuo assistente Forfè?")
+
+        with gr.Row():
+            btn1 = gr.Button("", visible=False)
+            btn2 = gr.Button("", visible=False)
+
+        send_btn = gr.Button("Invia")
+
+        # When user submits text
+        user_input.submit(chatbot_response, [user_input, chatbot], chatbot) \
+                .then(reset_textbox, None, user_input) \
+                .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
+                #.then(show_buttons, chatbot, [user_input, btn1, btn2]) 
+
+        # When user clicks send
+        send_btn.click(chatbot_response, [user_input, chatbot], chatbot) \
+                .then(reset_textbox, None, user_input) \
+                .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
+                #.then(show_buttons, chatbot, [user_input, btn1, btn2]) 
+
+        # Button clicks: Show textbox, hide buttons
+        btn1.click(lambda h: button_clicked(label_buttons[0], h), chatbot, chatbot) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
+            #.then(hide_buttons, None, [user_input, btn1, btn2])
+        
+        btn2.click(lambda h: button_clicked(label_buttons[1], h), chatbot, chatbot) \
+            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
+            #.then(hide_buttons, None, [user_input, btn1, btn2])
+
+    # 🔑 Login Button Action (Now updates visibility correctly)
+    login_button.click(
+        authenticate, 
+        [username_input, password_input], 
+        [login_section, chatbot_section, error_message]
+    )
+
+
+
+demo.launch(
+    debug=True, 
+    #share=True
+)

requirements.txt

@@ -0,0 +1,9 @@
+gradio==5.16.0
+langchain==0.3.18
+langchain_community==0.3.17
+langchain_core==0.3.34
+langchain_openai==0.3.5
+langchain_qdrant==0.2.0
+python-dotenv==1.0.1
+qdrant_client==1.13.2
+sentence_transformers==3.4.1

utilities/qdrant/QdrantLangchainManager.py

@@ -0,0 +1,206 @@
+import os
+from dotenv import load_dotenv
+
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams
+
+from langchain_openai import ChatOpenAI, OpenAIEmbeddings
+from langchain_qdrant import QdrantVectorStore
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_community.document_loaders import TextLoader, PyPDFLoader
+from langchain_core.documents import Document
+from langchain_core.messages import SystemMessage, HumanMessage
+
+from sentence_transformers import CrossEncoder
+
+class QdrantLangchainManager:
+    
+    def __init__(self, qdrant_url, qdrant_api_key, 
+                 system_message=None, llm=None, crossencoder=None,
+                 batch_size=500, chunk_size=2000, chunk_overlap=50, vector_size=1536, re_ranking_threshold=1):
+        
+        self.qdrant_url = qdrant_url
+        self.qdrant_api_key = qdrant_api_key
+        self.client = QdrantClient(url=qdrant_url, api_key=qdrant_api_key)
+        
+        self.system_message = system_message or """
+        You are a helpful assistant that assists users in finding solutions and answering their questions.
+        If a question is related to our specific environment, I will provide additional [context].
+        You should consider this [context] when formulating your response.
+        However, if the provided [context] is not relevant to the question, you can proceed based on your general knowledge.
+        Always prioritize clarity and usefulness in your answers.
+        
+        [context]
+        {context}
+        """
+        
+        self.llm = llm
+        self.crossencoder = crossencoder if crossencoder else None
+        
+        self.batch_size = batch_size
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.vector_size = vector_size
+        self.re_ranking_threshold = re_ranking_threshold
+        
+        self.vectorstore = None
+    
+    def create_collection(self, collection_name):
+        try:
+            if not self.client.collection_exists(collection_name):
+                self.client.create_collection(
+                    collection_name=collection_name,
+                    vectors_config=VectorParams(size=self.vector_size, distance=Distance.COSINE)
+                )
+                print(f"✅ Collection '{collection_name}' created successfully.")
+            else:
+                print(f"⚠️ Collection '{collection_name}' already exists.")
+            
+            self.vectorstore = QdrantVectorStore(
+                client=self.client,
+                collection_name=collection_name,
+                embedding=OpenAIEmbeddings(),
+            )
+            return True
+        except Exception as e:
+            print(f"❌ Error creating collection '{collection_name}': {e}")
+            return False
+    
+    def get_collection(self, collection_name):
+        try:
+            if not self.client.collection_exists(collection_name):
+                print(f"⚠️ Collection '{collection_name}' doesn't exist.")
+                return False
+            self.vectorstore = QdrantVectorStore(
+                client=self.client,
+                collection_name=collection_name,
+                embedding=OpenAIEmbeddings(),
+            )
+            return True
+        except Exception as e:
+            print(f"❌ Error getting collection '{collection_name}': {e}")
+            return False
+    
+    def insert_documents(self, file_path):
+        if not self.vectorstore:
+            print("⚠️ No collection initialized. Please create or load a collection first.")
+            return False
+        
+        try:
+            file_extension = os.path.splitext(file_path)[-1].lower()
+            
+            loader = TextLoader(file_path) if file_extension == ".txt" else PyPDFLoader(file_path) if file_extension == ".pdf" else None
+            if not loader:
+                raise ValueError(f"Unsupported file type: {file_extension}")
+            
+            docs = loader.load()
+            text_splitter = RecursiveCharacterTextSplitter(chunk_size=self.chunk_size, chunk_overlap=self.chunk_overlap)
+            chunks = text_splitter.split_documents(docs)
+            
+            max_iterations = 3
+            iteration = 0
+            merge_performed = True
+            while merge_performed and iteration < max_iterations:
+                chunks, merge_performed = self._merge_chunks(chunks)
+                iteration += 1
+                print(f"🔄 Merge Iteration {iteration}: {len(chunks)} chunks remain.")
+            
+            for i in range(0, len(chunks), self.batch_size):
+                batch = chunks[i:i + self.batch_size]
+                self.vectorstore.add_documents(batch)
+                print(f"✅ Inserted {len(batch)} documents (Batch {i // self.batch_size + 1})")
+            return True
+        except Exception as e:
+            print(f"❌ Error inserting documents: {e}")
+            return False
+        
+    def search_qdrant(self, query, top_k=3):
+        if not self.vectorstore:
+            print("⚠️ No collection initialized. Please create or load a collection first.")
+            return None
+
+        if not self.llm:
+            print("⚠️ LLM not initialized. Cannot process response.")
+            return None
+        
+        try:
+            docs_with_scores = self.vectorstore.similarity_search_with_score(query, k=top_k)
+
+            print("\n🔍 Search Results:")
+            context=""
+            for doc, score in docs_with_scores:
+                print(f"* [score={score:3f}] {doc.page_content[:10]}")
+
+            if self.crossencoder:
+                docs_with_scores = self._reranking(query, docs_with_scores)
+                for (doc, old_score), new_score in docs_with_scores:
+                    print(f"🔹 Old Score: {old_score:.2f} ➝ New Score: {new_score:.2f} - {doc.page_content[:10]}...")
+            
+            top_score = docs_with_scores[0][1] if docs_with_scores else 0
+            context = "\n\n".join([doc.page_content for (doc, _), new_score in docs_with_scores if new_score > self.re_ranking_threshold]) if top_score > 0 else ""
+            final_system_message = self.system_message.format(context=context) if context else self.system_message.replace("[context]\n{context}", "").strip()
+                        
+            messages = [SystemMessage(content=final_system_message), HumanMessage(content=query)]
+            ai_msg = self.llm.invoke(messages)
+            #print(ai_msg)
+            # Ensure `ai_msg` contains a valid response
+            if hasattr(ai_msg, "content"):
+                return ai_msg.content  # Extract only the useful content
+            else:
+                return None
+        except Exception as e:
+            print(f"❌ Error during search: {e}")
+            return None
+    
+    def delete_collection(self, collection_name):
+        try:
+            self.client.delete_collection(collection_name)
+            print(f"🚨 Collection '{collection_name}' has been deleted.")
+            self.vectorstore = None
+            return True
+        except Exception as e:
+            print(f"❌ Error deleting collection '{collection_name}': {e}")
+            return False
+
+    def _merge_chunks(self, chunks, min_size=500, max_size=2000):
+        if not chunks:
+            return [], False
+        
+        merged_chunks = []
+        temp_text = ""
+        merge_performed = False
+        
+        if len(chunks[0].page_content) < min_size and len(chunks) > 1:
+            chunks[1] = Document(page_content=chunks[0].page_content + " " + chunks[1].page_content)
+            chunks = chunks[1:]
+            merge_performed = True
+        
+        for chunk in chunks:
+            text = chunk.page_content
+            if not temp_text:
+                temp_text = text
+                continue
+            if len(text) < min_size:
+                temp_text += " " + text
+                merge_performed = True
+            else:
+                while len(temp_text) > max_size:
+                    merged_chunks.append(Document(page_content=temp_text[:max_size]))
+                    temp_text = temp_text[max_size:]
+                merged_chunks.append(Document(page_content=temp_text))
+                temp_text = text
+        
+        if temp_text:
+            merged_chunks.append(Document(page_content=temp_text))
+        
+        return merged_chunks, merge_performed
+    
+    def _reranking(self, query, docs_with_scores):
+        if not self.crossencoder:
+            print("⚠️ Crossencoder not initialized. Skipping reranking.")
+            return docs_with_scores
+        
+        reranker = CrossEncoder(self.crossencoder)
+        query_pairs = [(query, doc.page_content) for doc, _ in docs_with_scores]
+        new_scores = reranker.predict(query_pairs)
+        return sorted(zip(docs_with_scores, new_scores), key=lambda x: x[1], reverse=True)

utilities/qdrant/langchain_utils.py

@@ -0,0 +1,248 @@
+import os
+from dotenv import load_dotenv
+
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+
+from langchain_openai import ChatOpenAI
+from langchain_openai import OpenAIEmbeddings
+from langchain_qdrant import QdrantVectorStore
+
+from langchain.text_splitter import RecursiveCharacterTextSplitter,CharacterTextSplitter
+from langchain_community.document_loaders import TextLoader, PyPDFLoader
+from langchain_core.documents import Document
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
+
+from sentence_transformers import CrossEncoder
+
+load_dotenv()
+
+qdrant_url=os.getenv("QDRANT_URL")
+qdrant_api_key=os.getenv("QDRANT_API_KEY")
+
+BATCH_SIZE=500
+CHUNK_SIZE=2000
+CHUNK_OVERLAP=50
+VECTOR_SIZE=1536
+RE_RANKING_THRESHOLD=2
+
+system_message = """
+You are a helpful assistant that assists users in finding solutions and answering their questions.
+If a question is related to our specific environment, I will provide additional [context].
+You should consider this [context] when formulating your response.
+
+However, if the provided [context] is not relevant to the question, you can proceed based on your general knowledge.
+Always prioritize clarity and usefulness in your answers.
+    
+[context]
+{context}
+"""
+
+
+# Initialize Qdrant client
+client = QdrantClient(
+    url=qdrant_url, 
+    api_key=qdrant_api_key
+)
+
+llm = ChatOpenAI(
+    model="gpt-4o-mini", 
+    streaming=True,
+)
+
+messages = [SystemMessage(content=system_message)] 
+
+"""LANGCHAIN CRUD_OPERATIONS"""
+def langchain_create_collection(collection_name: str, vector_size: int = VECTOR_SIZE):
+    """Create a new Qdrant collection and return a LangChain vector store."""
+    # Create the collection (or recreate if it exists)
+    if not client.collection_exists(collection_name):
+        client.create_collection(
+            collection_name=collection_name,
+            vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE)
+        )
+        print(f"✅ Collection '{collection_name}' created successfully.")
+    else:
+        print(f"⚠️ Collection '{collection_name}' already exists.")
+
+    # Now initialize the LangChain Qdrant vector store
+    vectorstore = QdrantVectorStore(
+        client=client,
+        collection_name=collection_name,
+        embedding=OpenAIEmbeddings(),
+    )
+
+    return vectorstore
+
+def langchain_get_collection(collection_name: str, vector_size: int = VECTOR_SIZE):
+    """Create a new Qdrant collection and return a LangChain vector store."""
+    # Create the collection (or recreate if it exists)
+    if not client.collection_exists(collection_name):
+        print(f"⚠️ Collection '{collection_name}' doesn't exist.")
+        return None
+
+    # Now initialize the LangChain Qdrant vector store
+    vectorstore = QdrantVectorStore(
+        client=client,
+        collection_name=collection_name,
+        embedding=OpenAIEmbeddings(),
+    )
+
+    return vectorstore
+
+def langchain_insert_documents(file_path: str, vectorstore):
+    """Load, chunk, and insert documents into Qdrant via LangChain."""
+    
+    file_extension = os.path.splitext(file_path)[-1].lower()
+    
+    if file_extension == ".txt":
+        loader = TextLoader(file_path)
+    elif file_extension == ".pdf":
+        loader = PyPDFLoader(file_path)
+    else:
+        raise ValueError(f"Unsupported file type: {file_extension}")
+    
+    docs = loader.load()
+
+    text_splitter = RecursiveCharacterTextSplitter(
+        chunk_size=CHUNK_SIZE,  
+        chunk_overlap=CHUNK_OVERLAP,
+    )
+    chunks = text_splitter.split_documents(docs)
+    
+    # Iterative merging
+    max_iterations = 3  # Set max iterations
+    iteration = 0
+    merge_performed = True  # Initialize to True for first iteration
+
+    while merge_performed and iteration < max_iterations:
+        chunks, merge_performed = _merge_chunks(chunks)
+        iteration += 1
+        print(f"🔄 Merge Iteration {iteration}: {len(chunks)} chunks remain.")
+    
+    # how many chunks
+    print(f"📄 Loaded {len(docs)} documents from '{file_path}'.")
+    print(f"🔄 Final Chunks after merging: {len(chunks)} total.")
+    for i, chunk in enumerate(chunks, start=1):
+        print(f"Chunk {i}: {len(chunk.page_content)} characters - {chunk.page_content[:10]}...")
+    
+    # 🔥 Insert data in small batches to prevent timeout
+    for i in range(0, len(chunks), BATCH_SIZE):
+        batch = chunks[i:i + BATCH_SIZE]
+        vectorstore.add_documents(batch)
+        print(f"✅ Inserted {len(batch)} documents (Batch {i // BATCH_SIZE + 1})")
+
+    print(f"✅ Finished inserting {len(chunks)} total documents.")
+
+def langchain_search_qdrant(query: str, vectorstore, top_k: int = 3):
+    """Search Qdrant for the most relevant documents."""
+    # retriever = vectorstore.as_retriever(search_kwargs={"k": top_k})
+    # results = retriever.invoke(query)
+
+    docs_with_scores = vectorstore.similarity_search_with_score(query, k=top_k)
+
+    print("\n🔍 Search Results:")
+    context=""
+    for doc, score in docs_with_scores:
+        print(f"* [score={score:3f}] {doc.page_content[:10]}")
+
+    docs_re_ranked = _reranking(query, docs_with_scores)
+    for (doc, old_score), new_score in docs_re_ranked:
+        print(f"🔹 Old Score: {old_score:.2f} ➝ New Score: {new_score:.2f} - {doc.page_content[:10]}...")
+
+    top_score = docs_re_ranked[0][1]  # Get highest new score
+    if top_score > 0:
+        filtered_results = [doc for (doc, _), new_score in docs_re_ranked if new_score > RE_RANKING_THRESHOLD]
+        context = "\n\n".join([doc.page_content for doc in filtered_results])
+        final_system_message = system_message.format(context=context)
+    else:
+        final_system_message = system_message.replace("[context]\n{context}", "").strip()
+    
+    messages = [SystemMessage(content=final_system_message)] 
+    messages.append(HumanMessage(content=query))
+
+    ai_msg=llm.invoke(messages)
+    print(ai_msg)
+
+def langchain_update_document(vectorstore, collection_name: str, point_id: int, new_text: str):
+    """Update an existing document in Qdrant."""
+    # Delete old vector
+    client.delete(collection_name=collection_name, points=[point_id])
+
+    # Insert new data
+    new_vector = OpenAIEmbeddings().embed_query(new_text)
+    vectorstore.add_texts([new_text], metadatas=[{"id": point_id}], ids=[point_id])
+
+    print(f"✅ Updated document ID {point_id} in '{collection_name}'.")
+
+def langchain_delete_point(vectorstore, collection_name: str, point_id: int):
+    """Delete a specific vector from Qdrant."""
+    client.delete(collection_name=collection_name, points=[point_id])
+    print(f"🗑️ Deleted document ID {point_id} from '{collection_name}'.")
+
+def langchain_delete_collection(collection_name: str):
+    """Delete an entire collection from Qdrant."""
+    client.delete_collection(collection_name=collection_name)
+    print(f"🚨 Collection '{collection_name}' has been deleted.")
+
+"""PRIVATE FUNCTIONS"""
+def _merge_chunks(chunks, min_size=500, max_size=2000):
+    """Merge small chunks into larger ones while keeping document structure intact.
+    
+    Returns:
+    - merged_chunks: List of Document objects after merging
+    - merge_performed: Boolean indicating if merging happened
+    """
+    if not chunks:
+        return [], False
+
+    merged_chunks = []
+    temp_text = ""
+    merge_performed = False  # ✅ Track if we performed a merge
+
+    # ✅ Special case: Ensure the FIRST chunk is not too small
+    if len(chunks[0].page_content) < min_size and len(chunks) > 1:
+        chunks[1] = Document(page_content=chunks[0].page_content + " " + chunks[1].page_content)
+        chunks = chunks[1:]  # Remove the first chunk (merged into the second)
+        merge_performed = True  # ✅ Mark that a merge was performed
+
+    for chunk in chunks:
+        text = chunk.page_content  # Extract text content from Document
+
+        if not temp_text:
+            temp_text = text
+            continue
+
+        # Merge small chunks
+        if len(text) < min_size:
+            temp_text += " " + text  # Append small chunk to previous
+            merge_performed = True  # ✅ A merge happened
+        else:
+            # Ensure chunk does not exceed max_size
+            while len(temp_text) > max_size:
+                merged_chunks.append(Document(page_content=temp_text[:max_size]))  # ✅ Convert to Document
+                temp_text = temp_text[max_size:]
+
+            merged_chunks.append(Document(page_content=temp_text))  # ✅ Convert to Document
+            temp_text = text  # Start new chunk
+
+    # Add any remaining chunk
+    if temp_text:
+        merged_chunks.append(Document(page_content=temp_text))  # ✅ Convert to Document
+
+    return merged_chunks, merge_performed  # ✅ Return boolean
+
+def _reranking(query, docs_with_scores):
+    
+    reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
+
+    doc_texts = [doc.page_content for doc, _ in docs_with_scores]
+    query_pairs = [(query, doc) for doc in doc_texts]
+    new_scores = reranker.predict(query_pairs)
+
+    re_ranked_results = sorted(zip(docs_with_scores, new_scores), key=lambda x: x[1], reverse=True)
+
+    # for (doc, old_score), new_score in re_ranked_results:
+    #     print(f"🔹 Old Score: {old_score:.2f} ➝ New Score: {new_score:.2f} - {doc.page_content[:10]}...")
+    
+    return re_ranked_results

utilities/qdrant/test.py

@@ -0,0 +1,59 @@
+from dotenv import load_dotenv
+import os
+
+from langchain_utils import *
+from QdrantLangchainManager import *
+
+
+load_dotenv()
+
+# parameters
+qdrant_url=os.getenv("QDRANT_URL")
+qdrant_api_key=os.getenv("QDRANT_API_KEY")
+collection_name="5g reference"
+llm = ChatOpenAI(
+    model="gpt-4o-mini", 
+    streaming=True
+)
+crossencoder = "cross-encoder/ms-marco-MiniLM-L-6-v2"
+
+manager = QdrantLangchainManager(
+    qdrant_url=qdrant_url,
+    qdrant_api_key=qdrant_api_key,
+    llm=llm,
+    crossencoder=crossencoder
+)
+
+# result = manager.create_collection("my_collection")
+# print("The result is: ", result)
+# if result:
+#     result=manager.delete_collection("my_collection")
+#     print("The result is: ", result)
+
+result = manager.get_collection(collection_name)
+print("The result is: ", result)
+#query="who is jokerbirot?"
+query="Can you tell me something about 5g modem?"
+result = manager.search_qdrant(query)
+print(result)
+
+# manager.insert_documents("example.pdf", vectorstore)
+# manager.search_qdrant("example query", vectorstore)
+
+# langchain_delete_collection(collection_name)
+# vectorestore=langchain_create_collection(collection_name)
+# langchain_insert_documents("../../data/5G reference - en.pdf", vectorestore)
+
+# vectorestore=langchain_get_collection(collection_name)
+
+#query="who is jokerbirot?"
+#query="what is a qubit?"
+#query="Configuring a 5G SA Modem"
+#langchain_search_qdrant(query, vectorestore)
+
+# vectorstore=langchain_create_collection("jokerbirot_saga")
+# langchain_insert_documents("data/jokerbirot_saga.txt", vectorstore)
+
+# query="who is jokerbirot?"
+# langchain_search_qdrant(query, vectorstore)
+

utilities/qdrant/utils.py

@@ -0,0 +1,108 @@
+import os
+from dotenv import load_dotenv
+
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+
+load_dotenv()
+
+qdrant_url=os.getenv("QDRANT_URL")
+qdrant_api_key=os.getenv("QDRANT_API_KEY")
+
+BATCH_SIZE=500
+CHUNK_SIZE=2000
+CHUNK_OVERLAP=50
+VECTOR_SIZE=1536
+RE_RANKING_THRESHOLD=2
+
+system_message = """
+You are a helpful assistant that assists users in finding solutions and answering their questions.
+If a question is related to our specific environment, I will provide additional [context].
+You should consider this [context] when formulating your response.
+
+However, if the provided [context] is not relevant to the question, you can proceed based on your general knowledge.
+Always prioritize clarity and usefulness in your answers.
+    
+[context]
+{context}
+"""
+
+# Initialize Qdrant client
+client = QdrantClient(
+    url=qdrant_url, 
+    api_key=qdrant_api_key
+)
+
+def create_collection(collection_name: str, vector_size: int = VECTOR_SIZE):
+    """Create a new Qdrant collection."""
+    if not client.collection_exists(collection_name):
+        client.create_collection(
+            collection_name=collection_name,
+            vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE)
+        )
+        print(f"✅ Collection '{collection_name}' created successfully.")
+    else:
+        print(f"⚠️ Collection '{collection_name}' already exists.")
+
+def get_collection(collection_name: str, vector_size: int = VECTOR_SIZE):
+    """Create a new Qdrant collection and return a LangChain vector store."""
+    # Create the collection (or recreate if it exists)
+    if not client.collection_exists(collection_name):
+        print(f"⚠️ Collection '{collection_name}' doesn't exist.")
+
+def delete_collection(collection_name: str):
+    """Delete an entire collection."""
+    client.delete_collection(collection_name=collection_name)
+    print(f"🚨 Collection '{collection_name}' has been deleted.")
+
+def list_collections():
+    """List all collections in Qdrant."""
+    collections = client.get_collections()
+    print("🔹 Available Collections:")
+    for collection in collections.collections:
+        print(f"- {collection.name}")
+
+def insert_data(collection_name: str, point_id: int, vector: list, payload: dict):
+    """Insert a vector with metadata into Qdrant."""
+
+    """
+    Example Usage
+    sample_vector = np.random.rand(1536).tolist()  # Fake embedding vector
+    insert_data("sample_collection", point_id=1, vector=sample_vector, payload={"text": "This is a sample document."})
+    """
+
+    client.upsert(
+        collection_name=collection_name,
+        points=[
+            PointStruct(id=point_id, vector=vector, payload=payload)
+        ]
+    )
+    print(f"✅ Data inserted into '{collection_name}' with ID {point_id}")
+
+def delete_point(collection_name: str, point_id: int):
+    """Delete a specific point from a collection."""
+
+    """
+    Example Usage
+    delete_point("sample_collection", 1)
+    """
+
+    client.delete(collection_name=collection_name, points=[point_id])
+    print(f"🗑️ Deleted point ID {point_id} from '{collection_name}'.")
+
+def search_data(collection_name: str, query_vector: list, top_k: int = 3):
+    """Search for the closest vectors in Qdrant."""
+
+    """
+    Example Usage
+    search_data("sample_collection", query_vector=sample_vector)
+    """
+
+    results = client.search(
+        collection_name=collection_name,
+        query_vector=query_vector,
+        limit=top_k
+    )
+    print("🔍 Search Results:")
+    for res in results:
+        print(f"- ID: {res.id}, Score: {res.score}, Payload: {res.payload}")