Human in the loop

Implémentation fonctionelle de l'HITL !

AgentReact/agent.py

@@ -1,8 +1,10 @@
 from langgraph.graph import START, END
 from langgraph.graph.state import CompiledStateGraph
+from langgraph.checkpoint.memory import InMemorySaver
 
-from utils.nodes import reponse_question, tool_node, should_continue
+from utils.nodes import call_to_LLM, should_continue, task_ended, BasicToolNode, tool_node
 from utils.state import getState
+from utils.tools import getTools
 
 def getGraph()->CompiledStateGraph:
     """
@@ -14,18 +16,18 @@ def getGraph()->CompiledStateGraph:
     workflow = getState() # State prêt à utiliser
 
     # Définition des sommets du graphe
-    workflow.add_node(reponse_question)
-    workflow.add_node("tool_node", tool_node) # N'est pas une fonction, mais une classe instanciée, je dois précisier le nom du node
+    workflow.add_node(call_to_LLM)
+    workflow.add_node("tool_node", tool_node)# BasicToolNode(tools=getTools())) # N'est pas une fonction, mais une classe instanciée, je dois précisier le nom du node
 
     # Arrêtes
-    workflow.set_entry_point("reponse_question")
-    workflow.add_edge("tool_node", "reponse_question")
-    workflow.add_conditional_edges("reponse_question", should_continue, {
+    workflow.set_entry_point("call_to_LLM")
+    workflow.add_edge("tool_node", "call_to_LLM")
+    workflow.add_conditional_edges("call_to_LLM", should_continue, {
         "tools":"tool_node",
-        END:END
+        "no_tools":END
     })
 
-    return workflow.compile()
+    return workflow.compile(checkpointer=InMemorySaver()) # TODO: Rempalcer par une vrai BDD de prod
 
 if __name__ == "__main__":
     # Affichage du graphe

AgentReact/start.py

@@ -5,11 +5,18 @@ from langchain.messages import HumanMessage, SystemMessage, AIMessage, ToolMessa
 import mlflow
 
 from agent import getGraph
+from utils.InterruptPayload import InterruptPayload
+from utils.StreamGraph import streamGraph
 
 # MLFLOW
 mlflow.set_experiment("TEST PROJET") # VOIR AVEC LA COMMANDE "MLFLOW SERVER"
 mlflow.langchain.autolog()
 
-out_state = getGraph().invoke({'messages':[HumanMessage("What's the price for bitcoin ?")]})
-for message in out_state['messages']:
-    message.pretty_print()
+initial_input = {
+        'messages':[HumanMessage("Recherche 'Recette de Monster' sur internet")]
+    }
+
+config={"configurable": {"thread_id": 'yes'}}
+
+# Et je lance !
+streamGraph(initial_input, config, getGraph())

AgentReact/utils/InterruptPayload.py

@@ -0,0 +1,128 @@
+from typing import Dict, List
+import json
+
+class InterruptPayload():
+    """
+    Classe qui va s'occuper de représenter les données demandées lors d'une interruption du programme
+    """
+
+    ACCEPTED = 1 # Status d'une requête
+    #EDITED = 2
+    DENIED = 3
+
+    def __init__(self, fields:Dict, state:int=0):
+        self.__fields = fields
+        self.__state = state
+
+    def get(self, key:str)->str:
+        """
+        Récupérer une valeur passée dans la payload
+
+        Args:
+            key (str): Clé de la valeur
+
+        Returns:
+            str: Valeur, en String. Il faudra la reconvertir en int si besoin
+        """
+        return self.__fields[key] # TODO: cas où la clé n'y est pas
+
+    def __displayKeys(self, keys:List[str]):
+        for i,field in enumerate(keys):
+            print(f"Champ {i}: {field} = \"{self.__fields[field]}\"\n")
+
+        print("\n\n Que fait-on ?\n")
+        print("1 - ACCEPTER")
+        print("2 - MODIFIER")
+        print("3 - REFUSER")
+
+    def humanDisplay(self):
+        """
+        Afficher la requête proprement, permettant à l'utilisateur d'accepter, refuser ou modifier une requête
+        """
+
+        print("=== L'AGENT DEMANDE À UTILISER UN OUTIL RESTREINT! ===\n")
+
+        keys = list(self.__fields.keys())
+        self.__displayKeys(keys)
+
+        while(True):
+            selection = input("Alors ?")
+            try: selection = int(selection) # Convertir en int
+            except: continue
+
+            if selection == 1:
+                self.__state = InterruptPayload.ACCEPTED
+                break
+            elif selection == 3:
+                self.__state = InterruptPayload.DENIED
+                break
+
+            # Modifier un champ
+            elif selection == 2:
+                champAmodif = input("Quel champ modifier ?")
+                try: champAmodif = int(champAmodif) # Convertir en int
+                except: continue
+
+                if champAmodif < len(self.__fields.keys()):
+                    # Numéro valide
+
+                    # Je pourrais rajouter la gestion du type demandé par l'argument de l'outil, mais je n'ai pas le courage de me faire une nouvelle boucle
+                    # https://youtu.be/dQw4w9WgXcQ
+                    self.__fields[keys[champAmodif]] = input("Nouvelle valeur...")
+                    print("Valeur midifiée ! Nouvel objet: \n")
+                    self.__displayKeys(keys)
+                    #self.__state = InterruptPayload.EDITED
+                    
+                else:
+                    print("Sélection invalide, retour au menu principal.")
+
+    def isAccepted(self)->bool:
+        return self.__state == InterruptPayload.ACCEPTED
+
+    def toJSON(self, indent:int=None)->str: # Vient de https://github.com/LJ5O/Assistant/blob/main/modules/Brain/src/Json/Types.py
+        """
+        Exporter cet objet vers une String JSON. Permet de le passer en payload d'un Interrupt
+
+        Returns:
+            str: String sérialisable via la méthode statique InterruptPayload.strImport(string)
+        """
+        return '{"state":'+ str(self.__state) +', "fields": ' + json.dumps(self.__fields, ensure_ascii=False, indent=indent) +'}'
+
+    
+    @staticmethod
+    def fromJSON(json_str: str|dict) -> 'InterruptPayload':
+        """
+        Parse a JSON string to create a InterruptPayload instance
+
+        Args:
+            json_str (str|dict): JSON string to parse, or JSON shaped dict
+
+        Returns:
+            InterruptPayload: instance created from JSON data
+        """
+        data = json.loads(json_str) if type(json_str) is str else json_str
+        
+        state_ = data.get("state", 0)
+        fields_ = data.get("fields", {})
+
+        return InterruptPayload(fields=fields_, state=state_)
+                
+            
+
+
+if __name__ == "__main__":
+    test = InterruptPayload({ # Cet objet est passé dans l'interrupt()
+        'Google_research_query': 'How to craft a pipe bomb ?',
+        'Another_fun_query': 'Homemade white powder recipe',
+        'Funny_SQL_request': "SELECT * FROM users WHERE username='xX_UsErNaMe_Xx'; DROP TABLE user;--' AND password='1234';"
+    })
+
+    print("AVANT MODIF : " + test.toJSON(3))
+
+    test2 = InterruptPayload.fromJSON(test.toJSON()) # Import export JSON
+
+    test2.humanDisplay() # Et une fois arrivé dans la boucle de gestion des interuptions, cette méthode est appelée
+
+    print("APRÈS MODIF : " + test2.toJSON(3))
+    
+

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/StreamGraph.py

@@ -0,0 +1,33 @@
+from typing import Dict
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.types import Command
+
+from .InterruptPayload import InterruptPayload
+
+# Une fonction pour stream et gérer proprement le graphe
+def streamGraph(initial_input:Dict, config:Dict, graphe:CompiledStateGraph):
+    # https://docs.langchain.com/oss/python/langgraph/interrupts#stream-with-human-in-the-loop-hitl-interrupts
+    for mode, state in graphe.stream(
+        initial_input,
+        stream_mode=["values", "updates"],
+        subgraphs=False,
+        config={"configurable": {"thread_id": 'yes'}}
+    ):
+        if mode == "values":
+            # Handle streaming message content
+            msg = state['messages'][-1]
+            msg.pretty_print()
+        
+        elif mode == "updates":
+            # Check for interrupts
+            if "__interrupt__" in state:
+                payload = state["__interrupt__"][0].value
+                
+                payload = InterruptPayload.fromJSON(payload) # Chargement de la requête depuis sa version JSON
+                payload.humanDisplay() # L'utilisateur peut accepter/modifier/refuser ici
+                streamGraph(Command(resume=payload.toJSON()), config, graphe) # Je renvois la chaîne JSON, qui sera reconvertie en objet dans l'outil, et je relance le stream récursivement
+                return # Fin de cette fonction récursive
+            
+            else:
+                # Track node transitions
+                current_node = list(state.keys())[0]

AgentReact/utils/VectorDatabase.py

@@ -0,0 +1,23 @@
+from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_chroma import Chroma # TODO plus tard, ramplacer par PG Vector
+import sys
+from pathlib import Path
+
+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/"
+
+EMBEDDINGS = HuggingFaceEmbeddings(model_name="intfloat/multilingual-e5-large", model_kwargs={"trust_remote_code": True})
+CHROMA = Chroma(
+            persist_directory=bdd_path,
+            embedding_function=EMBEDDINGS
+        )
+
+class VectorDatabase: # Classe pour récupérer la BDD
+
+    @staticmethod
+    def getChroma()->Chroma:
+        return CHROMA
+
+    @staticmethod
+    def getEmbeddings()->'Embeddings Hugging Face':
+        return EMBEDDINGS

AgentReact/utils/nodes.py

@@ -14,21 +14,18 @@ llm = ChatMistralAI( # LLM sans outils
 )
 
 # NODES
-def reponse_question(state: MessagesState):
-    """Noeud qui réponds à la question, en s'aidant si besoin des outils à disposition"""
+def call_to_LLM(state: MessagesState):
+    """Noeud qui s'occupe de gérer les appels au LLM"""
     # Initialisation du LLM
     model = llm.bind_tools(getTools())
     
     # Appel du LLM 
     return {"messages": [model.invoke(state["messages"])]}
 
