191 lines
6.1 KiB
Python
191 lines
6.1 KiB
Python
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, 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, editTodo, read_file, ask_human, search_in_files, addTodo, removeTodo] |