Projet-Agent-IA/AgentReact/utils/tools.py

from langchain.tools import tool
from langgraph.prebuilt import InjectedState
from tavily import TavilyClient
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:
    """
    Rechercher une information sur internet

    Args:
        query (str): Terme recherché
    Returns:
        dict: Retour de la recherche
    """
    return TavilyClient().search(query, model='auto')
    

@tool
def write_file(file_path:str, content: str, append:bool=True) -> str:
    """
    Ecrire et ajouter du texte dans un fichier.

    Args:
        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: Le chemin d'accès relatif vers le fichier en cas de réussite, ou une erreur en cas d'echec
    """
    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, todoState: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.
        todoState (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 = todoState # 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.getChroma() # Récupère l'unique instance de cette BDD, c'est un SIngleton

    retrieved_docs = bdd.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, editTodo, read_file, ask_human, search_in_files, addTodo, removeTodo]