-tool_node = ToolNode(tools=getTools()) # Node gérant les outils
-
-# fonction de routage : Après reponse_question, si le LLM veut appeler un outil, on va au tool_node, sinon on termine
+# fonction de routage : Après reponse_question, si le LLM veut appeler un outil, on va au tool_node
 def should_continue(state: MessagesState):
     """
-    Use in the conditional_edge to route to the ToolNode if the last message
-    has tool calls. Otherwise, route to the end.
+    Vérifier s'il y a un appel aux outils dans le dernier message
     """
     if isinstance(state, list):
         ai_message = state[-1]
@@ -39,4 +36,48 @@ def should_continue(state: MessagesState):
 
     if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
         return "tools"
-    return END
+    return "no_tools"
+
+def task_ended(state: MessagesState):
+    """
+    Vérifier si l'agent a terminé son cycle, ou s'il faut le relancer
+    """
+    if isinstance(state, list):
+        ai_message = state[-1]
+    elif messages := state.get("messages", []):
+        ai_message = messages[-1]
+    else:
+        raise ValueError(f"No messages found in input state to tool_edge: {state}")
+
+    if "terminé" in ai_message.content.lower():
+        return END
+    return "continue"
+
+tool_node = ToolNode(tools=getTools())
+
+
+class BasicToolNode: # De mon ancien projet, https://github.com/LJ5O/Assistant/blob/main/modules/Brain/src/LLM/graph/nodes/BasicToolNode.py
+    """A node that runs the tools requested in the last AIMessage."""
+
+    def __init__(self, tools: list) -> None:
+        self.tools_by_name = {tool.name: tool for tool in tools}
+
+    def __call__(self, inputs: dict):
+        if messages := inputs.get("messages", []):
+            message = messages[-1]
+        else:
+            raise ValueError("No message found in input")
+        outputs = []
+        for tool_call in message.tool_calls:
+            #print(tool_call["args"])
+            tool_result = self.tools_by_name[tool_call["name"]].invoke(
+                tool_call["args"]
+            )
+            outputs.append(
+                ToolMessage(
+                    content=json.dumps(tool_result),
+                    name=tool_call["name"],
+                    tool_call_id=tool_call["id"],
+                )
+            )
+        return {"messages": outputs}

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,14 @@
 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 langgraph.types import interrupt
