IA requête API dynamique

बनाया गया: 18 जनवरी 2025

का उपयोग करके उत्तर दिया गया OpenAI 01 द्वारा OpenAI01

प्रश्न

fait un code qui fait en sorte que une IA puisse utilisé un endpoint est decouvrir quelle api il faut utilisé en fonction du message et quelle génére aussi une reponse en fonction de cela :
import sys
import json
import os
import requests
from groq import Groq

# ---------------------------------------------------------------------------
# Constantes / Variables globales
FICHIER_APIS = "apis_configurees.json"
MAX_CHARS_DIRECT_ANALYSIS = 3000
DEFAULT_GROQ_API_KEY = "gsk_votre_cle_par_defaut_ici" # Adapter si besoin

class ApiManager:
"""
Classe regroupant toutes les méthodes nécessaires :
- Lecture/écriture des APIs (JSON)
- Intéraction avec l'IA Groq pour résumé/analyse JSON
- Menu et actions (ajout, modification, test, etc.)
"""

# Liste globale des APIs configurées
apis_configurees = []

# -----------------------------------------------------------------------
# 1. Fonctions d'IO pour sauvegarder/charger les APIs
@staticmethod
def lire_int_in_range(prompt: str, min_val: int, max_val: int) -> int:
"""
Demande à l'utilisateur de saisir un entier dans [min_val..max_val].
Gère également le cas où l'utilisateur saisit '?' pour demander de l'aide.
Retourne l'entier validé.
"""
while True:
saisie = input(prompt).strip()

if saisie == "?":
print("Aide : veuillez saisir un entier compris entre "
f"{min_val} et {max_val}. Tapez Ctrl+C pour annuler.")
continue

try:
val = int(saisie)
if min_val <= val <= max_val:
return val
else:
print(f"Choix invalide (doit être entre {min_val} et {max_val}).\n")
except ValueError:
print("Choix invalide (entrez un nombre). Réessayez.\n")

@staticmethod
def sauvegarder_apis() -> None:
"""
Sauvegarde la liste des APIs dans FICHIER_APIS au format JSON.
"""
try:
with open(FICHIER_APIS, "w", encoding="utf-8") as f:
json.dump(ApiManager.apis_configurees, f, indent=2, ensure_ascii=False)
print(f"[INFO] Liste des APIs sauvegardée dans {FICHIER_APIS}.")
except Exception as e:
print(f"[ERREUR] Impossible de sauvegarder dans {FICHIER_APIS} : {e}")

@staticmethod
def charger_apis() -> None:
"""
Charge la liste des APIs depuis FICHIER_APIS, si présent.
"""
if os.path.exists(FICHIER_APIS):
try:
with open(FICHIER_APIS, "r", encoding="utf-8") as f:
ApiManager.apis_configurees = json.load(f)
print(f"[INFO] {len(ApiManager.apis_configurees)} API(s) chargée(s) depuis {FICHIER_APIS}.")
except Exception as e:
print(f"[ERREUR] Impossible de lire {FICHIER_APIS} : {e}")
ApiManager.apis_configurees = []
else:
ApiManager.apis_configurees = []

# -----------------------------------------------------------------------
# 2. Fonction de résumé si la doc est trop volumineuse
@staticmethod
def summarizer_groq(long_text: str, groq_api_key: str = DEFAULT_GROQ_API_KEY) -> str:
"""
Découpe la documentation en chunks et génère un résumé global (texte brut).
On utilise le modèle Chat via l'API Groq.
"""
client = Groq(api_key=groq_api_key)

chunk_size = 3000
chunks = []
start = 0
while start < len(long_text):
end = start + chunk_size
chunks.append(long_text[start:end])
start = end

summaries = []
for i, chunk_text in enumerate(chunks, start=1):
print(f"[INFO] Résumé du chunk {i}/{len(chunks)}...")

messages = [
{
"role": "system",
"content": (
"Tu es un assistant chargé de résumer la documentation d'API. "
"Retourne un résumé concis (en langage courant) du texte fourni. "
"Ne génère pas de JSON, pas de code, juste un texte résumé clair."
)
},
{
"role": "user",
"content": chunk_text
}
]

try:
completion = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=messages,
temperature=0.2,
top_p=1,
max_completion_tokens=600, # Par exemple pour limiter
stream=False
)
summary_text = completion.choices[0].message.content.strip()
summaries.append(summary_text)
except Exception as e:
print(f"[ERREUR] Impossible de résumer le chunk {i} : {e}")
summaries.append("")

combined_summary = "\n".join(summaries)
print(f"[INFO] Résumé global généré. Longueur du résumé : {len(combined_summary)} caractères.")
return combined_summary

# -----------------------------------------------------------------------
# 3. Fonction d'analyse finale en JSON
@staticmethod
def analyser_api_avec_groq(input_text: str, groq_api_key: str = DEFAULT_GROQ_API_KEY) -> dict:
"""
1) Si le texte est trop long, on génère un résumé via summarizer_groq.
2) On passe ensuite (ou directement) le texte final dans un prompt
qui demande un JSON strict.
3) Retourne un dict Python.
"""
# 1) Résume si trop volumineux
text_for_analysis = input_text
if len(input_text) > MAX_CHARS_DIRECT_ANALYSIS:
print("[INFO] La documentation est trop volumineuse, on génère un résumé d’abord.")
text_for_analysis = ApiManager.summarizer_groq(input_text, groq_api_key=groq_api_key)

# 2) Appel Chat Completion pour le JSON
client = Groq(api_key=groq_api_key)

messages = [
{
"role": "system",
"content": (
"Tu es un assistant d'analyse d'API. Ta réponse doit être EXCLUSIVEMENT du JSON valide. "
"Respecte ce schéma complet :\n\n"
"{\n"
" \"api_type\": \"REST | GraphQL | SOAP\",\n"
" \"base_url\": \"string\",\n"
" \"auth_info\": {\n"
" \"type\": \"string (Bearer, Basic, etc.)\",\n"
" \"key_name\": \"string\",\n"
" \"required\": true/false\n"
" },\n"
" \"common_headers\": {\n"
" \"Header-Name\": \"Header-Value\"\n"
" },\n"
" \"endpoints\": [\n"
" {\n"
" \"name\": \"string\",\n"
" \"path\": \"string\",\n"
" \"method\": \"GET | POST | ...\",\n"
" \"required_params\": [\"param1\", \"param2\"],\n"
" \"optional_params\": [\"param1\", \"param2\"],\n"
" \"description\": \"string\"\n"
" }\n"
" ],\n"
" \"suggested_name\": \"string\",\n"
" \"rate_limit\": \"string\",\n"
" \"confidence_score\": 0.0\n"
"}\n\n"
"Ne fournis aucun texte additionnel. Juste le JSON."
)
},
{
"role": "user",
"content": text_for_analysis
}
]

