add assistant script

core/demorrha.py

@@ -0,0 +1,171 @@
+#coding: utf-8
+
+import re
+from os import getenv
+from typing import Any
+from typing import Dict
+from typing import IO
+from typing import List
+from typing import Optional
+from typing import Tuple
+from typing import Union
+from var_app import GlobalSystemPrompts
+import streamlit as st
+from openai import OpenAI
+
+from dotenv import load_dotenv
+# Charger les variables d'environnement depuis le fichier .env
+
+class DemorrhaAssistant(object):
+    def __init__(self):
+        load_dotenv()
+        self.client = self.initialize_client() # OpenAI(api_key = getenv("OPENAI_API_KEY"))
+        self.vector_store = None
+        self.load_assistant()
+
+    def initialize_client(self):
+        # Initialiser le client OpenAI avec la clé API
+        api_key = getenv("OPENAI_API_KEY")
+        return OpenAI(api_key=api_key)
+
+    def search_assistant(self, assistant_name="Demorrha"):
+        last_id = None
+        while True:
+            # Lister tous les assistants
+            assistants_list = self.client.beta.assistants.list(
+                order="desc",
+                limit="20",
+                after=last_id
+            )
+
+            if len(assistants_list.data) < 1:
+                break
+
+            last_id = assistants_list.data[-1].id
+            for assistant in assistants_list.data:
+                if assistant.name == assistant_name:
+                    return assistant.id
+        return None
+
+    def search_vector_store(self, vector_store_name="Demorrha_Style"):
+        last_id=None
+        while True:
+            # Lister tous les assistants
+            vector_store_list = self.client.beta.vector_stores.list(
+                order="desc",
+                limit="20",
+                after=last_id
+            )
+
+            if len(vector_store_list.data) < 1:
+                break
+            last_id = vector_store_list.data[-1].id
+            for vector_store in vector_store_list.data:
+                if vector_store.name == f"{vector_store_name}":
+                    return vector_store.id
+        return None
+
+    def load_vector_store(self, vector_store_name="Demorrha_Style"):
+        vector_store_id = self.search_vector_store(vector_store_name)
+        if vector_store_id is None:
+            vector_store = self.client.beta.vector_stores.create(name=f"{vector_store_name}")
+            self.vector_store = vector_store
+        else:
+            self.vector_store = self.client.beta.vector_stores.retrieve(vector_store_id)
+        return self
+
+    def get_vector_store(self):
+        return self.vector_store
+
+    def upload_file(self, 
+                    file_path, 
+                    purpose="assistants"):
+        return self.client.files.create(
+            file=open(file_path, "rb"),
+            purpose=purpose
+        )
+
+    def list_files_in_vector_store(self, vector_store_id):
+        files_list = []
+        last_id=None
+        while True:
+            files_list = self.client.beta.vector_stores.files.list(
+                vector_store_id=vector_store_id,
+                limit="20",
+                after=last_id
+            )
+
+            if len(files_list.data) < 1:
+                break
+            last_id = files_list.data[-1].id
+            for file in files_list.data:
+                files_list.append(file)
+        return files_list
+
+    def attach_file_to_vectore_store(self,
+                                     vector_store_id,
+                                     file_id):
+        return self.client.beta.vector_stores.files.create(
+            vector_store_id=vector_store_id,
+            file_id=file_id
+        )
+
+    def load_assistant(self, assistant_name="Demorrha"):
+        system_prompt = GlobalSystemPrompts.linguascribe()
+        assistant_id = self.search_assistant(assistant_name)
+        if assistant_id is None:
+            self.assistant = self.client.beta.assistants.create(
+                model="gpt-4o-mini",
+                name="Demorrha",
+                description="Traite les messages des utilisateurs et génère une traduction.",
+                instructions=f"{system_prompt}",
+                temperature=0.1,
+                tools=[{"type": "file_search"}]
+            )
+        else:
+            self.assistant = self.client.beta.assistants.retrieve(assistant_id)
+
+
+    def add_file_to_vector_store(self, file_paths):
+        file_streams = [open(path, "rb") for path in file_paths]
+        file_batch = self.client.beta.vector_stores.file_batches.upload_and_poll(
+            vector_store_id=self.vector_store.id, files=file_streams
+        )
+        return file_batch
+    
+    def use_assistant(self):
+        # Utiliser l'assistant ici
+        pass
+    
+"""
+
+            file_paths = ["style.txt"]
+            file_streams = [open(path, "rb") for path in file_paths]
+
+            file_batch = client.beta.vector_stores.file_batches.upload_and_poll(
+            vector_store_id=vector_store.id, files=file_streams
+            )
+"""
+
+if __name__ == "__main__":
+    demorrha = DemorrhaAssistant()
+    vector_store = demorrha.load_vector_store().get_vector_store()
+    print(vector_store)
+
+    if vector_store.status == "completed":
+        if vector_store.file_counts.total > 0:
+            if vector_store.file_counts.completed == vector_store.file_counts.total:
+                print("Le chargement du vecteur est terminé.")
+        else:
+            file_paths = ["style.txt"]
+            file_batch = demorrha.add_file_to_vector_store(file_paths)
+            print("Fichier ajouté au vector_store:", file_batch)
+"""
+    # Exemple d'utilisation de l'assistant
+    assistant_id = demorrha.search_assistant("Demorrha")
+    if assistant_id:
+        response = demorrha.use_assistant(assistant_id)
+        print("Réponse de l'assistant:", response)
+    else:
+        print("Assistant non trouvé.")
+"""