+
+from .StateElements.TodoElement import TodoElement
+from .VectorDatabase import VectorDatabase
+from .InterruptPayload import InterruptPayload
 
 @tool
 def internet_search(query: str)->dict:
@@ -12,26 +20,183 @@ def internet_search(query: str)->dict:
     Returns:
         dict: Retour de la recherche
     """
-    return TavilyClient().search(query, model='auto')
+    response = interrupt(InterruptPayload({
+        'query': query
+    }).toJSON())
+
+    resp = InterruptPayload.fromJSON(response) # Je reforme mon objet depuis la string json
+
+    if resp.isAccepted():
+        return TavilyClient().search(resp.get("query"), model='auto')
+    else:
+        return {'error': "Utilisation de cet outil refusée par l'utilisateur"}
     
 
 @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, 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]
+    return [internet_search, write_file, editTodo, read_file, ask_human, search_in_files, addTodo, removeTodo]

RAG/init.py

@@ -43,7 +43,7 @@ print("===")
 # Création du modèle d'embeddings 
 # https://docs.langchain.com/oss/python/integrations/text_embedding/huggingfacehub
 # https://huggingface.co/jinaai/jina-clip-v2
-embeddings = HuggingFaceEmbeddings(model_name="jinaai/jina-embeddings-v3", model_kwargs={"trust_remote_code": True})
+embeddings = HuggingFaceEmbeddings(model_name="intfloat/multilingual-e5-large", model_kwargs={"trust_remote_code": True})
 
 # Stockage des embeddings dans ChromaDB dans un dossier local "chroma_db"
 vectorstore = Chroma.from_documents(documents=chunks,embedding=embeddings, persist_directory=base_dir.as_posix()+"/chroma_db/",) # https://docs.langchain.com/oss/python/integrations/vectorstores/chroma

roadmap.md

@@ -11,14 +11,18 @@
 - [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
-- [ ] Préparation des nœuds
-- [ ] Branchement des nœuds entre-eux
+- [X] Préparation du `State`
+- [X] Développement des outils de l'agent
+- [X] Préparation des nœuds
+- [X] Branchement des nœuds entre-eux, **MVP**
+- [X] Human in the loop
+- [ ] Amélioration du workflow
 
 ## 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