try:
chat_completion = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=messages,
temperature=0.0,
top_p=1,
max_completion_tokens=1500, # Ajuster selon la taille souhaitée
stream=False
)
content = chat_completion.choices[0].message.content.strip()

# Tentative de parsing JSON
data = json.loads(content)
return data

except Exception as e:
print(f"[ERREUR] Échec de la conversion en JSON : {e}")
print("[INFO] Retour d’un squelette par défaut.")
return {
"api_type": None,
"base_url": None,
"auth_info": {
"type": None,
"key_name": None,
"required": False
},
"common_headers": {},
"endpoints": [],
"suggested_name": "API_Inconnue",
"rate_limit": None,
"confidence_score": 0.0
}

# -----------------------------------------------------------------------
# 4. FONCTIONS D’OUTIL POUR L’INTERFACE CLI
@staticmethod
def selectionner_api() -> dict | None:
"""
Affiche la liste des API configurées, demande un choix, retourne le dict de l'API choisie.
Retourne None s'il n'y en a aucune.
"""
if not ApiManager.apis_configurees:
print("Aucune API n’est configurée.\n")
return None

print("=== Liste des APIs disponibles ===")
for i, api_info in enumerate(ApiManager.apis_configurees, start=1):
print(f"{i}. {api_info['name']}")

choix_num = ApiManager.lire_int_in_range(
prompt=f"Quelle API souhaitez-vous sélectionner ? [1..{len(ApiManager.apis_configurees)}] : ",
min_val=1,
max_val=len(ApiManager.apis_configurees)
)
return ApiManager.apis_configurees[choix_num - 1]

@staticmethod
def selectionner_endpoint(api_info: dict) -> dict | None:
"""
Affiche la liste des endpoints, demande un choix, retourne le dict du endpoint choisi.
Retourne None s'il n'y en a aucun.
"""
if not api_info["endpoints"]:
print("Cette API n’a aucun endpoint configuré.\n")
return None

print(f"Endpoints disponibles pour {api_info['name']} :")
for i, ep in enumerate(api_info["endpoints"], start=1):
print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})")

choix_ep_num = ApiManager.lire_int_in_range(
prompt=f"Quel endpoint tester ? [1..{len(api_info['endpoints'])}] : ",
min_val=1,
max_val=len(api_info["endpoints"])
)
return api_info["endpoints"][choix_ep_num - 1]

# -----------------------------------------------------------------------
# 5. Fonctions du menu
@staticmethod
def ajouter_api() -> None:
"""
Ajoute une nouvelle API en collant sa documentation (utilise l'IA Groq pour l'analyser).
"""
print("=== Ajout d’une nouvelle API ===")
print("Collez la documentation ci-dessous (tapez 'skibidi' sur une ligne vide pour finir) :")

lignes_doc = []
while True:
ligne = input()
if ligne.strip() == "skibidi":
break
lignes_doc.append(ligne)
doc_api = "\n".join(lignes_doc).strip()
if not doc_api:
doc_api = "Pas d'information sur l'API."

# Analyse par IA
resultats_analyse = ApiManager.analyser_api_avec_groq(doc_api)

confidence_score = resultats_analyse.get("confidence_score", 0.0)
if confidence_score < 0.5:
print(
"Le score de confiance de l’analyse est faible. "
"Les informations retournées peuvent être partielles ou incorrectes.\n"
)

# Champs retournés
api_type = resultats_analyse.get("api_type")
base_url = resultats_analyse.get("base_url")
auth_info = resultats_analyse.get("auth_info", {})
common_headers = resultats_analyse.get("common_headers", {})
endpoints = resultats_analyse.get("endpoints", [])
suggested_name = resultats_analyse.get("suggested_name", "NouvelleAPI")
rate_limit = resultats_analyse.get("rate_limit")

print(f"Nom suggéré pour l’API : {suggested_name}")
nom_api = input(f"Entrez le nom de l’API ({suggested_name}) : ").strip()
if not nom_api:
nom_api = suggested_name

# Si l'API nécessite une clé/token
api_key_value = None
if auth_info.get("required", False):
api_key_value = input("Entrez la valeur de la clé API ou du token : ")

# Construction de la config
nouvelle_api = {
"name": nom_api,
"type": api_type,
"base_url": base_url,
"auth_info": auth_info,
"api_key_value": api_key_value,
"common_headers": common_headers,
"endpoints": endpoints,
"rate_limit": rate_limit
}

ApiManager.apis_configurees.append(nouvelle_api)
print(f"\n[OK] L’API '{nom_api}' a été ajoutée.")
print(f" - Type : {api_type}")
print(f" - URL de base : {base_url}\n")

# Sauvegarde
ApiManager.sauvegarder_apis()

@staticmethod
def lister_apis() -> None:
"""
Affiche toutes les APIs configurées (avec leurs endpoints).
"""
print("=== Liste des API configurées ===")
if not ApiManager.apis_configurees:
print("Aucune API n’est configurée.\n")
return

for i, api_info in enumerate(ApiManager.apis_configurees, start=1):
print(f"{i}. Nom : {api_info['name']}")
print(f" URL de base : {api_info['base_url'] or 'Non renseignée'}")
print(f" Type : {api_info['type'] or 'Inconnu'}")
if api_info["rate_limit"]:
print(f" Limitation de débit : {api_info['rate_limit']}")
if api_info["endpoints"]:
print(" Endpoints :")
for ep in api_info["endpoints"]:
print(f" - {ep['name']} ({ep['method']} {ep['path']})")
else:
print(" Aucun endpoint configuré.")
print()
print()

@staticmethod
def modifier_api() -> None:
"""
Permet de modifier diverses infos sur une API existante : URL de base, clé, endpoints, etc.
"""
print("=== Modification d’une API ===")
api_choisie = ApiManager.selectionner_api()
if not api_choisie:
return

print(f"\nAPI sélectionnée : {api_choisie['name']}")
print("Que voulez-vous modifier ? (ex: 'changer l'URL de base', 'changer la clé API', "
"'ajouter un endpoint', 'modifier un endpoint', 'supprimer un endpoint')")
action = input("> ").lower()

if "url" in action:
new_url = input("Nouvelle URL de base : ")
api_choisie["base_url"] = new_url
print("URL de base modifiée.\n")

if "clé" in action or "key" in action:
new_key = input("Nouvelle clé API : ")
api_choisie["api_key_value"] = new_key
print("Clé API modifiée.\n")

