Fix tools

Quelques petits fixs

AgentReact/utils/StateElements/TodoElement.py

@@ -0,0 +1,40 @@
+# Classes utilisées pour représenter des données
+class TodoElement():
+    STATE_NOT_STARTED = 0 # Sorte d'enum qui représente l'état d'une tâche
+    STATE_STARTED = 1
+    STATE_COMPLETED = 2
+
+    name: str
+    state: int
+
+    def __init__(self, name:str, description:str=None):
+        self.name = name
+        self.description = description
+        self.state = TodoElement.STATE_NOT_STARTED
+    
+    def __str__(self)->str:
+        """
+        Affiche la tâche, son nom et son statut
+        Affichera aussi la description de la tâche si elle a été définie ET est en cours
+
+        Returns:
+            str: Représentation écrite de la tâche
+        """
+        return f"Tâche \"{self.name}\": {self.__getStateName()}." + \
+            (f" Description: {self.description}" if self.description and self.state == TodoElement.STATE_STARTED else '')
+
+    def __getStateName(self)->str:
+        if self.state == TodoElement.STATE_NOT_STARTED:
+            return "Non commencée"
+        elif self.state == TodoElement.STATE_STARTED:
+            return "En cours"
+        elif self.state == TodoElement.STATE_COMPLETED:
+            return "Terminée"
+        else:
+            return "Inconnu"
+
+if __name__ == "__main__":
+    test = TodoElement("TEST tâche", "OUI")
+    test.state = TodoElement.STATE_STARTED
+    print(test)
+    print([str(test)])

AgentReact/utils/VectorDatabase.py

@@ -0,0 +1,45 @@
+from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_chroma import Chroma # TODO plus tard, ramplacer par PG Vector
+import sys
+from pathlib import Path
+
+# Permet de garder ChromaDB en mémoire.
+# Cette classe est un Singleton, il n'y en aura qu'une seule et unique instance à tout moment
+# https://refactoring.guru/design-patterns/singleton
+class VectorDatabase:
+    instance = None
+
+    def __new__(cls): # Selon https://www.geeksforgeeks.org/python/singleton-pattern-in-python-a-complete-guide/
+        if cls.instance is None:
+            cls.instance = super().__new__(cls)
+            # J'initialise les attributs à None ici, permet de tester si la classe a déjà été init une première fois ou non
+            cls.instance.__embeddings = None
+            cls.instance.__chroma = None
+        return cls.instance
+
+    def __init__(self):
+        if self.__embeddings is not None: return
+
+        base_dir:str = Path(sys.argv[0]).resolve().parent.as_posix() # Récupérer le chemin vers le point d'entrée du programme
+        bdd_path:str = base_dir + "/chroma_db/"
+
+        self.__embeddings = HuggingFaceEmbeddings(model_name="jinaai/jina-embeddings-v3", model_kwargs={"trust_remote_code": True})
+        self.__chroma = Chroma(
+            persist_directory=bdd_path,
+            embedding_function=self.__embeddings
+        )
+
+    def getChroma(self)->Chroma:
+        return self.__chroma
+
+    def getEmbeddings(self)->'Embeddings Hugging Face':
+        return self.__embeddings
+
+if __name__ == "__main__":
+
+    test1 = VectorDatabase()
+    print('TEST 1 INIT')
+    test2 = VectorDatabase()
+
+    print(test1 is test2)
+    assert test1 is test2

AgentReact/utils/state.py

@@ -1,7 +1,17 @@
 from langgraph.graph import StateGraph, MessagesState
+from typing import List
+
+from .StateElements.TodoElement import TodoElement
+
+
+class CustomState(MessagesState):
+    todo: List[TodoElement] # Les tâches en cours
+
+    ragQuery: str # Requête envoyée au RAG, pour le cross-encodeur
+    ragDocuments: List[str] # Documents retrouvés par le RAG, pour le cross-encodeur
+
+    # TODO: Ajouter la source des documents sélectionnés pour la fin du rapport ?
 
