Comment créer un agent avec le Sandbox Novita AI, les produits LLM et Browser Use.

Table des matières

Qu'est-ce qu'un agent IA ?
Présentation du Sandbox Novita AI, des produits LLM et de Browser Use
Configuration de votre environnement de développement
Création de l'agent analyste de données exploratoires (EDA)
Conclusion

Ces derniers temps, nous avons observé une croissance exponentielle dans le domaine de l’intelligence artificielle, avec les grands modèles de langage (LLM), les modèles de langage visuel (VLM), etc., et plus récemment, les agents IA. Contrairement aux applications traditionnelles basées sur des règles, ces agents sont capables d’utiliser leurs capacités de raisonnement et de prise de décision pour effectuer des tâches complexes de manière autonome ou avec une intervention minimale de l’utilisateur. Pour ce faire, les agents IA ont souvent besoin de générer et d’exécuter du code dynamiquement, voire de contrôler l’ensemble de la machine virtuelle, et le sandbox Novita AI offre ces capacités.

Dans ce tutoriel, nous allons créer un agent IA qui servira d’analyste de données, et tout comme un analyste humain, il sera capable de trouver et télécharger un jeu de données en utilisant le navigateur selon nos instructions, et lorsque nous le lui demanderons, d’analyser, visualiser et exécuter du code pour nous répondre avec des informations tirées de ces données.

Qu’est-ce qu’un agent IA ?

Un agent IA est un programme logiciel qui utilise l’intelligence artificielle pour raisonner, établir des plans et mener des actions afin d’atteindre des objectifs spécifiques. Ces agents peuvent utiliser des outils, tels que des API, l’exécution de code et des recherches web, pour collecter des informations ou effectuer des tâches. Parfois, ils collaborent également avec d’autres agents, notamment pour des objectifs plus complexes, comme Deep Research.

Voici les composants typiques d’un agent IA :

Raisonnement/Planification : Lorsqu’un objectif lui est donné, l’agent élabore des plans étape par étape pour l’atteindre. Ce processus peut impliquer la réalisation de sous-tâches et la collecte d’informations supplémentaires pour guider ses prochaines actions. Cela est particulièrement courant pour les tâches complexes qui prennent plus de temps, et les modèles de raisonnement fonctionnent bien dans ces situations.
Utilisation d’outils : Les agents IA peuvent appeler des outils et services externes. Ceux-ci peuvent inclure des fonctions, des API ou d’autres ressources qui étendent leurs capacités.
Coordination : Plusieurs agents peuvent travailler ensemble en partageant les responsabilités, chacun contribuant à atteindre un objectif commun, complexe ou à long terme.

Présentation du Sandbox Novita AI, des produits LLM et de Browser Use

Sandbox Novita AI

Ce qu’est un sandbox et pourquoi il est important

Un sandbox est un environnement d’exécution sécurisé et isolé dans lequel du code non fiable peut être exécuté sans affecter le système hôte. C’est essentiellement un ordinateur virtuel léger permettant à votre agent IA d’exécuter du code, des commandes, de créer des fichiers, etc.

Novita AI propose ce sandbox dans le cloud pour que votre agent y accède rapidement à la demande, avec une facturation flexible à la seconde en fonction des ressources utilisées.

Fonctionnalités clés du sandbox Novita :

Isolation sécurisée : Chaque sandbox dispose de son propre système de fichiers et environnement isolés, protégeant les données et empêchant les interactions non souhaitées.
Démarrage rapide : Les instances de sandbox se lancent en moyenne en moins de ~200 ms, ce qui les rend idéales pour les scénarios à faible latence.
Prise en charge de plusieurs langages : Vous pouvez exécuter du code dans plusieurs langages de programmation, notamment Python, JavaScript, TypeScript, et plus encore.
Pause et reprise rapides : Mettez le sandbox en pause à tout moment et reprenez-le lorsque vous en avez besoin, l’état du système de fichiers et des processus étant entièrement restauré.
Exécution en arrière-plan : Prend en charge l’exécution de tâches en arrière-plan et est adapté aux scénarios nécessitant d’attendre un résultat.

API de modèles Novita :

Modèles Novita AI

Novita propose une vaste bibliothèque de modèles IA open source provenant de laboratoires de recherche leaders comme OpenAI, Google, DeepSeek et Qwen, etc. Ceux-ci incluent des modèles pour le langage, la vision, l’audio, la vidéo et les embeddings. Nos modèles de langage sont également entièrement compatibles avec le SDK OpenAI, donc passer d’OpenAI à Novita ne nécessite que la mise à jour de l’URL de base et de la clé API dans votre client, puis la sélection d’un modèle Novita.


from openai import OpenAI client = OpenAI( base_url=“https://api.novita.ai/v3/openai”, api_key=“<Votre clé API Novita>”, )

Présentation de Browser Use :

Browser Use est une bibliothèque Python open source qui permet aux agents IA d’interagir avec des navigateurs web à l’aide de commandes en langage naturel (par exemple « Vérifiez la météo à New York aujourd’hui »). Au lieu d’écrire des sélecteurs complexes basés sur des règles, l’IA gère la recherche et l’interaction avec les éléments. Et grâce à l’API de modèles Novita compatible OpenAI, vous pouvez utiliser n’importe quel LLM ou VLM (modèle de langage visuel) pour alimenter Browser Use.

Configuration de votre environnement de développement

Pour commencer, nous allons cloner le dépôt GitHub, configurer un environnement Python propre, installer toutes les dépendances requises et obtenir les clés Novita AI.

Cloner le dépôt GitHub et installer les dépendances avec uv

1. Installez uv (un gestionnaire de paquets Python léger)


pip install uv

2. Clonez le dépôt (dépôt GitHub) et naviguez à l’intérieur.


git clone https://github.com/Studio1HQ/AI-sandbox.git cd AI-sandbox

3. Créez et activez l’environnement virtuel uv


# Crée un environnement virtuel uv venv # Active l’environnement virtuel source .venv/bin/activate # Pour Mac/Linux # ou .venv\Scripts\activate # Pour Windows

4. Installez les dépendances du projet


# Installe les dépendances uv sync

Créer un compte Novita AI et obtenir une clé API

1. Inscrivez-vous sur novita.ai.

2. Dans le tableau de bord, survolez l’icône de profil utilisateur et cliquez sur API Keys dans le menu contextuel.

3. Sur la page de gestion des clés, cliquez sur Add New Key. Dans la fenêtre contextuelle, saisissez un nom pour votre clé, cliquez sur Confirm, puis copiez la clé générée.

4. Maintenant, dans le répertoire du projet, créez un fichier .env et collez ce qui suit.


NOVITA_API_KEY=“<COLLEZ VOTRE CLÉ API NOVITA ICI>” NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1”

Ajoutez des crédits sur votre compte Novita AI.

Pour utiliser le sandbox Novita, vous devez ajouter des crédits sur votre compte. Dans l’onglet du tableau de bord, cliquez sur ‘Billing.’ Ensuite, sur la page de facturation, ajoutez un moyen de paiement et rechargez au moins 10 $ de crédits.

Création de l’agent analyste de données exploratoires (EDA)

Maintenant que notre environnement est configuré et que nous avons notre clé API, lançons notre agent.

Téléchargement de jeux de données (ou de fichiers) via l’utilisation du navigateur par l’agent :

Tout comme un analyste humain, nous voulons que notre agent soit capable de recevoir des instructions et d’utiliser un navigateur pour obtenir les jeux de données. Ainsi, comme on le voit dans browser_agent.py, nous créons d’abord une structure de données Pydantic pour la sortie finale de l’agent, qui contiendra les noms des fichiers téléchargés et les résultats des tâches terminées à écrire dans un fichier.


… # en dessous du code existant class TaskFile(BaseModel): “”“Représente un fichier généré dans le cadre d’un résultat de tâche, par exemple des données scrapées ou des données recherchées.”“” filename: str = Field(…, description=“Nom du fichier incluant l’extension”) content: str = Field(…, description=“Contenu texte à écrire dans le fichier”) class AgentOutput(BaseModel): “”“Sortie finale agrégée de l’exécution de l’agent de navigateur.”“” downloaded_files: Optional[list[str]] = Field( None, description=“Liste des noms de fichiers téléchargés (avec extension), le cas échéant” ) task_files: Optional[list[TaskFile]] = Field( None, description=“Fichiers générés à partir des tâches utilisateur (par exemple, données scrapées ou recherchées), le cas échéant” )

… # en dessous du code existant
class TaskFile(BaseModel):
“”“Représente un fichier généré dans le cadre d’un résultat de tâche, par exemple des données scrapées ou des données recherchées.”“”

filename: str = Field(…, description=“Nom du fichier incluant l’extension”)
content: str = Field(…, description=“Contenu texte à écrire dans le fichier”)

class AgentOutput(BaseModel):
“”“Sortie finale agrégée de l’exécution de l’agent de navigateur.”“”

downloaded_files: Optional[list[str]] = Field(
None, description=“Liste des noms de fichiers téléchargés (avec extension), le cas échéant”
)
task_files: Optional[list[TaskFile]] = Field(
None, description=“Fichiers générés à partir des tâches utilisateur (par exemple, données scrapées ou recherchées), le cas échéant”
)

Vous trouverez ci-dessous un exemple de fonctionnement du navigateur de l’agent. Nous utiliserons un grand modèle de langage (LLM) pour la navigation web, ce qui devrait couvrir la plupart des cas d’usage. Cependant, pour des tâches d’interface utilisateur plus complexes (par exemple, la résolution de captcha), un modèle de langage visuel (VLM) est recommandé. Lors de l’utilisation d’un VLM, vous pouvez définir le niveau de détail visuel (« élevé », « bas » ou « automatique ») pour équilibrer le coût en tokens et la clarté visuelle.

Nous créons ensuite une session de navigateur, configurons le profil avec un chemin de téléchargement (./Download) et user_data_dir (défini sur None pour le mode navigation privée), et définissons notre modèle Pydantic comme contrôleur pour obtenir les sorties de l’agent structurées. Ensuite, nous démarrons l’agent avec await agent.run(), et les résultats finaux sont analysés pour obtenir les noms des fichiers téléchargés.


# Un exemple de l’agent de navigateur agent = Agent( task=“Allez sur Hugging Face, recherchez An-j96/SuperstoreData et ouvrez sa page, puis naviguez vers l’onglet Fichiers et téléchargez le fichier csv de données.”, llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key), use_vision=False, vision_detail_level=“auto”, # options disponibles [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=“./Download”, user_data_dir=None ) ), # définit le chemin du répertoire de téléchargement pour le navigateur. controller=Controller( output_model=AgentOutput ), # Demande à l’agent de renvoyer le nom du fichier téléchargé à la fin de la tâche. ) all_results = await agent.run() final_output = AgentOutput.model_validate_json( all_results.final_result() ) # analyse le résultat final de l’agent.

# Un exemple de l’agent de navigateur
agent = Agent(
task=“Allez sur Hugging Face, recherchez An-j96/SuperstoreData et ouvrez sa page, puis naviguez vers l’onglet Fichiers et téléchargez le fichier csv de données.”,
llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key),
use_vision=False,
vision_detail_level=“auto”, # options disponibles [‘low’, ‘high’, ‘auto’];
browser_session=BrowserSession(
browser_profile=BrowserProfile(
downloads_path=“./Download”, user_data_dir=None
)
), # définit le chemin du répertoire de téléchargement pour le navigateur.
controller=Controller(
output_model=AgentOutput
), # Demande à l’agent de renvoyer le nom du fichier téléchargé à la fin de la tâche.
)