if "ajouter un endpoint" in action:
print("=== Ajout d'un endpoint ===")
nom_ep = input("Nom du nouvel endpoint : ")
path_ep = input("Chemin (ex: /playlists/{playlist_id}/tracks) : ")
methode = input("Méthode HTTP (GET, POST, etc.) : ")
params_req = [p.strip() for p in input("Paramètres requis (séparés par des virgules) : ").split(",") if p.strip()]
params_opt = [p.strip() for p in input("Paramètres optionnels (séparés par des virgules) : ").split(",") if p.strip()]

new_ep = {
"name": nom_ep,
"path": path_ep,
"method": methode.upper(),
"required_params": params_req,
"optional_params": params_opt,
"description": ""
}
api_choisie["endpoints"].append(new_ep)
print(f"Endpoint '{nom_ep}' ajouté.\n")

if "modifier un endpoint" in action:
print("=== Modification d'un endpoint ===")
if not api_choisie["endpoints"]:
print("Aucun endpoint à modifier.")
return

for i, ep in enumerate(api_choisie["endpoints"], start=1):
print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})")

choix_ep = ApiManager.lire_int_in_range(
f"Quel endpoint modifier ? [1..{len(api_choisie['endpoints'])}] : ",
1,
len(api_choisie["endpoints"])
)
ep_choisi = api_choisie["endpoints"][choix_ep - 1]

print(f"Endpoint sélectionné : {ep_choisi['name']}")
modif_ep = input("Que voulez-vous faire ? (ex: 'ajouter un paramètre', 'changer la méthode', etc.) : ").lower()
if "ajouter un paramètre" in modif_ep:
nom_p = input("Nom du paramètre : ")
r_o = input("Paramètre requis ou optionnel ? (r/o) : ")
if r_o.lower() == 'r':
ep_choisi["required_params"].append(nom_p)
else:
ep_choisi["optional_params"].append(nom_p)
print(f"Paramètre '{nom_p}' ajouté.\n")

# D'autres modifications possibles, à adapter selon vos besoins...

if "supprimer un endpoint" in action:
print("=== Suppression d'un endpoint ===")
if not api_choisie["endpoints"]:
print("Aucun endpoint à supprimer.")
return

for i, ep in enumerate(api_choisie["endpoints"], start=1):
print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})")

choix_ep = ApiManager.lire_int_in_range(
f"Quel endpoint supprimer ? [1..{len(api_choisie['endpoints'])}] : ",
1,
len(api_choisie["endpoints"])
)
ep_asup = api_choisie["endpoints"][choix_ep - 1]
api_choisie["endpoints"].remove(ep_asup)
print(f"Endpoint '{ep_asup['name']}' supprimé.")

print("Modification(s) terminée(s).\n")
ApiManager.sauvegarder_apis()

@staticmethod
def supprimer_api() -> None:
"""
Supprime complètement une API de la liste (après confirmation).
"""
print("=== Suppression d’une API ===")
api_choisie = ApiManager.selectionner_api()
if not api_choisie:
return

confirm = input(f"Êtes-vous sûr de vouloir supprimer '{api_choisie['name']}' ? (o/n) : ").lower()
if confirm == 'o':
ApiManager.apis_configurees.remove(api_choisie)
print(f"L’API '{api_choisie['name']}' a été supprimée.\n")
ApiManager.sauvegarder_apis()
else:
print("Opération annulée.\n")

@staticmethod
def tester_endpoint() -> None:
"""
Permet de tester un endpoint en temps réel :
- Sélectionne une API
- Sélectionne un endpoint
- Demande les paramètres requis/optionnels
- Lance la requête HTTP et affiche la réponse.
"""
print("=== Tester un endpoint ===")
api_choisie = ApiManager.selectionner_api()
if not api_choisie:
return

ep_choisi = ApiManager.selectionner_endpoint(api_choisie)
if not ep_choisi:
return

# Paramètres requis
required_params_values = {}
for rp in ep_choisi.get("required_params", []):
val = input(f"Entrez la valeur pour '{rp}' : ")
required_params_values[rp] = val

# Paramètres optionnels
optional_params_values = {}
for op in ep_choisi.get("optional_params", []):
val = input(f"Valeur pour '{op}' (laisser vide si aucun) : ")
if val:
optional_params_values[op] = val

# Préparation de l’URL
base_url = api_choisie.get("base_url") or ""
path = ep_choisi["path"]

# Remplacement des placeholders {param} dans le path
query_params = {}
for k, v in required_params_values.items():
placeholder = f"{{{k}}}"
if placeholder in path:
path = path.replace(placeholder, v)
else:
query_params[k] = v

# Les optionnels vont en query params
query_params.update(optional_params_values)

# Gestion de l’authentification
headers = dict(api_choisie.get("common_headers", {}))
if "Accept" not in headers:
headers["Accept"] = "application/json"

auth_info = api_choisie.get("auth_info", {})
api_key_value = api_choisie.get("api_key_value")

if auth_info.get("required") and api_key_value:
# Exemple : si c’est un Bearer
if auth_info["type"] and "bearer" in auth_info["type"].lower():
key_name = auth_info.get("key_name", "Authorization")
headers[key_name] = f"Bearer {api_key_value}"
# Exemple : si c’est un Basic
elif auth_info["type"] and "basic" in auth_info["type"].lower():
key_name = auth_info.get("key_name", "Authorization")
headers[key_name] = f"Basic {api_key_value}"
else:
# Parfois, c'est un param "api_key" dans l'URL
if auth_info.get("key_name", "").lower() == "api_key":
query_params["api_key"] = api_key_value
else:
# Sinon, on le place dans un header
key_name = auth_info.get("key_name", "X-API-Key")
headers[key_name] = api_key_value

# Construction de l'URL finale
url = base_url + path
method = ep_choisi["method"].upper()

print(f"\nTest de l'endpoint : {method} {url}")
print(f"En-têtes : {headers}")
print(f"Paramètres (query/json) : {query_params}\n")

try:
if method == "GET":
r = requests.get(url, headers=headers, params=query_params)
elif method == "POST":
r = requests.post(url, headers=headers, json=query_params)
elif method == "PUT":
r = requests.put(url, headers=headers, json=query_params)
elif method == "DELETE":
r = requests.delete(url, headers=headers, json=query_params)
else:
print("Méthode non gérée.")
return

print(f"Statut : {r.status_code}")
try:
print("Réponse JSON :")
print(json.dumps(r.json(), indent=2, ensure_ascii=False))
except Exception:
print("Réponse brute :")
print(r.text)