-class hjgzefvuiyguhzfvihuozdef(MessagesState): # J'ai du mal à nommer mes classes ._.
-    pass
 
 def getState()->StateGraph:
     """
@@ -10,4 +20,4 @@ def getState()->StateGraph:
     Returns:
         StateGraph: prêt à utiliser
     """
-    return StateGraph(hjgzefvuiyguhzfvihuozdef)
+    return StateGraph(CustomState)

AgentReact/utils/tools.py

@@ -1,6 +1,12 @@
 from langchain.tools import tool
+from langgraph.prebuilt import InjectedState
 from tavily import TavilyClient
-from typing import List
+from pathlib import Path
+from typing import List, Dict, Annotated
+import sys
+from .StateElements.TodoElement import TodoElement
+
+from .VectorDatabase import VectorDatabase
 
 @tool
 def internet_search(query: str)->dict:
@@ -16,22 +22,170 @@ def internet_search(query: str)->dict:
     
 
 @tool
-def write_file(content: str) -> str:
+def write_file(file_path:str, content: str, append:bool=True) -> str:
     """
-    Écrire les données dans un fichier
+    Ecrire et ajouter du texte dans un fichier.
 
     Args:
-        content (str): Contenu du fichier à écrire
+        file_path (str): Chemin d'accès relatif vers le fichier à écrire.
+        content (str): Contenu à écrire dans le fichier.
+        append (bool, optional): Faut-il AJOUTER(True) au fichier, ou REMPLACER son contenu(False) ? True par défaut.
 
     Returns:
-        str: Résultat de l'écriture
+        str: Le chemin d'accès relatif vers le fichier en cas de réussite, ou une erreur en cas d'echec
     """