all_results = await agent.run()
final_output = AgentOutput.model_validate_json(
all_results.final_result()
) # analyse le résultat final de l’agent.

(Remarque : l’exécution d’un agent de navigateur sur votre machine locale lancera une instance de navigateur réelle pour que vous puissiez le voir en action.) La méthode complète collectera les noms des fichiers téléchargés, écrira les résultats des tâches dans un fichier et renverra tous les chemins de fichiers pour un téléchargement ultérieur.


… # en dessous du code existant async def downloading_task_for_browser_agent( task: str, api_key: str, model: str, model_api_base_url: str, use_vision: bool, download_dir_path: str = “./Download”, ) -> Tuple[str, list[str]]: “”“ Effectuera la tâche de téléchargement de l’utilisateur via l’utilisation du navigateur et renverra le chemin du répertoire de téléchargement et les noms des fichiers téléchargés. Retourne : Tuple de (répertoire_téléchargement, noms_fichiers_avec_extension) ”“” agent = Agent( task=task, llm=ChatOpenAI( base_url=model_api_base_url, model=model, api_key=api_key, max_completion_tokens=20_000, frequency_penalty=0, # Cette pénalité peut légèrement affecter l’utilisation des outils ; laissez-la à 0. ), use_vision=use_vision, vision_detail_level=“auto”, # options disponibles [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=download_dir_path, user_data_dir=None, # “./browser_user_data” ) ), # définit le chemin du répertoire de téléchargement pour le navigateur. controller=Controller( output_model=AgentOutput ), # Demande à l’agent de renvoyer le résultat de l’exécution de la tâche selon le schéma AgentOutput. max_failures=5, ) try: # Exécute l’agent et structure sa sortie all_results = await agent.run() final_output: AgentOutput = AgentOutput.model_validate_json( all_results.final_result() ) if final_output.task_files: console.print( Panel( f"[bold yellow]Écriture des résultats de tâche dans des fichiers…[/bold yellow] {final_output.task_files}“, title=“Résultats de tâche”, border_style=“white”, ) ) # Écrit chaque résultat de tâche dans un fichier du répertoire de téléchargement task_result_files: list[str] = [] for task_file in final_output.task_files or []: file_path = Path(task_file.filename) # Empêche les traversées de chemin ou les chemins absolus non sécurisés if file_path.is_absolute() or “…” in file_path.parts: raise ValueError( f"L’agent a transmis un chemin de fichier non sécurisé comme nom de fichier : {file_path}” ) # Place le chemin de fichier dans le répertoire de téléchargement file_path = Path(download_dir_path) / file_path # Vérifie que le répertoire de téléchargement existe, sinon le crée. file_path.parent.mkdir(parents=True, exist_ok=True) # Écrit le contenu du résultat de tâche dans un fichier with open(file_path, “w”, encoding=“utf-8”) as f: f.write(task_file.content) task_result_files.append(task_file.filename) if task_result_files: console.print( Panel( f"[bold green]Résultats de tâche écrits dans des fichiers :[/bold green] {task_result_files}“, title=“Résultats de tâche”, border_style=“white”, ) ) # Combine les fichiers téléchargés avec les fichiers de résultats de tâche file_results = (final_output.downloaded_files or []) + task_result_files if file_results: console.print( Panel( f”[bold green]Fichiers disponibles :[/bold green] {file_results} dans {download_dir_path}“, title=“Fichiers téléchargés”, border_style=“green”, ) ) else: raise RuntimeError(“Aucun fichier n’a été téléchargé ou écrit.”) except Exception as e: file_results = None console.print( Panel( f”[bold red]Erreur :[/bold red] {str(e)}\ ", title=“Erreur d’exécution”, border_style=“red”, ) ) return (download_dir_path, file_results)

… # en dessous du code existant
async def downloading_task_for_browser_agent(
task: str,
api_key: str,
model: str,
model_api_base_url: str,
use_vision: bool,
download_dir_path: str = “./Download”,
) -> Tuple[str, list[str]]:
“”“
Effectuera la tâche de téléchargement de l’utilisateur via l’utilisation du navigateur et renverra le chemin du répertoire de téléchargement et les
noms des fichiers téléchargés.

Retourne :
Tuple de (répertoire_téléchargement, noms_fichiers_avec_extension)
”“”

agent = Agent(
task=task,
llm=ChatOpenAI(
base_url=model_api_base_url,
model=model,
api_key=api_key,
max_completion_tokens=20_000,
frequency_penalty=0, # Cette pénalité peut légèrement affecter l’utilisation des outils ; laissez-la à 0.
),
use_vision=use_vision,
vision_detail_level=“auto”, # options disponibles [‘low’, ‘high’, ‘auto’];
browser_session=BrowserSession(
browser_profile=BrowserProfile(
downloads_path=download_dir_path,
user_data_dir=None, # “./browser_user_data”
)
), # définit le chemin du répertoire de téléchargement pour le navigateur.
controller=Controller(
output_model=AgentOutput
), # Demande à l’agent de renvoyer le résultat de l’exécution de la tâche selon le schéma AgentOutput.
max_failures=5,
)