except Exception as e:
print(f"Erreur lors de l’appel : {e}")

# -----------------------------------------------------------------------
# 6. MAIN
@staticmethod
def main() -> None:
"""
Fonction principale de l'application (menu en boucle).
"""
# On charge la liste des APIs au démarrage
ApiManager.charger_apis()

while True:
print("=== Menu Principal ===")
print("1. Ajouter une API (assisté par IA)")
print("2. Liste des API configurées")
print("3. Modifier une API")
print("4. Supprimer une API")
print("5. Tester un endpoint")
print("6. Quitter")

choix_num = ApiManager.lire_int_in_range("Choisissez une option [1..6] : ", 1, 6)

if choix_num == 1:
ApiManager.ajouter_api()
elif choix_num == 2:
ApiManager.lister_apis()
elif choix_num == 3:
ApiManager.modifier_api()
elif choix_num == 4:
ApiManager.supprimer_api()
elif choix_num == 5:
ApiManager.tester_endpoint()
elif choix_num == 6:
print("Au revoir.")
sys.exit(0)

# ---------------------------------------------------------------------------
# Point d'entrée du script
if __name__ == "__main__":
ApiManager.main()

विचार

Uncovering API insights

I’m thinking through how an AI can read a user message, pinpoint the relevant API, call it, and generate a fitting response, all while analyzing APIs from JSON and utilizing a custom summarizer.

Integrating functions

I'm considering adding a feature to read user messages, identify the correct endpoint, call it, and generate a response. This might be a standalone function or integrated into "tester_endpoint."

Crafting the function

I'm detailing "gerer_message_ia" to handle user messages by parsing with GPT or Groq, identifying the API endpoint, calling it, and generating a response in French.

Crafting the ideal function

I'm refining "gerer_message_ia" to integrate with existing code, leveraging "analyser_api_avec_groq" to pinpoint the most suitable endpoint, and efficiently generate a tailored response.

Crafting a solution

I'm mapping out "interpreter_message_et_appeler_api" or "gerer_message_utilisateur" to efficiently identify the relevant API, call the endpoint, and provide a concise response, ensuring a seamless user experience.

Ceating a function

I’m developing "gerer_message_ia" to gather API details, generate a JSON request, call the API, and return a concise French response, ensuring a robust and direct approach.

Formulating the function

I’m piecing together "gerer_requete_naturelle" to align with user needs, integrating with the existing code and identifying the most suitable API based on the message to generate a response in French.

Mapping the process

I’m detailing the steps: parsing the user message to identify the API and endpoint, calling it, and crafting a concise response, all while maintaining consistency with the existing code's "Groq" usage.

Crafting the function

I'm developing a final code snippet in French to integrate "gerer_requete_naturelle," ensuring it enriches or complements existing API management functions like "tester_endpoint" or "main."

Thought about dynamic API selection pendant 15 secondes

उत्तर

Voici un exemple d’extension du code existant pour qu’une IA (via Groq) décide automatiquement :

  1. Quelle API/endoint utiliser en fonction d’une requête « en langage naturel » de l’utilisateur.
  2. Quels paramètres renseigner pour l’appel.
  3. Comment formuler la réponse finale à renvoyer à l’utilisateur.

Le principe est le suivant :

  • On présente à Groq la liste (résumée) des APIs configurées (avec leurs endpoints, types, paramètres, etc.).
  • On lui fournit la requête utilisateur (message).
  • On lui demande de renvoyer exclusivement un JSON indiquant :
    • chosen_api: Le nom (ou l’index) de l’API à appeler
    • chosen_endpoint: Le nom (ou l’index) de l’endpoint à appeler
    • params_values: un dictionnaire {"param1": "val1", ...} correspondant aux valeurs extraites du message pour les paramètres requis/optionnels.
  • On parse ce JSON côté Python : on récupère chosen_api, chosen_endpoint et params_values.
  • On exécute l’appel HTTP correspondant (de manière similaire à la fonction tester_endpoint).
  • On réinjecte la réponse brute de l’API dans une nouvelle requête Groq pour générer une réponse finale, plus « naturelle ».

Vous trouverez ci-dessous un code complet (intégrant le vôtre) avec une nouvelle méthode gerer_requete_naturelle. Pour tester rapidement, vous pouvez ajouter un nouvel item de menu (« 7. Requête en langage naturel ») ou simplement l’appeler en direct depuis un script séparé.

