Files
Projet-Agent-IA/AgentReact/utils/InterruptPayload.py
LJ5O 523cea84fe Import Export JSON InterruptPayload
Interrupt a besoin d'un objet sérializable JSON
2026-02-06 21:34:36 +01:00

117 lines
4.0 KiB
Python

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 __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))