try:
# Exécute l’agent et structure sa sortie
all_results = await agent.run()
final_output: AgentOutput = AgentOutput.model_validate_json(
all_results.final_result()
)

if final_output.task_files:
console.print(
Panel(
f"[bold yellow]Écriture des résultats de tâche dans des fichiers…[/bold yellow] {final_output.task_files}“,
title=“Résultats de tâche”,
border_style=“white”,
)
)

# Écrit chaque résultat de tâche dans un fichier du répertoire de téléchargement
task_result_files: list[str] = []
for task_file in final_output.task_files or []:
file_path = Path(task_file.filename)

# Empêche les traversées de chemin ou les chemins absolus non sécurisés
if file_path.is_absolute() or “…” in file_path.parts:
raise ValueError(
f"L’agent a transmis un chemin de fichier non sécurisé comme nom de fichier : {file_path}”
)

# Place le chemin de fichier dans le répertoire de téléchargement
file_path = Path(download_dir_path) / file_path

# Vérifie que le répertoire de téléchargement existe, sinon le crée.
file_path.parent.mkdir(parents=True, exist_ok=True)

# Écrit le contenu du résultat de tâche dans un fichier
with open(file_path, “w”, encoding=“utf-8”) as f:
f.write(task_file.content)

task_result_files.append(task_file.filename)

if task_result_files:
console.print(
Panel(
f"[bold green]Résultats de tâche écrits dans des fichiers :[/bold green] {task_result_files}“,
title=“Résultats de tâche”,
border_style=“white”,
)
)

# Combine les fichiers téléchargés avec les fichiers de résultats de tâche
file_results = (final_output.downloaded_files or []) + task_result_files

if file_results:
console.print(
Panel(
f”[bold green]Fichiers disponibles :[/bold green] {file_results} dans {download_dir_path}“,
title=“Fichiers téléchargés”,
border_style=“green”,
)
)
else:
raise RuntimeError(“Aucun fichier n’a été téléchargé ou écrit.”)

except Exception as e:
file_results = None
console.print(
Panel(
f”[bold red]Erreur :[/bold red] {str(e)}\ ",
title=“Erreur d’exécution”,
border_style=“red”,
)
)

return (download_dir_path, file_results)

L’appel à l’agent est encapsulé dans un bloc try-except, donc si une défaillance se produit, une exception est levée et file_results est défini sur None. Enfin, la méthode renvoie à la fois le download_dir_path et les chemins de file_results.

Définition des schémas des outils disponibles pour notre agent :

Maintenant que nous avons terminé avec l’utilisation du navigateur, il est temps de définir les schémas des outils disponibles pour notre agent EDA. Nous fournirons quatre outils :

run_python_code : permet à l’agent d’exécuter du code Python à l’intérieur du sandbox.
run_on_command_line : permet à l’agent d’exécuter des commandes dans le terminal du sandbox (par exemple, installer des paquets Python).
sync_with_user : permet à l’agent de synchroniser les fichiers et répertoires créés ou mis à jour depuis le sandbox vers votre dossier de synchronisation local.
delete_from_user_sync_folder : permet à l’agent de supprimer des fichiers ou des répertoires du dossier de synchronisation local.

Ensemble, tous ces outils donnent à l’agent un contrôle total sur l’exécution de code, l’utilisation du terminal et la synchronisation de fichiers entre le sandbox et votre système local.

Dans sandbox_eda.py, nous voyons le schéma des outils :

run_python_code, qui prend simplement le code Python à exécuter comme paramètre d’entrée et renvoie un résultat le cas échéant.


{ “type”: “function”, “function”: { “name”: “run_python_code”, “description”: “Exécute le code python et renvoie le résultat le cas échéant.”, “parameters”: { “type”: “object”, “properties”: { “python_code”: { “type”: “string”, “description”: “Le code Python à exécuter.”, } }, “required”: [“python_code”], }, }, },

run_on_command_line, qui prend à nouveau simplement la commande à exécuter comme paramètre d’entrée et renvoie un résultat le cas échéant.