import sys import json import os import requests from groq import Groq # --------------------------------------------------------------------------- # Constantes / Variables globales FICHIER_APIS = "apis_configurees.json" MAX_CHARS_DIRECT_ANALYSIS = 3000 DEFAULT_GROQ_API_KEY = "gsk_votre_cle_par_defaut_ici" # Adapter si besoin class ApiManager: """ Classe regroupant toutes les méthodes nécessaires : - Lecture/écriture des APIs (JSON) - Intéraction avec l'IA Groq pour résumé/analyse JSON - Menu et actions (ajout, modification, test, etc.) """ # Liste globale des APIs configurées apis_configurees = [] # ----------------------------------------------------------------------- # 1. Fonctions d'IO pour sauvegarder/charger les APIs @staticmethod def lire_int_in_range(prompt: str, min_val: int, max_val: int) -> int: """ Demande à l'utilisateur de saisir un entier dans [min_val..max_val]. Gère également le cas où l'utilisateur saisit '?' pour demander de l'aide. Retourne l'entier validé. """ while True: saisie = input(prompt).strip() if saisie == "?": print("Aide : veuillez saisir un entier compris entre " f"{min_val} et {max_val}. Tapez Ctrl+C pour annuler.") continue try: val = int(saisie) if min_val <= val <= max_val: return val else: print(f"Choix invalide (doit être entre {min_val} et {max_val}).\n") except ValueError: print("Choix invalide (entrez un nombre). Réessayez.\n") @staticmethod def sauvegarder_apis() -> None: """ Sauvegarde la liste des APIs dans FICHIER_APIS au format JSON. """ try: with open(FICHIER_APIS, "w", encoding="utf-8") as f: json.dump(ApiManager.apis_configurees, f, indent=2, ensure_ascii=False) print(f"[INFO] Liste des APIs sauvegardée dans {FICHIER_APIS}.") except Exception as e: print(f"[ERREUR] Impossible de sauvegarder dans {FICHIER_APIS} : {e}") @staticmethod def charger_apis() -> None: """ Charge la liste des APIs depuis FICHIER_APIS, si présent. """ if os.path.exists(FICHIER_APIS): try: with open(FICHIER_APIS, "r", encoding="utf-8") as f: ApiManager.apis_configurees = json.load(f) print(f"[INFO] {len(ApiManager.apis_configurees)} API(s) chargée(s) depuis {FICHIER_APIS}.") except Exception as e: print(f"[ERREUR] Impossible de lire {FICHIER_APIS} : {e}") ApiManager.apis_configurees = [] else: ApiManager.apis_configurees = [] # ----------------------------------------------------------------------- # 2. Fonction de résumé si la doc est trop volumineuse @staticmethod def summarizer_groq(long_text: str, groq_api_key: str = DEFAULT_GROQ_API_KEY) -> str: """ Découpe la documentation en chunks et génère un résumé global (texte brut). On utilise le modèle Chat via l'API Groq. """ client = Groq(api_key=groq_api_key) chunk_size = 3000 chunks = [] start = 0 while start < len(long_text): end = start + chunk_size chunks.append(long_text[start:end]) start = end summaries = [] for i, chunk_text in enumerate(chunks, start=1): print(f"[INFO] Résumé du chunk {i}/{len(chunks)}...") messages = [ { "role": "system", "content": ( "Tu es un assistant chargé de résumer la documentation d'API. " "Retourne un résumé concis (en langage courant) du texte fourni. " "Ne génère pas de JSON, pas de code, juste un texte résumé clair." ) }, { "role": "user", "content": chunk_text } ] try: completion = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=messages, temperature=0.2, top_p=1, max_completion_tokens=600, # Par exemple pour limiter stream=False ) summary_text = completion.choices[0].message.content.strip() summaries.append(summary_text) except Exception as e: print(f"[ERREUR] Impossible de résumer le chunk {i} : {e}") summaries.append("") combined_summary = "\n".join(summaries) print(f"[INFO] Résumé global généré. Longueur du résumé : {len(combined_summary)} caractères.") return combined_summary # ----------------------------------------------------------------------- # 3. Fonction d'analyse finale en JSON @staticmethod def analyser_api_avec_groq(input_text: str, groq_api_key: str = DEFAULT_GROQ_API_KEY) -> dict: """ 1) Si le texte est trop long, on génère un résumé via summarizer_groq. 2) On passe ensuite (ou directement) le texte final dans un prompt qui demande un JSON strict. 3) Retourne un dict Python. """ # 1) Résume si trop volumineux text_for_analysis = input_text if len(input_text) > MAX_CHARS_DIRECT_ANALYSIS: print("[INFO] La documentation est trop volumineuse, on génère un résumé d’abord.") text_for_analysis = ApiManager.summarizer_groq(input_text, groq_api_key=groq_api_key) # 2) Appel Chat Completion pour le JSON client = Groq(api_key=groq_api_key) messages = [ { "role": "system", "content": ( "Tu es un assistant d'analyse d'API. Ta réponse doit être EXCLUSIVEMENT du JSON valide. " "Respecte ce schéma complet :\n\n" "{\n" " \"api_type\": \"REST | GraphQL | SOAP\",\n" " \"base_url\": \"string\",\n" " \"auth_info\": {\n" " \"type\": \"string (Bearer, Basic, etc.)\",\n" " \"key_name\": \"string\",\n" " \"required\": true/false\n" " },\n" " \"common_headers\": {\n" " \"Header-Name\": \"Header-Value\"\n" " },\n" " \"endpoints\": [\n" " {\n" " \"name\": \"string\",\n" " \"path\": \"string\",\n" " \"method\": \"GET | POST | ...\",\n" " \"required_params\": [\"param1\", \"param2\"],\n" " \"optional_params\": [\"param1\", \"param2\"],\n" " \"description\": \"string\"\n" " }\n" " ],\n" " \"suggested_name\": \"string\",\n" " \"rate_limit\": \"string\",\n" " \"confidence_score\": 0.0\n" "}\n\n" "Ne fournis aucun texte additionnel. Juste le JSON." ) }, { "role": "user", "content": text_for_analysis } ] try: chat_completion = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=messages, temperature=0.0, top_p=1, max_completion_tokens=1500, # Ajuster selon la taille souhaitée stream=False ) content = chat_completion.choices[0].message.content.strip() # Tentative de parsing JSON data = json.loads(content) return data except Exception as e: print(f"[ERREUR] Échec de la conversion en JSON : {e}") print("[INFO] Retour d’un squelette par défaut.") return { "api_type": None, "base_url": None, "auth_info": { "type": None, "key_name": None, "required": False }, "common_headers": {}, "endpoints": [], "suggested_name": "API_Inconnue", "rate_limit": None, "confidence_score": 0.0 } # ----------------------------------------------------------------------- # 4. FONCTIONS D’OUTIL POUR L’INTERFACE CLI @staticmethod def selectionner_api() -> dict | None: """ Affiche la liste des API configurées, demande un choix, retourne le dict de l'API choisie. Retourne None s'il n'y en a aucune. """ if not ApiManager.apis_configurees: print("Aucune API n’est configurée.\n") return None print("=== Liste des APIs disponibles ===") for i, api_info in enumerate(ApiManager.apis_configurees, start=1): print(f"{i}. {api_info['name']}") choix_num = ApiManager.lire_int_in_range( prompt=f"Quelle API souhaitez-vous sélectionner ? [1..{len(ApiManager.apis_configurees)}] : ", min_val=1, max_val=len(ApiManager.apis_configurees) ) return ApiManager.apis_configurees[choix_num - 1] @staticmethod def selectionner_endpoint(api_info: dict) -> dict | None: """ Affiche la liste des endpoints, demande un choix, retourne le dict du endpoint choisi. Retourne None s'il n'y en a aucun. """ if not api_info["endpoints"]: print("Cette API n’a aucun endpoint configuré.\n") return None print(f"Endpoints disponibles pour {api_info['name']} :") for i, ep in enumerate(api_info["endpoints"], start=1): print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})") choix_ep_num = ApiManager.lire_int_in_range( prompt=f"Quel endpoint tester ? [1..{len(api_info['endpoints'])}] : ", min_val=1, max_val=len(api_info["endpoints"]) ) return api_info["endpoints"][choix_ep_num - 1] # ----------------------------------------------------------------------- # 5. Fonctions du menu @staticmethod def ajouter_api() -> None: """ Ajoute une nouvelle API en collant sa documentation (utilise l'IA Groq pour l'analyser). """ print("=== Ajout d’une nouvelle API ===") print("Collez la documentation ci-dessous (tapez 'skibidi' sur une ligne vide pour finir) :") lignes_doc = [] while True: ligne = input() if ligne.strip() == "skibidi": break lignes_doc.append(ligne) doc_api = "\n".join(lignes_doc).strip() if not doc_api: doc_api = "Pas d'information sur l'API." # Analyse par IA resultats_analyse = ApiManager.analyser_api_avec_groq(doc_api) confidence_score = resultats_analyse.get("confidence_score", 0.0) if confidence_score < 0.5: print( "Le score de confiance de l’analyse est faible. " "Les informations retournées peuvent être partielles ou incorrectes.\n" ) # Champs retournés api_type = resultats_analyse.get("api_type") base_url = resultats_analyse.get("base_url") auth_info = resultats_analyse.get("auth_info", {}) common_headers = resultats_analyse.get("common_headers", {}) endpoints = resultats_analyse.get("endpoints", []) suggested_name = resultats_analyse.get("suggested_name", "NouvelleAPI") rate_limit = resultats_analyse.get("rate_limit") print(f"Nom suggéré pour l’API : {suggested_name}") nom_api = input(f"Entrez le nom de l’API ({suggested_name}) : ").strip() if not nom_api: nom_api = suggested_name # Si l'API nécessite une clé/token api_key_value = None if auth_info.get("required", False): api_key_value = input("Entrez la valeur de la clé API ou du token : ") # Construction de la config nouvelle_api = { "name": nom_api, "type": api_type, "base_url": base_url, "auth_info": auth_info, "api_key_value": api_key_value, "common_headers": common_headers, "endpoints": endpoints, "rate_limit": rate_limit } ApiManager.apis_configurees.append(nouvelle_api) print(f"\n[OK] L’API '{nom_api}' a été ajoutée.") print(f" - Type : {api_type}") print(f" - URL de base : {base_url}\n") # Sauvegarde ApiManager.sauvegarder_apis() @staticmethod def lister_apis() -> None: """ Affiche toutes les APIs configurées (avec leurs endpoints). """ print("=== Liste des API configurées ===") if not ApiManager.apis_configurees: print("Aucune API n’est configurée.\n") return for i, api_info in enumerate(ApiManager.apis_configurees, start=1): print(f"{i}. Nom : {api_info['name']}") print(f" URL de base : {api_info['base_url'] or 'Non renseignée'}") print(f" Type : {api_info['type'] or 'Inconnu'}") if api_info["rate_limit"]: print(f" Limitation de débit : {api_info['rate_limit']}") if api_info["endpoints"]: print(" Endpoints :") for ep in api_info["endpoints"]: print(f" - {ep['name']} ({ep['method']} {ep['path']})") else: print(" Aucun endpoint configuré.") print() print() @staticmethod def modifier_api() -> None: """ Permet de modifier diverses infos sur une API existante : URL de base, clé, endpoints, etc. """ print("=== Modification d’une API ===") api_choisie = ApiManager.selectionner_api() if not api_choisie: return print(f"\nAPI sélectionnée : {api_choisie['name']}") print("Que voulez-vous modifier ? (ex: 'changer l'URL de base', 'changer la clé API', " "'ajouter un endpoint', 'modifier un endpoint', 'supprimer un endpoint')") action = input("> ").lower() if "url" in action: new_url = input("Nouvelle URL de base : ") api_choisie["base_url"] = new_url print("URL de base modifiée.\n") if "clé" in action or "key" in action: new_key = input("Nouvelle clé API : ") api_choisie["api_key_value"] = new_key print("Clé API modifiée.\n") if "ajouter un endpoint" in action: print("=== Ajout d'un endpoint ===") nom_ep = input("Nom du nouvel endpoint : ") path_ep = input("Chemin (ex: /playlists/{playlist_id}/tracks) : ") methode = input("Méthode HTTP (GET, POST, etc.) : ") params_req = [p.strip() for p in input("Paramètres requis (séparés par des virgules) : ").split(",") if p.strip()] params_opt = [p.strip() for p in input("Paramètres optionnels (séparés par des virgules) : ").split(",") if p.strip()] new_ep = { "name": nom_ep, "path": path_ep, "method": methode.upper(), "required_params": params_req, "optional_params": params_opt, "description": "" } api_choisie["endpoints"].append(new_ep) print(f"Endpoint '{nom_ep}' ajouté.\n") if "modifier un endpoint" in action: print("=== Modification d'un endpoint ===") if not api_choisie["endpoints"]: print("Aucun endpoint à modifier.") return for i, ep in enumerate(api_choisie["endpoints"], start=1): print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})") choix_ep = ApiManager.lire_int_in_range( f"Quel endpoint modifier ? [1..{len(api_choisie['endpoints'])}] : ", 1, len(api_choisie["endpoints"]) ) ep_choisi = api_choisie["endpoints"][choix_ep - 1] print(f"Endpoint sélectionné : {ep_choisi['name']}") modif_ep = input("Que voulez-vous faire ? (ex: 'ajouter un paramètre', 'changer la méthode', etc.) : ").lower() if "ajouter un paramètre" in modif_ep: nom_p = input("Nom du paramètre : ") r_o = input("Paramètre requis ou optionnel ? (r/o) : ") if r_o.lower() == 'r': ep_choisi["required_params"].append(nom_p) else: ep_choisi["optional_params"].append(nom_p) print(f"Paramètre '{nom_p}' ajouté.\n") # D'autres modifications possibles, à adapter selon vos besoins... if "supprimer un endpoint" in action: print("=== Suppression d'un endpoint ===") if not api_choisie["endpoints"]: print("Aucun endpoint à supprimer.") return for i, ep in enumerate(api_choisie["endpoints"], start=1): print(f"{i}. {ep['name']} ({ep['method']} {ep['path']})") choix_ep = ApiManager.lire_int_in_range( f"Quel endpoint supprimer ? [1..{len(api_choisie['endpoints'])}] : ", 1, len(api_choisie["endpoints"]) ) ep_asup = api_choisie["endpoints"][choix_ep - 1] api_choisie["endpoints"].remove(ep_asup) print(f"Endpoint '{ep_asup['name']}' supprimé.") print("Modification(s) terminée(s).\n") ApiManager.sauvegarder_apis() @staticmethod def supprimer_api() -> None: """ Supprime complètement une API de la liste (après confirmation). """ print("=== Suppression d’une API ===") api_choisie = ApiManager.selectionner_api() if not api_choisie: return confirm = input(f"Êtes-vous sûr de vouloir supprimer '{api_choisie['name']}' ? (o/n) : ").lower() if confirm == 'o': ApiManager.apis_configurees.remove(api_choisie) print(f"L’API '{api_choisie['name']}' a été supprimée.\n") ApiManager.sauvegarder_apis() else: print("Opération annulée.\n") @staticmethod def tester_endpoint() -> None: """ Permet de tester un endpoint en temps réel : - Sélectionne une API - Sélectionne un endpoint - Demande les paramètres requis/optionnels - Lance la requête HTTP et affiche la réponse. """ print("=== Tester un endpoint ===") api_choisie = ApiManager.selectionner_api() if not api_choisie: return ep_choisi = ApiManager.selectionner_endpoint(api_choisie) if not ep_choisi: return # Paramètres requis required_params_values = {} for rp in ep_choisi.get("required_params", []): val = input(f"Entrez la valeur pour '{rp}' : ") required_params_values[rp] = val # Paramètres optionnels optional_params_values = {} for op in ep_choisi.get("optional_params", []): val = input(f"Valeur pour '{op}' (laisser vide si aucun) : ") if val: optional_params_values[op] = val # On utilise la fonction utilitaire interne : response = ApiManager.appeler_endpoint_direct( api_choisie, ep_choisi, required_params_values, optional_params_values ) if response is None: print("[ERREUR] Appel impossible ou méthode non gérée.") return r, status_code = response print(f"Statut HTTP : {status_code}") print("==== Contenu de la réponse ====") try: print(json.dumps(r.json(), indent=2, ensure_ascii=False)) except Exception: print(r.text) @staticmethod def appeler_endpoint_direct( api_info: dict, endpoint_info: dict, required_params_values: dict, optional_params_values: dict ): """ Fonction de bas niveau pour appeler effectivement un endpoint donné. Retourne (objet_response, status_code) ou None si problème. """ # Préparation de l’URL base_url = api_info.get("base_url") or "" path = endpoint_info["path"] method = endpoint_info["method"].upper() # Remplacement des placeholders {param} dans le path query_params = {} for k, v in required_params_values.items(): placeholder = f"{{{k}}}" if placeholder in path: path = path.replace(placeholder, v) else: query_params[k] = v # Les optionnels vont aussi dans les query params (ou body JSON si besoin) query_params.update(optional_params_values) # Gestion de l’authentification headers = dict(api_info.get("common_headers", {})) if "Accept" not in headers: headers["Accept"] = "application/json" auth_info = api_info.get("auth_info", {}) api_key_value = api_info.get("api_key_value") if auth_info.get("required") and api_key_value: # Exemple : si c’est un Bearer if auth_info["type"] and "bearer" in auth_info["type"].lower(): key_name = auth_info.get("key_name", "Authorization") headers[key_name] = f"Bearer {api_key_value}" # Exemple : si c’est un Basic elif auth_info["type"] and "basic" in auth_info["type"].lower(): key_name = auth_info.get("key_name", "Authorization") headers[key_name] = f"Basic {api_key_value}" else: # Parfois, c'est un param "api_key" dans l'URL if auth_info.get("key_name", "").lower() == "api_key": query_params["api_key"] = api_key_value else: # Sinon, on le place dans un header key_name = auth_info.get("key_name", "X-API-Key") headers[key_name] = api_key_value url = base_url + path # Exécution de la requête try: if method == "GET": r = requests.get(url, headers=headers, params=query_params) elif method == "POST": r = requests.post(url, headers=headers, json=query_params) elif method == "PUT": r = requests.put(url, headers=headers, json=query_params) elif method == "DELETE": r = requests.delete(url, headers=headers, json=query_params) else: print("Méthode non gérée.") return None return r, r.status_code except Exception as e: print(f"Erreur lors de l’appel : {e}") return None # ----------------------------------------------------------------------- # ======================= NOUVELLE PARTIE ================================ # ----------------------------------------------------------------------- @staticmethod def gerer_requete_naturelle(message: str, groq_api_key: str = DEFAULT_GROQ_API_KEY) -> str: """ 1) Présente la liste des APIs + endpoints au modèle Groq, pour qu'il choisisse laquelle appeler en fonction du message utilisateur. 2) Reçoit en retour un JSON strict avec : { "chosen_api": "NomOuIndexDeLApi", "chosen_endpoint": "NomOuIndexDeLEndpoint", "params_values": { "paramRequisOuOptionnel": "valeur" } } 3) On appelle la fonction appeler_endpoint_direct pour exécuter la requête. 4) On envoie la réponse brute (ou JSON) à Groq pour qu’il génère une réponse en langage naturel. 5) On renvoie cette réponse finale. """ if not ApiManager.apis_configurees: return "Aucune API n’est configurée, impossible de satisfaire la requête." # 1) Construire un résumé JSON de toutes les APIs et de leurs endpoints # pour donner au modèle de quoi choisir. apis_description = [] for idx_api, api_info in enumerate(ApiManager.apis_configurees, start=1): endpoints = [] for idx_ep, ep in enumerate(api_info["endpoints"], start=1): endpoints.append({ "index": idx_ep, "name": ep["name"], "method": ep["method"], "path": ep["path"], "required_params": ep["required_params"], "optional_params": ep["optional_params"] }) apis_description.append({ "index": idx_api, "name": api_info["name"], "base_url": api_info["base_url"], "endpoints": endpoints }) # On fournit cette description + le message utilisateur à Groq client = Groq(api_key=groq_api_key) system_instructions = ( "Tu es un assistant chargé d'orchestrer l'appel à l'une des APIs listées.\n" "Tu disposes d'un JSON décrivant les APIs connues (avec leurs endpoints et paramètres). " "Tu disposes aussi du message utilisateur en langage naturel.\n\n" "Tu dois décider :\n" " - Quelle API (par index ou par nom) est la plus pertinente ?\n" " - Quel endpoint (par index ou par nom) est le plus pertinent ?\n" " - Quels paramètres/valeurs en extrais-tu du message ?\n\n" "Ta réponse DOIT être du JSON strict, sans texte supplémentaire, sans explications.\n" "Structure de réponse OBLIGATOIRE :\n" "{\n" " \"chosen_api\": \"...\", // index ou nom, p.ex. 1 ou \"WeatherAPI\"\n" " \"chosen_endpoint\": \"...\", // index ou nom, p.ex. 2 ou \"getCityWeather\"\n" " \"params_values\": {\n" " \"paramRequis\": \"valeur\",\n" " \"paramOptionnel\": \"valeur\"\n" " }\n" "}\n" "Aucun autre champ n'est autorisé.\n" ) messages = [ { "role": "system", "content": system_instructions }, { "role": "user", "content": json.dumps({ "apis": apis_description, "user_request": message }, ensure_ascii=False, indent=2) } ] try: choix_call = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=messages, temperature=0.1, top_p=1, max_completion_tokens=800, stream=False ).choices[0].message.content.strip() # 2) On parse la réponse JSON data_call = json.loads(choix_call) chosen_api = data_call.get("chosen_api") chosen_endpoint = data_call.get("chosen_endpoint") params_values = data_call.get("params_values", {}) except Exception as e: return f"[ERREUR IA] Impossible de déterminer l'API/endpoint : {e}" # Trouver l’API par index ou par nom api_info = None if isinstance(chosen_api, int): # Appel direct via l’index (en 1-based) idx = chosen_api - 1 if 0 <= idx < len(ApiManager.apis_configurees): api_info = ApiManager.apis_configurees[idx] else: # Recherche par nom (approximatif, au besoin on fait .lower()) for a in ApiManager.apis_configurees: if a["name"].lower() == str(chosen_api).lower(): api_info = a break if not api_info: return f"[ERREUR] L’IA a choisi une API inexistante : {chosen_api}" # Trouver l’endpoint par index ou par nom endpoint_info = None if isinstance(chosen_endpoint, int): idx_ep = chosen_endpoint - 1 if 0 <= idx_ep < len(api_info["endpoints"]): endpoint_info = api_info["endpoints"][idx_ep] else: for ep in api_info["endpoints"]: if ep["name"].lower() == str(chosen_endpoint).lower(): endpoint_info = ep break if not endpoint_info: return f"[ERREUR] L’IA a choisi un endpoint inexistant : {chosen_endpoint}" # Séparer les required_params / optional_params req_list = endpoint_info.get("required_params", []) opt_list = endpoint_info.get("optional_params", []) req_values = {} opt_values = {} for k, v in params_values.items(): # On regarde si ce param fait partie des "required" ou "optional" if k in req_list: req_values[k] = v elif k in opt_list: opt_values[k] = v else: # S’il n’est ni dans required ni optional, # on peut soit l’ajouter aux optionnels par défaut, # soit l’ignorer. Ici, on l’ajoute aux optionnels. opt_values[k] = v # 3) Appeler effectivement l’endpoint resultat_appel = ApiManager.appeler_endpoint_direct( api_info, endpoint_info, req_values, opt_values ) if not resultat_appel: return "[ERREUR] Appel impossible (méthode non gérée ou autre)." r, status_code = resultat_appel # 4) Convertir la réponse brute en un court résumé &laquo; en langage naturel &raquo; try: response_json = r.json() raw_str = json.dumps(response_json, ensure_ascii=False) except: # Pas du JSON, on renvoie le text brut raw_str = r.text # On fait un 2ème appel à Groq pour résumer la réponse en langage courant summarizer_messages = [ { "role": "system", "content": ( "Tu es un assistant chargé de rédiger une réponse pour l'utilisateur final, " "en te basant sur la réponse brute (JSON ou texte) d'une API. " "Sois clair et concis.\n" "Ne donne pas le JSON brut, mais un résumé informatif ou la donnée principale.\n" ) }, { "role": "user", "content": ( f"L'utilisateur a demandé : '{message}'.\n\n" f"Voici la réponse brute de l'API :\n{raw_str}\n\n" "Rédige la réponse en langage naturel :" ) } ] try: final_answer = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=summarizer_messages, temperature=0.1, top_p=1, max_completion_tokens=600, stream=False ).choices[0].message.content.strip() except Exception as e: return f"[ERREUR IA] Impossible de produire la réponse finale : {e}" return final_answer # ----------------------------------------------------------------------- # 6. MAIN @staticmethod def main() -> None: """ Fonction principale de l'application (menu en boucle). """ # On charge la liste des APIs au démarrage ApiManager.charger_apis() while True: print("=== Menu Principal ===") print("1. Ajouter une API (assisté par IA)") print("2. Liste des API configurées") print("3. Modifier une API") print("4. Supprimer une API") print("5. Tester un endpoint") print("6. Requête en langage naturel (IA décide l’endpoint)") print("7. Quitter") choix_num = ApiManager.lire_int_in_range("Choisissez une option [1..7] : ", 1, 7) if choix_num == 1: ApiManager.ajouter_api() elif choix_num == 2: ApiManager.lister_apis() elif choix_num == 3: ApiManager.modifier_api() elif choix_num == 4: ApiManager.supprimer_api() elif choix_num == 5: ApiManager.tester_endpoint() elif choix_num == 6: # On demande un message à l'utilisateur message_utilisateur = input("Tapez votre requête : ") reponse_ia = ApiManager.gerer_requete_naturelle(message_utilisateur) print("\n==== Réponse IA ====") print(reponse_ia) print() elif choix_num == 7: print("Au revoir.") sys.exit(0) # --------------------------------------------------------------------------- # Point d'entrée du script if __name__ == "__main__": ApiManager.main()