-    print("==ECRITURE FICHIER==")
-    print(content)
-    return "Fichier écrit"
+    try:
+        base_dir:str = Path(sys.argv[0]).resolve().parent.as_posix() # Récupérer le chemin vers le point d'entrée du programme
+        full_path:str = base_dir + (file_path if file_path.startswith('/') else f'/{file_path}') # Puis générer le chemin vers le fichier
+
+        mode = "a" if append else "w" # Mode d'écriture
+        with open(full_path, mode, encoding="utf-8") as f: # Puis j'écris
+                f.write(content)
+
+        return str(full_path)
+
+    except Exception as e:
+        return f"Erreur lors de l'écriture: {str(e)}"
+
+@tool
+def editTodo(index:int, state:int, state: Annotated[dict, InjectedState])->bool: # https://stackoverflow.com/a/79525434
+    """
+    Modifier l'état d'une tâche (TODO)
+
+    Args:
+        index (int): Index de la tâche à modifier, en commançant à 0 pour la première tâche.
+        state (int): Nouvel état. 0 pour "non commencé, 1 pour "en cours", 2 pour "complété"
+
+    Returns:
+        bool: Réussite de l'opération, ou non.
+    """
+    if len(state["todo"]) <= index:
+        # Erreur, l'index est trop grand
+        return False
+
+    state["todo"][index].state = state # Modification de l'état de cette tâche
+
+    # Toutes les tâches complétées ?
+    found = False
+    for task in state["todo"]: # Pour chaque tâche
+        if task.state != 2: # Si elle n'est pas terminée
+            found = True
+            break
+
+    if not found: state["todo"] = [] # Toutes les tâches terminées, je peux clear la TODO list du state
+    return True
+
+@tool
+def addTodo(name:str, description:str, state: Annotated[dict, InjectedState])->bool:
+    """
+    Ajouter une nouvelle tâche/TODO
+
+    Args:
+        name (str): Nom de cette tâche
+        description (str): Une ou deux phrases pour décrire le travail à effectuer dans ce TODO
+
+    Returns:
+        bool: Réussite de l'opération, ou non
+    """
+    if state["todo"] is None: state["todo"] = []
+
+    state["todo"].append(TodoElement(name, description))
+    return True
+
+@tool
+def removeTodo(index:int, state: Annotated[dict, InjectedState])->bool:
+    """
+    Retirer une tâche/TODO de la liste des tâches
+
+    Args:
+        index (int): Position de la tâche dans la liste, commence à 0 pour le premier TODO
+
+    Returns:
+        bool: Réussite de l'opération, ou non
+    """
+    if len(state["todo"]) <= index:
+        # Erreur, l'index est trop grand
+        return False
+
+    state['todo'].pop(index)
+    return True
+
+@tool
+def read_file(file_path: str) -> str:
+    """
+    Lire le contenu d'un fichier texte.
+
+    Args:
+        file_path (str): Chemin d'accès relatif vers le fichier à lire.
+
+    Returns:
+        str: Le contenu du fichier, ou un message d'erreur.
+    """
+    try:
+        base_dir:str = Path(sys.argv[0]).resolve().parent.as_posix() # Récupérer le chemin vers le point d'entrée du programme
+        full_path:str = base_dir + (file_path if file_path.startswith('/') else f'/{file_path}') # Puis générer le chemin vers le fichier
+
+        with open(full_path, "r", encoding="utf-8") as f:
+            content = f.read()
+
+        return content
+
+    except Exception as e:
+        return f"Erreur lors de la lecture : {str(e)}"
+
+@tool
+def ask_human(request:str)->str:
+    """
+    Demander quelque chose à un assistant humain. Permet d'obtenir des informations supplémentaires,
+    ou qu'une action soit réalisée.
+
+    Args:
+        request (str): Ce qui est demandé à l'humain
+
+    Returns:
+        str: Réponse de l'humain
+    """
+    print("--- L'IA A BESOIN D'UN HUMAIN ! ---")
+    print(f"L'IA demande : {request}")
+    
+    user_response = input("Réponse humaine: ") # Input bloque le système en attendant l'humain
+    # J'aurais possiblement utiliser d'autres approches comme https://docs.langchain.com/oss/javascript/langchain/human-in-the-loop
+    # Mais Human in the loop se place AVANT l'outil. Ici, l'outil consiste justement à demander quelque chose à un humain.
+    
+    print("-------")
+    
+    return user_response
+
+@tool
+def search_in_files(query:str, state: Annotated[dict, InjectedState])->str:
+    """
+    Rechercher quelque chose dans les documents enregistrés localement.
+    Dans le cas actuel, ces documents sont des rapports hebdomadaires de stage.
+
+    Args:
+        query (str): La requête recherchée.
+
+    Returns:
+        str: Échantillons de documents correspondants, concaténés en une seule chaîne de caractères.
+    """
+    bdd = VectorDatabase() # Récupère l'unique instance de cette BDD, c'est un SIngleton
+
+    retrieved_docs = bdd.getChroma().similarity_search(query, k=5) # 5 documents
+
+    # Conversion des documents en texte
+    docs_content = "\n".join(
+        [f"Document {i+1}:\n{doc.page_content}" for i,doc in enumerate(retrieved_docs)]
+    )
+
+    # Sauvegarde des données dans le State
+    state["ragQuery"] = query
+    state["ragDocuments"] = retrieved_docs
+
+    return docs_content # Retourne la liste de documents trouvés
+
 
 def getTools()->List['Tools']:
     """
     Récupérer la liste des tools
     """
-    return [internet_search, write_file]
+    return [internet_search, write_file, editTodo, read_file, ask_human, search_in_files, addTodo, removeTodo]

roadmap.md

@@ -11,14 +11,16 @@
 - [X] Lecture des documents et mise en base de données vectorielle
 
 ## Mise en place de l'agent
-- [ ] Préparation du `State`
-- [ ] Développement des outils de l'agent
+- [X] Préparation du `State`
+- [X] Développement des outils de l'agent
 - [ ] Préparation des nœuds
 - [ ] Branchement des nœuds entre-eux
 
 ## Amélioration de l'agent
+- [ ] Cross-encoding sur la sortie du **RAG**
 - [ ] Sauvegarde de l'état de l'agent
 - [ ] Système de redémarrage après un arrêt
+- [ ] Gestion de la taille du contexte - Résumé de l'historique des messages
 - [ ] Détection de *prompt injection*
 - [ ] Génération d'un PDF en sortie du système