{ “type”: “function”, “function”: { “name”: “run_on_command_line”, “description”: “Exécute la commande sur la ligne de commande et renvoie le résultat le cas échéant.”, “parameters”: { “type”: “object”, “properties”: { “command”: { “type”: “string”, “description”: “La commande à exécuter sur la ligne de commande.”, } }, “required”: [“command”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “run_on_command_line”,
“description”: “Exécute la commande sur la ligne de commande et renvoie le résultat le cas échéant.”,
“parameters”: {
“type”: “object”,
“properties”: {
“command”: {
“type”: “string”,
“description”: “La commande à exécuter sur la ligne de commande.”,
}
},
“required”: [“command”],
},
},
},

sync_with_user, qui prend deux paramètres d’entrée :
- sandbox_path : le chemin vers le fichier ou le répertoire à l’intérieur du sandbox.
- path_on_user_sync_folder : le chemin que l’agent souhaite que le fichier ou le répertoire prenne dans le dossier de synchronisation de l’utilisateur.

Ce deuxième chemin sera de la forme (par exemple /new.txt), en supposant que le dossier de synchronisation est le répertoire parent. Par exemple, nous résoudrons plus tard /new.txt en sync_folder/new.txt.


{ “type”: “function”, “function”: { “name”: “sync_with_user”, “description”: “Synchronisera un fichier ou un répertoire sur le sandbox, vers le dossier de synchronisation de l’utilisateur sur son ordinateur”, “parameters”: { “type”: “object”, “properties”: { “sandbox_path”: { “type”: “string”, “description”: “Chemin vers le fichier ou le répertoire sur le sandbox.”, }, “path_on_user_sync_folder”: { “type”: “string”, “description”: “Chemin relatif où le fichier ou le répertoire sera placé dans le dossier de synchronisation de l’utilisateur. Par exemple, ‘/hello.txt’ est placé directement dans le dossier de synchronisation, tandis que ‘/run1/hello.txt’ sera placé dans un sous-dossier ‘run1’ du dossier de synchronisation.”, }, }, “required”: [“sandbox_path”, “path_on_user_sync_folder”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “sync_with_user”,
“description”: “Synchronisera un fichier ou un répertoire sur le sandbox, vers le dossier de synchronisation de l’utilisateur sur son ordinateur”,
“parameters”: {
“type”: “object”,
“properties”: {
“sandbox_path”: {
“type”: “string”,
“description”: “Chemin vers le fichier ou le répertoire sur le sandbox.”,
},
“path_on_user_sync_folder”: {
“type”: “string”,
“description”: “Chemin relatif où le fichier ou le répertoire sera placé dans le dossier de synchronisation de l’utilisateur. Par exemple, ‘/hello.txt’ est placé directement dans le dossier de synchronisation, tandis que ‘/run1/hello.txt’ sera placé dans un sous-dossier ‘run1’ du dossier de synchronisation.”,
},
},
“required”: [“sandbox_path”, “path_on_user_sync_folder”],
},
},
},

delete_from_user_sync_folder, qui ne prend qu’un seul paramètre d’entrée :
- path_on_user_sync_folder : le chemin vers le fichier ou le répertoire que l’agent souhaite supprimer du dossier de synchronisation de l’utilisateur.

Notez que cela suit les mêmes hypothèses de chemin que dans sync_with_user.


{ “type”: “function”, “function”: { “name”: “delete_from_user_sync_folder”, “description”: “Supprimera un fichier ou un répertoire du dossier de synchronisation sur l’ordinateur de l’utilisateur”, “parameters”: { “type”: “object”, “properties”: { “path_on_user_sync_folder”: { “type”: “string”, “description”: “Chemin relatif vers le fichier ou le répertoire dans le dossier de synchronisation de l’utilisateur. Par exemple, ‘/hello.txt’ le supprimera directement du dossier de synchronisation, tandis que ‘/run1/hello.txt’ le supprimera directement du sous-dossier ‘run1’ du dossier de synchronisation.”, } }, “required”: [“path_on_user_sync_folder”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “delete_from_user_sync_folder”,
“description”: “Supprimera un fichier ou un répertoire du dossier de synchronisation sur l’ordinateur de l’utilisateur”,
“parameters”: {
“type”: “object”,
“properties”: {
“path_on_user_sync_folder”: {
“type”: “string”,
“description”: “Chemin relatif vers le fichier ou le répertoire dans le dossier de synchronisation de l’utilisateur. Par exemple, ‘/hello.txt’ le supprimera directement du dossier de synchronisation, tandis que ‘/run1/hello.txt’ le supprimera directement du sous-dossier ‘run1’ du dossier de synchronisation.”,
}
},
“required”: [“path_on_user_sync_folder”],
},
},
},

Implémentation des fonctions disponibles :

Tout d’abord, nous avons une classe SandboxEDA qui prend les paramètres suivants :

sandbox : l’instance du sandbox.
model_base_url et model_api_key : pour se connecter au modèle Novita.
max_consecutive_function_calls_allowed : par défaut à 30 pour éviter les boucles infinies d’appels de fonctions de l’agent, comme nous le verrons plus tard.


… # en dessous du code existant class SandboxEDA: def __init__( self, sandbox: Sandbox, model_api_base_url: str, model_api_key: str, max_consecutive_function_calls_allowed: int = 30, ): self.sandbox = sandbox self.model_api_base_url = model_api_base_url self.model_api_key = model_api_key self.max_consecutive_function_calls_allowed = ( max_consecutive_function_calls_allowed )

Ensuite, run_python_code en tant que méthode de la classe SandboxEDA. La méthode prend le code Python en entrée et utilise l’instance sandbox pour l’exécuter. Lorsque les sorties sont renvoyées, toutes les sorties d’image (note : elles sont encodées en base64) sont enregistrées dans le répertoire temp_image_output. Enfin, la méthode renvoie un dictionnaire contenant les sorties d’image, les autres sorties, les journaux et toutes les erreurs.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def run_python_code(self, python_code: str) -> dict: “”“ Exécute le code python sur le sandbox, et s’il y a des images, les enregistre localement. Args : python_code (str) : Le code python à exécuter. Retourne : dict : Contenant les sorties d’image en base64 et les autres sorties (stdout, journaux, erreur, etc.). ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # Parcourt les images encodées en base64 et les enregistre dans un fichier au format : temp-{timestamp}.png dans le répertoire ./temp_image_output for b64_image in image_outputs: timestamp = int(time.time_ns()) image_filename = Path(f"./temp_image_output/temp-{timestamp}.png") # Crée le répertoire temp_image_output s’il n’existe pas déjà. image_filename.parent.mkdir(parents=True, exist_ok=True) with open(image_filename, “wb”) as f: f.write(base64.b64decode(b64_image)) return { “image_outputs”: image_outputs, “other_outputs”: { “outputs”: [result for result in execution.results if not result.png], “logs”: execution.logs, “error”: execution.error, }, }

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def run_python_code(self, python_code: str) -> dict:
“”“
Exécute le code python sur le sandbox, et s’il y a des images, les enregistre localement.

Args :
python_code (str) : Le code python à exécuter.

Retourne :
dict : Contenant les sorties d’image en base64 et les autres sorties (stdout, journaux, erreur, etc.).
”“”
execution = self.sandbox.run_code(python_code, language=“python”)

image_outputs = [result.png for result in execution.results if result.png]

# Parcourt les images encodées en base64 et les enregistre dans un fichier au format : temp-{timestamp}.png dans le répertoire ./temp_image_output
for b64_image in image_outputs:
timestamp = int(time.time_ns())
image_filename = Path(f"./temp_image_output/temp-{timestamp}.png")

# Crée le répertoire temp_image_output s’il n’existe pas déjà.
image_filename.parent.mkdir(parents=True, exist_ok=True)

with open(image_filename, “wb”) as f:
f.write(base64.b64decode(b64_image))

return {
“image_outputs”: image_outputs,
“other_outputs”: {
“outputs”: [result for result in execution.results if not result.png],
“logs”: execution.logs,
“error”: execution.error,
},
}

Ensuite, la méthode run_on_command_line. Celle-ci exécutera de la même manière la commande sur l’instance du sandbox, puis renverra un dictionnaire avec les sorties. Si l’exécution échoue, elle renvoie un dictionnaire avec None pour les sorties et définit “execution error” sur le message d’erreur.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def run_on_command_line(self, command: str) -> dict: “”“ Exécute la commande sur le sandbox. Args : command (str) : La commande à exécuter. Retourne : dict : Contenant la sortie de la commande et l’erreur d’exécution le cas échéant. ”“” try: result = self.sandbox.commands.run(command) return { “output”: { “stdout”: result.stdout, “stderr”: result.stderr, “exit_code”: result.exit_code, “error”: result.error, }, “execution error”: None, } except Exception as e: return {“output”: None, “execution error”: str(e)}

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def run_on_command_line(self, command: str) -> dict:
“”“
Exécute la commande sur le sandbox.

Args :
command (str) : La commande à exécuter.

Retourne :
dict : Contenant la sortie de la commande et l’erreur d’exécution le cas échéant.
”“”

try:
result = self.sandbox.commands.run(command)
return {
“output”: {
“stdout”: result.stdout,
“stderr”: result.stderr,
“exit_code”: result.exit_code,
“error”: result.error,
},
“execution error”: None,
}

except Exception as e:
return {“output”: None, “execution error”: str(e)}

Et la méthode sync_with_user. Comme décrit précédemment, elle prendra le sandbox_path et le path_on_user_sync_folder. Si le chemin du sandbox pointe vers un fichier, elle télécharge le fichier depuis le sandbox vers les emplacements correspondants du dossier de synchronisation. S’il s’agit d’un répertoire, la méthode parcourt récursivement tout le contenu descendant et télécharge chaque fichier vers ses emplacements correspondants. En cas de succès, nous renvoyons “Sync Successful”, sinon nous renvoyons le message d’exception.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def sync_with_user(self, sandbox_path, path_on_user_sync_folder): “”“ Télécharge un fichier ou un répertoire depuis le sandbox vers le dossier de synchronisation de l’utilisateur. Args : sandbox_path (str) : Le chemin du fichier ou du répertoire à synchroniser dans le sandbox. path_on_user_sync_folder (str) : Le chemin de destination relatif du fichier ou du répertoire dans le dossier de synchronisation de l’utilisateur. Retourne : str : “Sync Successful” si le fichier ou le répertoire a été synchronisé avec succès, sinon un message d’erreur. ”“” try: path_info = self.sandbox.files.get_info(sandbox_path) if path_info.type == FileType.DIR: # Si c’est un répertoire, parcourt le contenu et télécharge les fichiers. dir_contents = self.sandbox.files.list(sandbox_path) for content in dir_contents: path_to_content_in_sync_folder = Path( path_on_user_sync_folder ).joinpath(content.name) self.sync_with_user(content.path, path_to_content_in_sync_folder) elif path_info.type == FileType.FILE: # Vérifie que le fichier est toujours dans ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Rend le chemin relatif en supprimant tout composant racine ou de lecteur relative_path = sandbox_path_obj.relative_to( sandbox_path_obj.anchor or “.” ) # Chemin final dans sync_folder file_path = Path(“sync_folder”) / relative_path # Crée tout répertoire dans le chemin qui n’existe pas déjà. file_path.parent.mkdir(parents=True, exist_ok=True) # Télécharge le fichier dans le dossier de synchronisation. file_content = self.sandbox.files.read(sandbox_path, “bytes”) with open(file_path, “wb”) as f: f.write(file_content) return “Sync Successful” except Exception as e: return str(e)

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def sync_with_user(self, sandbox_path, path_on_user_sync_folder):
“”“
Télécharge un fichier ou un répertoire depuis le sandbox vers le dossier de synchronisation de l’utilisateur.

Args :
sandbox_path (str) : Le chemin du fichier ou du répertoire à synchroniser dans le sandbox.
path_on_user_sync_folder (str) : Le chemin de destination relatif du fichier ou du répertoire dans le dossier de synchronisation de l’utilisateur.

Retourne :
str : “Sync Successful” si le fichier ou le répertoire a été synchronisé avec succès, sinon un message d’erreur.
”“”

try:
path_info = self.sandbox.files.get_info(sandbox_path)

if path_info.type == FileType.DIR:
# Si c’est un répertoire, parcourt le contenu et télécharge les fichiers.
dir_contents = self.sandbox.files.list(sandbox_path)
for content in dir_contents:
path_to_content_in_sync_folder = Path(
path_on_user_sync_folder
).joinpath(content.name)
self.sync_with_user(content.path, path_to_content_in_sync_folder)

elif path_info.type == FileType.FILE:
# Vérifie que le fichier est toujours dans ./sync_folder.
sandbox_path_obj = Path(path_on_user_sync_folder)

# Rend le chemin relatif en supprimant tout composant racine ou de lecteur
relative_path = sandbox_path_obj.relative_to(
sandbox_path_obj.anchor or “.”
)

# Chemin final dans sync_folder
file_path = Path(“sync_folder”) / relative_path

# Crée tout répertoire dans le chemin qui n’existe pas déjà.
file_path.parent.mkdir(parents=True, exist_ok=True)

# Télécharge le fichier dans le dossier de synchronisation.
file_content = self.sandbox.files.read(sandbox_path, “bytes”)
with open(file_path, “wb”) as f:
f.write(file_content)

return “Sync Successful”

except Exception as e:
return str(e)

Enfin, la méthode delete_from_user_sync_folder. Comme déjà décrit, elle prend également path_on_user_sync_folder. Nous supprimons ensuite le fichier ou le répertoire correspondant s’il existe et renvoyons “Deletion Successful”, sinon nous renvoyons le message d’exception.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def delete_from_user_sync_folder(self, path_on_user_sync_folder): “”“ Supprime un fichier ou un répertoire du dossier de synchronisation de l’utilisateur. Args : path_on_user_sync_folder (str) : Le chemin du fichier ou du répertoire à supprimer dans le dossier de synchronisation de l’utilisateur. Retourne : str : “Deletion Successful” si le fichier ou le répertoire a été supprimé avec succès, sinon un message d’erreur. ”“” # Vérifie que le fichier est toujours dans ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Rend le chemin relatif en supprimant tout composant racine ou de lecteur relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”) # Chemin final dans sync_folder delete_path = Path(“sync_folder”) / relative_path try: if not delete_path.exists(): raise Exception( f"Fichier ou répertoire n’existe pas au chemin {path_on_user_sync_folder} dans le dossier de synchronisation." ) if delete_path.is_file(): delete_path.unlink() elif delete_path.is_dir(): shutil.rmtree(str(delete_path)) return “Deletion Successful” except Exception as e: return str(e)

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def delete_from_user_sync_folder(self, path_on_user_sync_folder):
“”“
Supprime un fichier ou un répertoire du dossier de synchronisation de l’utilisateur.

Args :
path_on_user_sync_folder (str) : Le chemin du fichier ou du répertoire à supprimer dans le dossier de synchronisation de l’utilisateur.

Retourne :
str : “Deletion Successful” si le fichier ou le répertoire a été supprimé avec succès, sinon un message d’erreur.
”“”
# Vérifie que le fichier est toujours dans ./sync_folder.
sandbox_path_obj = Path(path_on_user_sync_folder)

# Rend le chemin relatif en supprimant tout composant racine ou de lecteur
relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”)

# Chemin final dans sync_folder
delete_path = Path(“sync_folder”) / relative_path

try:
if not delete_path.exists():
raise Exception(
f"Fichier ou répertoire n’existe pas au chemin {path_on_user_sync_folder} dans le dossier de synchronisation."
)

if delete_path.is_file():
delete_path.unlink()

elif delete_path.is_dir():
shutil.rmtree(str(delete_path))

return “Deletion Successful”

except Exception as e:
return str(e)

Autres fonctions du sandbox :

1. Méthode de téléchargement de fichiers vers le sandbox.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def upload_files_to_sandbox( self, file_paths: list[str], file_names_in_sandbox: list[str] ): “”“ Télécharge des fichiers vers le sandbox. Args : file_paths (list[str]) : Chemins des fichiers à télécharger (ex : [”./Download/data.csv", “./Download/data2.csv”]). file_names_in_sandbox (list[str]) : Les noms que les fichiers prendront dans le sandbox (ex : [“data.csv”, “data2.csv”]). Note : Les fichiers seront téléchargés dans le répertoire /home/user du sandbox (ex : ./home/user/data.csv, ./home/user/data2.csv). “”“ console.print( f”[yellow]Téléchargement de(s) fichier(s) aux emplacements {file_paths} vers le Sandbox[/yellow] (id : {self.sandbox.sandbox_id})“ ) for file_path, file_name_in_sandbox in zip(file_paths, file_names_in_sandbox): with open(file_path, “rb”) as file: self.sandbox.files.write(file_name_in_sandbox, file) console.print( f”[bold cyan]Fichier(s) {file_paths} téléchargé(s) vers le Sandbox[/bold cyan] (id : {self.sandbox.sandbox_id})" )

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def upload_files_to_sandbox(
self, file_paths: list[str], file_names_in_sandbox: list[str]
):
“”“
Télécharge des fichiers vers le sandbox.

Args :
file_paths (list[str]) : Chemins des fichiers à télécharger (ex : [”./Download/data.csv", “./Download/data2.csv”]).
file_names_in_sandbox (list[str]) : Les noms que les fichiers prendront dans le sandbox (ex : [“data.csv”, “data2.csv”]).

Note :
Les fichiers seront téléchargés dans le répertoire /home/user du sandbox (ex : ./home/user/data.csv, ./home/user/data2.csv).
“”“

console.print(
f”[yellow]Téléchargement de(s) fichier(s) aux emplacements {file_paths} vers le Sandbox[/yellow] (id : {self.sandbox.sandbox_id})“
)

for file_path, file_name_in_sandbox in zip(file_paths, file_names_in_sandbox):
with open(file_path, “rb”) as file:
self.sandbox.files.write(file_name_in_sandbox, file)

console.print(
f”[bold cyan]Fichier(s) {file_paths} téléchargé(s) vers le Sandbox[/bold cyan] (id : {self.sandbox.sandbox_id})"
)

2. Méthode de listage du contenu du répertoire principal du sandbox (/home/user).


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def list_files_in_sandbox_main_dir(self) -> list[str]: return [i.name for i in self.sandbox.files.list(“/home/user”)]

Mise en place de l’interaction avec l’agent :

Nous allons maintenant voir une méthode pour interagir avec l’agent. Avant cela, examinons l’instruction de prompt système paramétrée dans prompts/system_prompt.py.


SYSTEM_PROMPT = “”“ You are an Exploratory Data Analysis (EDA) agent and you have access to a sandbox (with internet access) where you can: - Execute python code using the run_python_code function call. - You can basically do anything you can do on a linux machine via the run_on_command_line or run_python_code function call. - You can sync whatever directory (may be preferred for structure eg website) or file you have created, written to or updated to the user’s sync folder on their local machine through the sync_with_user function call. - You can delete any of those directory or file from the user’s sync folder on their local machine through the delete_from_user_sync_folder function call. Your current PWD is ‘/home/user’ and below are the files in it. {list_sandbox_files} Note: - The sandbox already comes pre-installed with the usual data analysis packages but if there’s a package you are not sure exists or your code had an import error due to a missing package, you can check if it’s installed and if not install it. - For image outputs (e.g from data visualization) make sure it is png format. Function Call Guidelines: - Always use run_python_code to perform any task unless you absolutely need to use run_on_command_line (e.g to install packages, etc) - Chain function calls when needed: After receiving results from one function call, immediately make additional calls if more information is required - Gather just the needed information first: Respond to the user only when you have at least enough information from function calls to provide a good answer - Be efficient: Although there is a maximum limit of {max_consecutive_function_calls_allowed} consecutive function calls try to make as less calls as possible to get just enough information. - Don’t just assume the user will read the output of the tool call respond to them with your answer. Be a helpful assistant to the user who is probably trying to perform EDA on dataset files ({downloaded_dataset_names}) at (/home/user/) directory. You can perform the following function calls: {available_function_calls_schema} ”“”

SYSTEM_PROMPT = “”“
You are an Exploratory Data Analysis (EDA) agent and you have access to a sandbox (with internet access) where you can:

- Execute python code using the run_python_code function call.
- You can basically do anything you can do on a linux machine via the run_on_command_line or run_python_code function call.
- You can sync whatever directory (may be preferred for structure eg website) or file you have created, written to or updated to the user’s sync folder on their local machine through the sync_with_user function call.
- You can delete any of those directory or file from the user’s sync folder on their local machine through the delete_from_user_sync_folder function call.

Your current PWD is ‘/home/user’ and below are the files in it.
{list_sandbox_files}

Note:
- The sandbox already comes pre-installed with the usual data analysis packages but if there’s a package you
are not sure exists or your code had an import error due to a missing package, you can check if it’s installed and if not install it.

- For image outputs (e.g from data visualization) make sure it is png format.

Function Call Guidelines:
- Always use run_python_code to perform any task unless you absolutely need to use run_on_command_line (e.g to install packages, etc)
- Chain function calls when needed: After receiving results from one function call, immediately make additional calls if more information is required
- Gather just the needed information first: Respond to the user only when you have at least enough information from function calls to provide a good answer
- Be efficient: Although there is a maximum limit of {max_consecutive_function_calls_allowed} consecutive function calls try to make as less calls as possible to get just enough information.
- Don’t just assume the user will read the output of the tool call respond to them with your answer.

Be a helpful assistant to the user who is probably trying to perform EDA on dataset files ({downloaded_dataset_names}) at (/home/user/) directory.

You can perform the following function calls:
{available_function_calls_schema}
”“”

Passons maintenant à la méthode eda_chat de la classe SandboxEDA. Cette méthode prend les noms des fichiers qui ont été téléchargés vers le sandbox (nous gérerons le téléchargement avant de démarrer la discussion, comme nous le verrons plus tard) et le nom du modèle Novita à utiliser.

1. Tout d’abord, nous configurons le client OpenAI pour pointer vers Novita via l’URL de base, et initialisons la conversation avec le prompt système comme premier message.


… # en dessous du code existant class SandboxEDA: … # en dessous du code existant def eda_chat( self, downloaded_dataset_names: list[str], model_for_eda: str, ): “”“ Session EDA interactive avec un agent IA capable d’exécution de code et de commandes terminales Args : downloaded_dataset_names (list[str]) : Les noms des jeux de données téléchargés. model_for_eda (str, optionnel) : Le modèle sous-jacent à utiliser. ”“” console.print( Panel( “[bold green]Session EDA démarrée[/bold green]\ Tapez ‘quit()’ pour quitter.”, title=“Analyse de données exploratoires”, border_style=“green”, ) ) client = OpenAI( base_url=self.model_api_base_url, api_key=self.model_api_key, ) # Initialise la conversation avec le prompt système messages = [ { “role”: “system”, “content”: SYSTEM_PROMPT.format( downloaded_dataset_names=str(downloaded_dataset_names), list_sandbox_files=str(self.list_files_in_sandbox_main_dir()), available_function_calls_schema=str( AVAILABLE_FUNCTION_CALL_SCHEMAS ), max_consecutive_function_calls_allowed=self.max_consecutive_function_calls_allowed, ), } ]

… # en dessous du code existant

class SandboxEDA:

… # en dessous du code existant

def eda_chat(
self,
downloaded_dataset_names: list[str],
model_for_eda: str,
):
“”“
Session EDA interactive avec un agent IA capable d’exécution de code et de commandes terminales

Args :
downloaded_dataset_names (list[str]) : Les noms des jeux de données téléchargés.
model_for_eda (str, optionnel) : Le modèle sous-jacent à utiliser.
”“”

console.print(
Panel(
“[bold green]Session EDA démarrée[/bold green]\ Tapez ‘quit()’ pour quitter.”,
title=“Analyse de données exploratoires”,
border_style=“green”,
)
)

client = OpenAI(
base_url=self.model_api_base_url,
api_key=self.model_api_key,
)

# Initialise la conversation avec le prompt système
messages = [
{
“role”: “system”,
“content”: SYSTEM_PROMPT.format(
downloaded_dataset_names=str(downloaded_dataset_names),
list_sandbox_files=str(self.list_files_in_sandbox_main_dir()),
available_function_calls_schema=str(
AVAILABLE_FUNCTION_CALL_SCHEMAS
),
max_consecutive_function_calls_allowed=self.max_consecutive_function_calls_allowed,
),
}
]

2. Ensuite, la boucle de discussion principale est une boucle while, dans laquelle nous prenons le message de l’utilisateur et l’ajoutons à l’historique des messages existant. Et pour éviter des appels d’outils consécutifs infinis, à l’intérieur de la boucle while, il y a une boucle for jusqu’à la limite, levant une exception lorsque celle-ci est atteinte. Le modèle est ensuite invité avec les messages.


… # en dessous du code existant # Boucle de discussion principale while True: user_input = Prompt.ask(“\ [bold yellow]>>> Message utilisateur[/bold yellow]”) if user_input.lower().strip() == “quit()”: break messages.append({“role”: “user”, “content”: user_input}) # Gère les appels d’outils consécutifs potentiels avec une limite de sécurité pour éviter les boucles infinies for i in range(self.max_consecutive_function_calls_allowed + 1): if i == self.max_consecutive_function_calls_allowed: raise Exception( f"Les appels d’outils consécutifs de l’agent ne doivent pas dépasser {self.max_consecutive_function_calls_allowed}." ) response = client.chat.completions.create( model=model_for_eda, messages=messages, tools=AVAILABLE_FUNCTION_CALL_SCHEMAS, frequency_penalty=0, ) response_message = response.choices[0].message

… # en dessous du code existant

# Boucle de discussion principale
while True:
user_input = Prompt.ask(“\ [bold yellow]>>> Message utilisateur[/bold yellow]”)
if user_input.lower().strip() == “quit()”:
break

messages.append({“role”: “user”, “content”: user_input})

# Gère les appels d’outils consécutifs potentiels avec une limite de sécurité pour éviter les boucles infinies
for i in range(self.max_consecutive_function_calls_allowed + 1):

if i == self.max_consecutive_function_calls_allowed:
raise Exception(
f"Les appels d’outils consécutifs de l’agent ne doivent pas dépasser {self.max_consecutive_function_calls_allowed}."
)

response = client.chat.completions.create(
model=model_for_eda,
messages=messages,
tools=AVAILABLE_FUNCTION_CALL_SCHEMAS,
frequency_penalty=0,
)

response_message = response.choices[0].message

3. Ensuite, nous vérifions si le modèle a décidé d’effectuer un appel d’outil. Si c’est le cas, nous exécutons l’outil en utilisant les arguments fournis par le modèle, affichons la sortie dans le terminal pour l’utilisateur (ou affichons une image si applicable), puis renvoyons la sortie au modèle, en commençant par l’appel d’outil run_python_code.


… # en dessous du code existant tool_calls = response_message.tool_calls if tool_calls: messages.append( response_message ) # Ajoute le message assistant qui a déclenché les appels d’outils # Exécute chaque appel d’outil demandé for tool_call in tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments) if name == “run_python_code”: console.print( Panel( args[“python_code”], title=“Agent exécutant du code Python”, border_style=“blue”, ) ) code_result = self.run_python_code(args[“python_code”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, “name”: name, # S’il y a des sorties d’image (par exemple, visualisation de données), comme il n’est pas encore possible de renvoyer des images # depuis un appel d’outil, nous informons simplement l’agent que l’image a été affichée à l’utilisateur. “content”: [ { “type”: “text”, “text”: ( f"LES IMAGES ONT DÉJÀ ÉTÉ AFFICHÉES À L’UTILISATEUR DANS LE TERMINAL ET ENREGISTRÉES DANS DES FICHIERS TEMPORAIRES ex : temp-{{timestamp}}.png sur l’ordinateur de l’utilisateur dans le répertoire ./temp_image_output, LES AUTRES SORTIES SONT CI-DESSOUS\ {code_result[‘other_outputs’]}“ if code_result[“image_outputs”] else f”{code_result[‘other_outputs’]}" ), } ], } ) display_sandbox_code_output(code_result)

… # en dessous du code existant

tool_calls = response_message.tool_calls

if tool_calls:

messages.append(
response_message
) # Ajoute le message assistant qui a déclenché les appels d’outils

# Exécute chaque appel d’outil demandé
for tool_call in tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)

if name == “run_python_code”:
console.print(
Panel(
args[“python_code”],
title=“Agent exécutant du code Python”,
border_style=“blue”,
)
)

code_result = self.run_python_code(args[“python_code”])
messages.append(
{
“tool_call_id”: tool_call.id,
“role”: “tool”,
“name”: name,
# S’il y a des sorties d’image (par exemple, visualisation de données), comme il n’est pas encore possible de renvoyer des images
# depuis un appel d’outil, nous informons simplement l’agent que l’image a été affichée à l’utilisateur.
“content”: [
{
“type”: “text”,
“text”: (
f"LES IMAGES ONT DÉJÀ ÉTÉ AFFICHÉES À L’UTILISATEUR DANS LE TERMINAL ET ENREGISTRÉES DANS DES FICHIERS TEMPORAIRES ex : temp-{{timestamp}}.png sur l’ordinateur de l’utilisateur dans le répertoire ./temp_image_output, LES AUTRES SORTIES SONT CI-DESSOUS\ {code_result[‘other_outputs’]}“
if code_result[“image_outputs”]
else f”{code_result[‘other_outputs’]}"
),
}
],
}
)

display_sandbox_code_output(code_result)

3b. Appel d’outil run_on_command_line.


… # en dessous du code existant elif name == “run_on_command_line”: console.print( Panel( args[“command”], title=“Agent exécutant une commande sur le terminal”, border_style=“blue”, ) ) command_result = self.run_on_command_line(args[“command”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil “name”: name, “content”: str(command_result), } ) display_sandbox_command_output(command_result)

… # en dessous du code existant

elif name == “run_on_command_line”:
console.print(
Panel(
args[“command”],
title=“Agent exécutant une commande sur le terminal”,
border_style=“blue”,
)
)

command_result = self.run_on_command_line(args[“command”])
messages.append(
{
“tool_call_id”: tool_call.id,
“role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil
“name”: name,
“content”: str(command_result),
}
)

display_sandbox_command_output(command_result)

3c. Appel d’outil sync_with_user.


… # en dessous du code existant elif name == “sync_with_user”: console.print( Panel( f"[bold yellow]L’agent a commencé la synchronisation de {args[‘sandbox_path’]} vers le dossier de synchronisation de l’utilisateur ({args[‘path_on_user_sync_folder’]})[/bold yellow]", title=“Synchronisation de fichiers”, border_style=“white”, ) ) sync_result = self.sync_with_user( args[“sandbox_path”], args[“path_on_user_sync_folder”] ) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil “name”: name, “content”: sync_result, } ) … # raccourci pour la brièveté

… # en dessous du code existant

elif name == “sync_with_user”:
console.print(
Panel(
f"[bold yellow]L’agent a commencé la synchronisation de {args[‘sandbox_path’]} vers le dossier de synchronisation de l’utilisateur ({args[‘path_on_user_sync_folder’]})[/bold yellow]",
title=“Synchronisation de fichiers”,
border_style=“white”,
)
)

sync_result = self.sync_with_user(
args[“sandbox_path”], args[“path_on_user_sync_folder”]
)
messages.append(
{
“tool_call_id”: tool_call.id,
“role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil
“name”: name,
“content”: sync_result,
}
)

… # raccourci pour la brièveté

3d. Appel d’outil delete_from_user_sync_folder et levée d’une erreur inconnue si des appels de fonction inexistants sont présents.


… # en dessous du code existant elif name == “delete_from_user_sync_folder”: console.print( Panel( f"[bold yellow]L’agent supprime le(s) fichier(s) du dossier de synchronisation de l’utilisateur ({args[‘path_on_user_sync_folder’]})[/bold yellow]“, title=“Synchronisation de fichiers”, border_style=“white”, ) ) delete_result = self.delete_from_user_sync_folder( args[“path_on_user_sync_folder”] ) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil “name”: name, “content”: delete_result, } ) … # raccourci pour la brièveté else: raise ValueError(f"Appel de fonction inconnu : {name}”)

… # en dessous du code existant

elif name == “delete_from_user_sync_folder”:
console.print(
Panel(
f"[bold yellow]L’agent supprime le(s) fichier(s) du dossier de synchronisation de l’utilisateur ({args[‘path_on_user_sync_folder’]})[/bold yellow]“,
title=“Synchronisation de fichiers”,
border_style=“white”,
)
)

delete_result = self.delete_from_user_sync_folder(
args[“path_on_user_sync_folder”]
)
messages.append(
{
“tool_call_id”: tool_call.id,
“role”: “tool”, # Indique que ce message provient de l’utilisation d’un outil
“name”: name,
“content”: delete_result,
}
)

… # raccourci pour la brièveté

else:
raise ValueError(f"Appel de fonction inconnu : {name}”)

4. Si la dernière réponse de l’agent n’est pas un appel d’outil, affichez sa réponse à l’utilisateur et sortez de la boucle de limite d’appels d’outils consécutifs.


… # en dessous du code existant else: # Pas d’appels d’outils, affiche simplement la réponse de l’assistant après l’avoir ajoutée aux messages. messages.append( {“role”: “assistant”, “content”: response_message.content} ) console.print( f"[bold green]>>> Réponse de l’assistant : {response_message.content} [/]" ) break

Orchestration du flux de l’agent :

Enfin, nous avons main.py qui sert de point d’entrée de notre application, reliant tous les éléments. À l’intérieur, la méthode start_eda lance une nouvelle session de sandbox. Le paramètre sandbox_timeout détermine la durée pendant laquelle le sandbox reste actif avant d’être automatiquement terminé ; pour notre démo, nous le définirons sur 900 secondes (≈15 minutes).

Après la création du sandbox, nous y téléchargeons des fichiers, puis lançons la méthode eda_chat pour commencer à interagir avec l’agent.


… # en dessous du code existant def start_eda( model_for_eda: str, dataset_paths: list[str], dataset_file_names: list[str], api_key_for_sandbox_and_model: str, model_api_base_url: str, sandbox_domain: str, sandbox_template: str, sandbox_timeout: int, ): with Sandbox( template=sandbox_template, api_key=api_key_for_sandbox_and_model, domain=sandbox_domain, timeout=sandbox_timeout, ) as sandbox: try: sandbox_eda = SandboxEDA( sandbox, model_api_base_url, api_key_for_sandbox_and_model ) console.print( f"[bold cyan]Sandbox démarré[/bold cyan] (id : {sandbox.sandbox_id})“ ) sandbox_eda.upload_files_to_sandbox(dataset_paths, dataset_file_names) sandbox_eda.eda_chat(dataset_file_names, model_for_eda) console.print( f”\ \ [bold cyan]------ Session EDA terminée pour le sandbox (id : {sandbox.sandbox_id}) ------[/]“ ) finally: console.print( f”[bold cyan]----- Sandbox fermé (id : {sandbox.sandbox_id})-----[/]\ " )

… # en dessous du code existant

def start_eda(
model_for_eda: str,
dataset_paths: list[str],
dataset_file_names: list[str],
api_key_for_sandbox_and_model: str,
model_api_base_url: str,
sandbox_domain: str,
sandbox_template: str,
sandbox_timeout: int,
):

with Sandbox(
template=sandbox_template,
api_key=api_key_for_sandbox_and_model,
domain=sandbox_domain,
timeout=sandbox_timeout,
) as sandbox:

try:
sandbox_eda = SandboxEDA(
sandbox, model_api_base_url, api_key_for_sandbox_and_model
)

console.print(
f"[bold cyan]Sandbox démarré[/bold cyan] (id : {sandbox.sandbox_id})“
)

sandbox_eda.upload_files_to_sandbox(dataset_paths, dataset_file_names)

sandbox_eda.eda_chat(dataset_file_names, model_for_eda)

console.print(
f”\ \ [bold cyan]------ Session EDA terminée pour le sandbox (id : {sandbox.sandbox_id}) ------[/]“
)
finally:
console.print(
f”[bold cyan]----- Sandbox fermé (id : {sandbox.sandbox_id})-----[/]\ "
)

Vous trouverez ci-dessous la méthode main, point de départ de notre application :


… # en dessous du code existant async def main( api_key_for_sandbox_and_model: str, model_api_base_url: str, model_for_browser_agent: str, enable_vision_for_browser_agent: bool, model_for_eda: str, sandbox_domain: str, sandbox_template: str, sandbox_timeout_seconds: int, ): while True: # Bannière de bienvenue console.print( Panel( “[bold white]Bienvenue dans l’analyse de données exploratoires agentique[/bold white]\ \ “ ”[grey]Comment souhaitez-vous procéder :[/grey]\ “ ”[grey]1.[/grey] Télécharger un jeu de données d’abord.\ “ ”[grey]2.[/grey] Procéder avec un jeu de données déjà téléchargé.\ “ ”[grey]3.[/grey] Quitter”, title=“MENU PRINCIPAL”, border_style=“green”, width=70, ) ) choice = Prompt.ask( “\ [bold yellow]Entrez votre choix[/bold yellow]”, choices=[“1”, “2”, “3”] ).strip() if choice == “1”: result = await choice_download_dataset( api_key_for_sandbox_and_model, model_api_base_url, model_for_browser_agent, enable_vision_for_browser_agent, ) if result: download_path, filenames = result DATASET_PATHS = [ str(Path(download_path) / filename) for filename in filenames ] DATASET_FILE_NAMES = filenames else: continue # L’utilisateur est retourné au menu principal elif choice == “2”: result = choice_proceed_with_already_downloaded_datasets() if result: DATASET_PATHS = result DATASET_FILE_NAMES = [os.path.basename(path) for path in result] else: continue # puisque l’utilisateur a cliqué pour revenir au menu principal. elif choice == “3”: break # Démarre la session EDA start_eda( model_for_eda, DATASET_PATHS, DATASET_FILE_NAMES, api_key_for_sandbox_and_model, model_api_base_url, sandbox_domain, sandbox_template, sandbox_timeout_seconds, )

… # en dessous du code existant

async def main(
api_key_for_sandbox_and_model: str,
model_api_base_url: str,
model_for_browser_agent: str,
enable_vision_for_browser_agent: bool,
model_for_eda: str,
sandbox_domain: str,
sandbox_template: str,
sandbox_timeout_seconds: int,
):

while True:

# Bannière de bienvenue
console.print(
Panel(
“[bold white]Bienvenue dans l’analyse de données exploratoires agentique[/bold white]\ \ “
”[grey]Comment souhaitez-vous procéder :[/grey]\ “
”[grey]1.[/grey] Télécharger un jeu de données d’abord.\ “
”[grey]2.[/grey] Procéder avec un jeu de données déjà téléchargé.\ “
”[grey]3.[/grey] Quitter”,
title=“MENU PRINCIPAL”,
border_style=“green”,
width=70,
)
)

choice = Prompt.ask(
“\ [bold yellow]Entrez votre choix[/bold yellow]”, choices=[“1”, “2”, “3”]
).strip()

if choice == “1”:
result = await choice_download_dataset(
api_key_for_sandbox_and_model,
model_api_base_url,
model_for_browser_agent,
enable_vision_for_browser_agent,
)
if result:
download_path, filenames = result
DATASET_PATHS = [
str(Path(download_path) / filename) for filename in filenames
]
DATASET_FILE_NAMES = filenames
else:
continue # L’utilisateur est retourné au menu principal

elif choice == “2”:
result = choice_proceed_with_already_downloaded_datasets()
if result:
DATASET_PATHS = result
DATASET_FILE_NAMES = [os.path.basename(path) for path in result]
else:
continue # puisque l’utilisateur a cliqué pour revenir au menu principal.

elif choice == “3”:
break

# Démarre la session EDA
start_eda(
model_for_eda,
DATASET_PATHS,
DATASET_FILE_NAMES,
api_key_for_sandbox_and_model,
model_api_base_url,
sandbox_domain,
sandbox_template,
sandbox_timeout_seconds,
)

Comme main.py sera exécuté en tant que script, nous ajoutons le code ci-dessous, en passant également nos variables d’environnement, le délai d’expiration du sandbox et les modèles Novita que nous avons l’intention d’utiliser :


… # en dessous du code existant if __name__ == “__main__”: NOVITA_API_KEY = os.getenv(“NOVITA_API_KEY”) NOVITA_BASE_URL = os.getenv(“NOVITA_BASE_URL”) NOVITA_E2B_DOMAIN = os.getenv(“NOVITA_E2B_DOMAIN”) NOVITA_E2B_TEMPLATE = os.getenv(“NOVITA_E2B_TEMPLATE”) NOVITA_MODEL_FOR_BROWSER_AGENT = “qwen/qwen3-coder-480b-a35b-instruct” ENABLE_VISION_FOR_BROWSER_AGENT = ( False # Si vrai, assurez-vous que le modèle de l’agent de navigateur a des capacités visuelles. ) NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct” NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900 secondes (15 minutes), l’instance du sandbox sera tuée automatiquement après. asyncio.run( main( NOVITA_API_KEY, NOVITA_BASE_URL, NOVITA_MODEL_FOR_BROWSER_AGENT, ENABLE_VISION_FOR_BROWSER_AGENT, NOVITA_MODEL_FOR_EDA, NOVITA_E2B_DOMAIN, NOVITA_E2B_TEMPLATE, NOVITA_SANDBOX_TIMEOUT_SECONDS, ) )

… # en dessous du code existant

if __name__ == “__main__”:
NOVITA_API_KEY = os.getenv(“NOVITA_API_KEY”)
NOVITA_BASE_URL = os.getenv(“NOVITA_BASE_URL”)
NOVITA_E2B_DOMAIN = os.getenv(“NOVITA_E2B_DOMAIN”)
NOVITA_E2B_TEMPLATE = os.getenv(“NOVITA_E2B_TEMPLATE”)
NOVITA_MODEL_FOR_BROWSER_AGENT = “qwen/qwen3-coder-480b-a35b-instruct”
ENABLE_VISION_FOR_BROWSER_AGENT = (
False # Si vrai, assurez-vous que le modèle de l’agent de navigateur a des capacités visuelles.
)
NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct”
NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900 secondes (15 minutes), l’instance du sandbox sera tuée automatiquement après.

asyncio.run(
main(
NOVITA_API_KEY,
NOVITA_BASE_URL,
NOVITA_MODEL_FOR_BROWSER_AGENT,
ENABLE_VISION_FOR_BROWSER_AGENT,
NOVITA_MODEL_FOR_EDA,
NOVITA_E2B_DOMAIN,
NOVITA_E2B_TEMPLATE,
NOVITA_SANDBOX_TIMEOUT_SECONDS,
)
)

Test de l’exécution de notre agent EDA :

Exécutez la commande suivante dans le terminal


uv run main.py

Conclusion

Félicitations pour avoir créé votre agent analyste de données exploratoires. Vous pouvez maintenant lui demander de créer des sites web, des PowerPoint, etc., en utilisant des analyses/informations provenant de n’importe quels fichiers de jeu de données, et les résultats seront synchronisés directement sur votre ordinateur local.

Pour un petit rappel, dans cet article, vous avez appris à créer un agent capable de recevoir des instructions, puis d’utiliser le navigateur pour naviguer sur le web et télécharger des fichiers, d’exécuter du code et des commandes dans le Sandbox Novita, et de synchroniser des fichiers et des répertoires vers votre ordinateur local.

Ce n’est que la partie émergée de l’iceberg, vous pouvez étendre votre agent pour vous connecter à des bases de données, intégrer des outils comme Google Docs via MCP, et bien plus encore. Rendez-vous sur Novita pour donner vie à vos idées !

Novita AI est une plateforme cloud IA qui offre aux développeurs un moyen simple de déployer des modèles IA grâce à notre API simple, tout en fournissant un cloud GPU abordable et fiable pour construire et mettre à l’échelle.

Comment créer un agent avec le Sandbox Novita AI, les produits LLM et Browser Use.

Qu’est-ce qu’un agent IA ?

Voici les composants typiques d’un agent IA :

Présentation du Sandbox Novita AI, des produits LLM et de Browser Use