Explication de la nouvelle méthode gerer_requete_naturelle

  1. Récupération de la liste d’APIs : on parcourt ApiManager.apis_configurees pour construire un objet apis_description, qui liste pour chaque API :

    • un index (à partir de 1)
    • le name, la base_url
    • la liste de ses endpoints (avec leurs indices, noms, méthodes, paramètres…).
  2. Appel Groq (1er prompt) :

    • On envoie d’abord au modèle un system message qui exige une réponse en JSON strict (sans texte) contenant :
      { "chosen_api": "nomOuIndex", "chosen_endpoint": "nomOuIndex", "params_values": { "param1": "valeur1", "param2": "valeur2" } }
    • On joint en role=user un JSON contenant la liste des APIs (apis_description) + le message de l’utilisateur (user_request).
    • Groq doit alors choisir la bonne API, l’endpoint adéquat et déterminer les valeurs de paramètres.
  3. Parsing de la réponse : on parse le JSON retourné pour extraire chosen_api, chosen_endpoint et params_values. Ensuite on identifie réellement l’API et l’endpoint (qu’ils soient référencés par index ou par nom).

  4. Appel effectif de l’endpoint : via la fonction interne appeler_endpoint_direct déjà créée. Celle-ci reconstruit l’URL, gère l’authentification, envoie la requête requests et renvoie la réponse.

  5. Appel Groq (2e prompt) : on envoie la réponse brute de l’API (JSON ou texte) + le message original de l’utilisateur à un second prompt Groq, qui doit générer une réponse finale « en langage courant ».

  6. Renvoi : on obtient le texte final à renvoyer à l’utilisateur.


Ainsi, l’utilisateur final peut poser des questions ou formuler une requête du genre :

« Je voudrais la météo à Paris pour la journée de demain »

L’IA va choisir l’API Météo si elle existe, sélectionner le bon endpoint (par ex. « getCityForecast »), remplir les paramètres (city = "Paris", date = "2025-01-19"), puis faire l’appel et générer la réponse finale.

इस प्रश्नोत्तर को साझा करें