Ajustements et tentative de génération du rapport

This commit is contained in:
2026-02-12 11:59:55 +01:00
parent f1d0c7e342
commit 84a27ea6c7
6 changed files with 82 additions and 46 deletions

View File

@@ -5,7 +5,7 @@ from langchain_core.messages import ToolMessage
from langgraph.types import Command
from tavily import TavilyClient
from pathlib import Path
from typing import List, Dict, Annotated
from typing import List, Dict, Annotated, Tuple
import sys
import os
from langgraph.types import interrupt
@@ -25,26 +25,53 @@ def append_part_to_report(contenu:str)->str:
Returns:
str: Retour, une confirmation, ou un message d'erreur
"""
# Récupérer le chemin vers le point d'entrée
base_dir: Path = Path(sys.argv[0]).resolve().parent
full_path = base_dir / "RAPPORT_STAGE.md"
query= interrupt(InterruptPayload({
'content': contenu
}).toJSON())
response = InterruptPayload.fromJSON(query)
if response.isAccepted():
with open(full_path, "a", encoding="utf-8") as f: # Écrire le contenu
f.write("\n"+response.get("content"))
return "Requête acceptée et validée ! Tu peux considérer cette tâche comme complétée."
else:
return "ERREUR! L'utilisateur a refusé ta demande. Tu devrais lui demander pourquoi avoir refusé, et comment améliorer cette partie."
@tool
def list_files(folder:str)->str:
"""
Retrouver la liste des fichiers dans un dossier
Args:
folder (str): Le chemin relatif vers le dossier
Returns:
str: La liste de tous les fichiers dans ce dossier
"""
try:
# Récupérer le chemin vers le point d'entrée
base_dir: Path = Path(sys.argv[0]).resolve().parent
full_path = reports_dir / "RAPPORT_STAGE.md"
full_path: Path = base_dir / folder
query= interrupt(InterruptPayload({
content: contenu
}).toJSON())
if not full_path.exists():
return f"Le dossier '{folder}' n'existe pas."
response = InterruptPayload.fromJSON(query)
if response.isAccepted():
with open(full_path, "a", encoding="utf-8") as f: # Écrire le contenu
f.write(response.get("content"))
if not full_path.is_dir():
return f"Le chemin '{folder}' n'est pas un dossier."
return "Requête acceptée et validée ! Tu peux considérer cette tâche comme complétée."
else:
return "ERREUR! L'utilisateur a refusé ta demande. Tu devrais lui demander pourquoi avoir refusé, et comment améliorer cette partie."
files = [f.name for f in full_path.iterdir()]
if not files:
return f"Le dossier '{folder}' est vide."
return "\n".join(files)
except Exception as e:
return f"Erreur lors de l'écriture: {str(e)}"
return f"Erreur lors de la lecture du dossier : {str(e)}"
@tool
@@ -99,40 +126,21 @@ def editTodo(index:int, todoState:int, state: Annotated[dict, InjectedState], to
})
@tool
def addTodo(name:str, description:str, state: Annotated[dict, InjectedState], tool_call_id: Annotated[str, InjectedToolCallId])->Command:
def setTodo(todoList:List[Tuple[str, str]], state: Annotated[dict, InjectedState], tool_call_id: Annotated[str, InjectedToolCallId])->Command:
"""
Ajouter une nouvelle tâche/TODO
Définir la liste des tâches à faire / TODO.
Permet aussi de la supprimer en appelant avec une liste vide.
Args:
name (str): Nom de cette tâche
description (str): Une ou deux phrases pour décrire le travail à effectuer dans ce TODO
todoList (List[Tuple[str, str]]): Une liste de tuples (str, str), donc le premier str est le nom de la tâche, et le second sa description, le travail à effectuer dans ce TODO
"""
if "todo" not in state.keys(): state["todo"] = []
todo = []
state["todo"] = [TodoElement.fromJSON(e) for e in state["todo"]] # Convertion vers de vraies instances
state["todo"].append(TodoElement(name, description))
for t in todoList:
todo.append(TodoElement(t[0], t[1]))
return Command(update={
"messages": [ToolMessage(content="Réussite!", tool_call_id=tool_call_id)],
"todo": [x.toJSON() for x in state["todo"]] # Update du state, # medium.com/@o39joey/a-comprehensive-guide-to-langgraph-managing-agent-state-with-tools-ae932206c7d7
})
@tool
def removeTodo(index:int, state: Annotated[dict, InjectedState], tool_call_id: Annotated[str, InjectedToolCallId])->Command:
"""
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
"""
if "todo" not in state.keys(): return Command(update={"messages": [ToolMessage(content="Echec!", tool_call_id=tool_call_id)]})
if len(state["todo"]) <= index:
# Erreur, l'index est trop grand
return Command(update={"messages": [ToolMessage(content="Index en dehors de la liste, echec!", tool_call_id=tool_call_id)]})
state['todo'].pop(index)
return Command(update={
"messages": [ToolMessage(content="Réussite!", tool_call_id=tool_call_id)],
"todo": [x for x in state["todo"]] # Update du state, # medium.com/@o39joey/a-comprehensive-guide-to-langgraph-managing-agent-state-with-tools-ae932206c7d7
"todo": [x.toJSON() for x in todo] # Update du state, # medium.com/@o39joey/a-comprehensive-guide-to-langgraph-managing-agent-state-with-tools-ae932206c7d7
})
@tool
@@ -284,7 +292,7 @@ def getTools()->List['Tools']:
"""
Récupérer la liste des tools
"""
return [internet_search, append_part_to_report, editTodo, read_file, search_in_files, addTodo, removeTodo, get_skill]
return [internet_search, append_part_to_report, read_file, search_in_files, get_skill, list_files] # editTodo, setTodo
def getWeeklyReportTools()->List['Tools']:
"""