Cómo construir un agente con Novita AI Sandbox, productos LLM y uso del navegador.

Tabla de contenido

¿Qué es un agente de IA?
Descripción general de Novita AI Sandbox, productos LLM y uso del navegador
Configuración de tu entorno de desarrollo
Construcción del agente de análisis exploratorio de datos (EDA)
Conclusión

Recientemente, hemos visto un crecimiento exponencial en el campo de la inteligencia artificial con Modelos de Lenguaje Grande (LLM), Modelos de Lenguaje y Visión, etc., y más recientemente, agentes de IA. A diferencia de las aplicaciones tradicionales basadas en reglas, estos agentes son capaces de utilizar sus capacidades de razonamiento y toma de decisiones para realizar tareas complejas de forma autónoma o con una intervención mínima del usuario. Para hacer esto, los agentes de IA a menudo necesitan generar y ejecutar código dinámicamente, o incluso controlar toda la máquina virtual, y Novita AI sandbox proporciona estas capacidades.

En este tutorial, crearemos un agente de IA que servirá como nuestro analista de datos, y al igual que un analista humano, podrá buscar y descargar un conjunto de datos usando el Navegador según nuestras instrucciones, y cuando se le solicite, analizar, visualizar y ejecutar código para responder con conocimientos extraídos para nosotros.

¿Qué es un agente de IA?

Un agente de IA es un programa de software que utiliza inteligencia artificial para razonar, hacer planes y llevar a cabo acciones para alcanzar objetivos específicos. Estos agentes pueden usar herramientas, como API, ejecución de código y búsquedas web, para recopilar información o realizar tareas. A veces, también colaboran con otros agentes, particularmente para objetivos más complejos, como Deep Research.

Estos son los componentes típicos de un agente de IA:

Razonamiento/Planificación: Cuando se le da un objetivo, el agente crea planes paso a paso para lograrlo. Este proceso puede implicar completar subtareas y recopilar información adicional para guiar sus próximas acciones. Esto es especialmente común para tareas complejas que toman más tiempo, y los modelos de razonamiento funcionan bien en estas situaciones.
Uso de herramientas: Los agentes de IA pueden llamar a herramientas y servicios externos. Estos pueden incluir funciones, API u otros recursos que amplían sus capacidades.
Coordinación: Múltiples agentes pueden trabajar juntos compartiendo responsabilidades, cada uno ayudando a alcanzar un objetivo compartido, complejo o a largo plazo.

Descripción general de Novita AI Sandbox, productos LLM y uso del navegador

Novita AI Sandbox

Qué es un sandbox y por qué es importante

Un sandbox es un entorno de ejecución seguro y aislado donde se puede ejecutar código no confiable sin afectar el sistema host. Básicamente es una computadora virtual ligera para que tu agente de IA ejecute código, comandos, cree archivos, etc.

Novita AI proporciona este sandbox en la nube para que tu agente lo acceda rápidamente bajo demanda, con facturación flexible por segundo basada en los recursos utilizados.

Características clave del sandbox de Novita:

Aislamiento seguro: Cada sandbox tiene su propio sistema de archivos y entorno aislados, protegiendo los datos y evitando interacciones no deseadas.
Inicio rápido: Las instancias de sandbox se inician en menos de ~200 ms de promedio, lo que lo hace ideal para escenarios de baja latencia.
Soporte multilenguaje: Puedes ejecutar código en múltiples lenguajes de programación, incluyendo Python, JavaScript, TypeScript y más.
Pausa y reanudación rápidas: Pausa el sandbox en cualquier momento y reanúdalo cuando lo necesites, con el estado del sistema de archivos y los procesos completamente restaurado.
Ejecución en segundo plano: Admite ejecución de tareas en segundo plano y es adecuado para escenarios que requieren esperar un resultado.

API de modelos de Novita:

Novita AI Models

Novita ofrece una vasta biblioteca de modelos de IA de código abierto de laboratorios de investigación líderes como OpenAI, Google, DeepSeek y Qwen, etc. Estos incluyen modelos para lenguaje, visión, audio, video y embeddings. Nuestros modelos de lenguaje también son totalmente compatibles con el SDK de OpenAI, por lo que cambiar de OpenAI a Novita solo requiere actualizar la URL base y la clave de API en tu cliente, luego seleccionar un modelo de Novita.


from openai import OpenAI client = OpenAI( base_url=“https://api.novita.ai/v3/openai”, api_key=“”, )

Descripción general del uso del navegador:

Browser Use es una biblioteca de Python de código abierto que permite a los agentes de IA interactuar con navegadores web usando comandos en lenguaje natural (por ejemplo, “Consultar el clima en NY hoy”). En lugar de escribir selectores complejos basados en reglas, la IA se encarga de encontrar elementos e interactuar con ellos. Y dado que la API de modelos de Novita es compatible con OpenAI, puedes usar cualquier LLM o VLM (modelo de lenguaje y visión) para impulsar Browser Use.

Configuración de tu entorno de desarrollo

Para empezar, clonaremos el repositorio de GitHub, configuraremos un entorno de Python limpio, instalaremos todas las dependencias requeridas y obtendremos las claves de Novita AI.

Clona el repositorio de GitHub e instala las dependencias con uv.

1. Instala uv (un gestor de paquetes de Python ligero)


pip install uv

2. Clona el repositorio (repositorio de GitHub) y navega hasta él.


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

3. Crea y activa el entorno virtual de uv


# Crea un entorno virtual uv venv # Activa el entorno virtual source .venv/bin/activate # Para Mac/Linux # o .venv\Scripts\activate # Para Windows

4. Instala las dependencias del proyecto


# Instala las dependencias uv sync

Creación de una cuenta de Novita AI y obtención de una clave de API

1. Regístrate en novita.ai.

2. En el panel de control, pasa el cursor sobre el icono de perfil de usuario y haz clic en Claves de API en la ventana emergente.

3. En la página de Gestión de claves, haz clic en Agregar nueva clave. En la ventana emergente, ingresa un nombre para tu clave, haz clic en Confirmar y luego copia la clave generada.

4. Ahora dentro del directorio del proyecto, crea un archivo .env y pega el siguiente contenido.


NOVITA_API_KEY=“<PEGA TU CLAVE DE API DE NOVITA AQUÍ>” NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1”

Agrega créditos a tu cuenta de Novita AI.

Para usar el sandbox de Novita, necesitas agregar créditos a tu cuenta. En la pestaña del panel de control, haz clic en ‘Facturación’. Luego, en la página de facturación, agrega un método de pago y carga al menos $10 en créditos.

Construcción del agente de análisis exploratorio de datos (EDA)

Ahora que nuestro entorno está configurado y tenemos nuestra clave de API, vamos a poner en marcha nuestro agente.

Descarga de conjuntos de datos (o archivos) mediante el agente de navegador:

Al igual que un analista humano, queremos que nuestro agente pueda tomar instrucciones y usar un navegador para obtener los conjuntos de datos. Entonces, como se ve en browser_agent.py, primero creamos una estructura de datos Pydantic para la salida final del agente, que tomará los nombres de los archivos descargados y los resultados de las tareas completadas para escribirlos en un archivo.


… # código existente debajo class TaskFile(BaseModel): “”“Representa un archivo generado como parte de un resultado de tarea, por ejemplo datos extraídos o investigados.”“” filename: str = Field(…, description=“Nombre del archivo incluyendo la extensión”) content: str = Field(…, description=“Contenido de texto a escribir en el archivo”) class AgentOutput(BaseModel): “”“Salida agregada final de la ejecución del agente de navegador.”“” downloaded_files: Optional[list[str]] = Field( None, description=“Lista de nombres de archivos descargados (con extensión), si los hay” ) task_files: Optional[list[TaskFile]] = Field( None, description=“Archivos generados a partir de tareas del usuario (por ejemplo, datos extraídos o investigados), si los hay” )

… # código existente debajo
class TaskFile(BaseModel):
“”“Representa un archivo generado como parte de un resultado de tarea, por ejemplo datos extraídos o investigados.”“”

filename: str = Field(…, description=“Nombre del archivo incluyendo la extensión”)
content: str = Field(…, description=“Contenido de texto a escribir en el archivo”)

class AgentOutput(BaseModel):
“”“Salida agregada final de la ejecución del agente de navegador.”“”

downloaded_files: Optional[list[str]] = Field(
None, description=“Lista de nombres de archivos descargados (con extensión), si los hay”
)
task_files: Optional[list[TaskFile]] = Field(
None, description=“Archivos generados a partir de tareas del usuario (por ejemplo, datos extraídos o investigados), si los hay”
)

A continuación, se muestra un ejemplo de cómo funcionará el agente de navegador. Usaremos un Modelo de Lenguaje Grande (LLM) para la navegación web, lo que debería cubrir la mayoría de los casos de uso. Sin embargo, para tareas de interfaz de usuario más complejas (por ejemplo, resolución de captcha), se recomienda un Modelo de Lenguaje y Visión (VLM). Al usar un VLM, puedes establecer el nivel de detalle de visión (‘alto’, ‘bajo’ o ‘automático’) para equilibrar el costo de tokens con la claridad de la visión.

Luego creamos una sesión de navegador, configuramos el perfil con una ruta de descarga (./Download) y user_data_dir (establecido en None para el modo incógnito), y establecemos nuestro modelo Pydantic como controlador para obtener las salidas del agente estructuradas. Luego, iniciaremos el agente con await agent.run(), y los resultados finales se analizan para obtener los nombres de los archivos descargados.


# Ejemplo de cómo se verá el agente de navegador agent = Agent( task=“Ve a Hugging Face, busca An-j96/SuperstoreData y abre su página, luego navega a la pestaña Archivos y descarga el archivo csv de datos.”, llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key), use_vision=False, vision_detail_level=“auto”, # opciones disponibles [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=“./Download”, user_data_dir=None ) ), # establece la ruta del directorio de descarga para el navegador. controller=Controller( output_model=AgentOutput ), # Hace que el agente devuelva el nombre del archivo descargado al final de la tarea. ) all_results = await agent.run() final_output = AgentOutput.model_validate_json( all_results.final_result() ) # analiza el resultado final del agente.

# Ejemplo de cómo se verá el agente de navegador
agent = Agent(
task=“Ve a Hugging Face, busca An-j96/SuperstoreData y abre su página, luego navega a la pestaña Archivos y descarga el archivo csv de datos.”,
llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key),
use_vision=False,
vision_detail_level=“auto”, # opciones disponibles [‘low’, ‘high’, ‘auto’];
browser_session=BrowserSession(
browser_profile=BrowserProfile(
downloads_path=“./Download”, user_data_dir=None
)
), # establece la ruta del directorio de descarga para el navegador.
controller=Controller(
output_model=AgentOutput
), # Hace que el agente devuelva el nombre del archivo descargado al final de la tarea.
)

all_results = await agent.run()
final_output = AgentOutput.model_validate_json(
all_results.final_result()
) # analiza el resultado final del agente.

(Nota: Ejecutar un agente de navegador en tu máquina local iniciará una instancia real del navegador para que puedas verlo en acción.) El método completo recopilará los nombres de los archivos descargados, escribirá los resultados de las tareas en un archivo y devolverá todas las rutas de archivos para cargarlas más tarde.


… # código existente debajo 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]]: “”“ Realizará la tarea de descarga del usuario mediante el uso del navegador y devolverá la ruta del directorio de descarga y los nombres de los archivos descargados. Devuelve: Tupla de (directorio_descarga, nombres_archivos_con_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, # Esta penalización puede afectar ligeramente el uso de herramientas; mantener en 0. ), use_vision=use_vision, vision_detail_level=“auto”, # opciones disponibles [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=download_dir_path, user_data_dir=None, # “./browser_user_data” ) ), # establece la ruta del directorio de descarga para el navegador. controller=Controller( output_model=AgentOutput ), # Hace que el agente devuelva el resultado de la ejecución de la tarea según el esquema de AgentOutput. max_failures=5, ) try: # Ejecuta el agente y estructura su salida 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]Escribiendo resultados de tareas en archivos…[/bold yellow] {final_output.task_files}“, title=“Resultados de tareas”, border_style=“white”, ) ) # Escribe cada resultado de tarea en un archivo en el directorio de descarga task_result_files: list[str] = [] for task_file in final_output.task_files or []: file_path = Path(task_file.filename) # Evita traversal de rutas o rutas absolutas inseguras if file_path.is_absolute() or “…” in file_path.parts: raise ValueError( f"El agente pasó una ruta de archivo no segura como nombre de archivo: {file_path}” ) # Apunta la ruta del archivo dentro del directorio de descarga file_path = Path(download_dir_path) / file_path # Asegura que el directorio de descarga exista, de lo contrario lo crea. file_path.parent.mkdir(parents=True, exist_ok=True) # Escribe el contenido del resultado de la tarea en un archivo 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]Resultados de tareas escritos en archivos:[/bold green] {task_result_files}“, title=“Resultados de tareas”, border_style=“white”, ) ) # Combina los archivos descargados con los archivos de resultados de tareas file_results = (final_output.downloaded_files or []) + task_result_files if file_results: console.print( Panel( f”[bold green]Archivos disponibles:[/bold green] {file_results} en {download_dir_path}“, title=“Archivos descargados”, border_style=“green”, ) ) else: raise RuntimeError(“No se descargaron ni escribieron archivos.”) except Exception as e: file_results = None console.print( Panel( f”[bold red]Error:[/bold red] {str(e)}\ ", title=“Error de ejecución”, border_style=“red”, ) ) return (download_dir_path, file_results)

… # código existente debajo
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]]:
“”“
Realizará la tarea de descarga del usuario mediante el uso del navegador y devolverá la ruta del directorio de descarga y los
nombres de los archivos descargados.

Devuelve:
Tupla de (directorio_descarga, nombres_archivos_con_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, # Esta penalización puede afectar ligeramente el uso de herramientas; mantener en 0.
),
use_vision=use_vision,
vision_detail_level=“auto”, # opciones disponibles [‘low’, ‘high’, ‘auto’];
browser_session=BrowserSession(
browser_profile=BrowserProfile(
downloads_path=download_dir_path,
user_data_dir=None, # “./browser_user_data”
)
), # establece la ruta del directorio de descarga para el navegador.
controller=Controller(
output_model=AgentOutput
), # Hace que el agente devuelva el resultado de la ejecución de la tarea según el esquema de AgentOutput.
max_failures=5,
)

try:
# Ejecuta el agente y estructura su salida
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]Escribiendo resultados de tareas en archivos…[/bold yellow] {final_output.task_files}“,
title=“Resultados de tareas”,
border_style=“white”,
)
)

# Escribe cada resultado de tarea en un archivo en el directorio de descarga
task_result_files: list[str] = []
for task_file in final_output.task_files or []:
file_path = Path(task_file.filename)

# Evita traversal de rutas o rutas absolutas inseguras
if file_path.is_absolute() or “…” in file_path.parts:
raise ValueError(
f"El agente pasó una ruta de archivo no segura como nombre de archivo: {file_path}”
)

# Apunta la ruta del archivo dentro del directorio de descarga
file_path = Path(download_dir_path) / file_path

# Asegura que el directorio de descarga exista, de lo contrario lo crea.
file_path.parent.mkdir(parents=True, exist_ok=True)

# Escribe el contenido del resultado de la tarea en un archivo
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]Resultados de tareas escritos en archivos:[/bold green] {task_result_files}“,
title=“Resultados de tareas”,
border_style=“white”,
)
)

# Combina los archivos descargados con los archivos de resultados de tareas
file_results = (final_output.downloaded_files or []) + task_result_files

if file_results:
console.print(
Panel(
f”[bold green]Archivos disponibles:[/bold green] {file_results} en {download_dir_path}“,
title=“Archivos descargados”,
border_style=“green”,
)
)
else:
raise RuntimeError(“No se descargaron ni escribieron archivos.”)

except Exception as e:
file_results = None
console.print(
Panel(
f”[bold red]Error:[/bold red] {str(e)}\ ",
title=“Error de ejecución”,
border_style=“red”,
)
)

return (download_dir_path, file_results)

La llamada al agente está envuelta en un bloque try-except, por lo que si se produce un error, se lanza una excepción y file_results se establece en None. Finalmente, el método devuelve tanto la download_dir_path como las rutas de file_results.

Definición de esquemas de las herramientas disponibles para nuestro agente:

Ahora que hemos terminado con el uso del navegador, es hora de definir los esquemas de las herramientas disponibles para nuestro agente EDA. Proporcionaremos cuatro herramientas:

run_python_code: permite al agente ejecutar código Python dentro del sandbox.
run_on_command_line: permite al agente ejecutar comandos en la terminal del sandbox (por ejemplo, instalar paquetes de Python).
sync_with_user: permite al agente sincronizar archivos y directorios creados o actualizados desde el sandbox a tu carpeta de sincronización local.
delete_from_user_sync_folder: permite al agente eliminar archivos o directorios de la carpeta de sincronización local.

Juntas, todas estas herramientas le dan al agente control total sobre la ejecución de código, el uso de la terminal y la sincronización de archivos entre el sandbox y tu sistema local.

En sandbox_eda.py, vemos el esquema de las herramientas:

run_python_code, que simplemente toma el código Python a ejecutar como parámetro de entrada y devuelve un resultado si lo hay.


{ “type”: “function”, “function”: { “name”: “run_python_code”, “description”: “Ejecuta el código python y devuelve el resultado si lo hay.”, “parameters”: { “type”: “object”, “properties”: { “python_code”: { “type”: “string”, “description”: “El código Python a ejecutar.”, } }, “required”: [“python_code”], }, }, },

run_on_command_line, que nuevamente simplemente toma el comando a ejecutar como parámetro de entrada y devuelve un resultado si lo hay.


{ “type”: “function”, “function”: { “name”: “run_on_command_line”, “description”: “Ejecuta el comando en la línea de comandos y devuelve el resultado si lo hay.”, “parameters”: { “type”: “object”, “properties”: { “command”: { “type”: “string”, “description”: “El comando a ejecutar en la línea de comandos.”, } }, “required”: [“command”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “run_on_command_line”,
“description”: “Ejecuta el comando en la línea de comandos y devuelve el resultado si lo hay.”,
“parameters”: {
“type”: “object”,
“properties”: {
“command”: {
“type”: “string”,
“description”: “El comando a ejecutar en la línea de comandos.”,
}
},
“required”: [“command”],
},
},
},

sync_with_user, que toma dos parámetros de entrada:
- sandbox_path: la ruta al archivo o directorio dentro del sandbox.
- path_on_user_sync_folder: la ruta que el agente quiere que el archivo o directorio tenga dentro de la carpeta de sincronización del usuario.

Esta segunda ruta será de la forma (por ejemplo, /nuevo.txt), con la suposición de que la carpeta de sincronización es la principal. Por ejemplo, más tarde resolveremos /nuevo.txt a carpeta_sincronizacion/nuevo.txt.


{ “type”: “function”, “function”: { “name”: “sync_with_user”, “description”: “Sincroniza un archivo o directorio del sandbox a la carpeta de sincronización del usuario en su computadora”, “parameters”: { “type”: “object”, “properties”: { “sandbox_path”: { “type”: “string”, “description”: “Ruta al archivo o directorio en el sandbox.”, }, “path_on_user_sync_folder”: { “type”: “string”, “description”: “Ruta relativa donde se colocará el archivo o directorio dentro de la carpeta de sincronización del usuario. Por ejemplo, ‘/hola.txt’ se coloca directamente en la carpeta de sincronización, mientras que ‘/ejecucion1/hola.txt’ se colocará en una subcarpeta ‘ejecucion1’ dentro de la carpeta de sincronización.”, }, }, “required”: [“sandbox_path”, “path_on_user_sync_folder”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “sync_with_user”,
“description”: “Sincroniza un archivo o directorio del sandbox a la carpeta de sincronización del usuario en su computadora”,
“parameters”: {
“type”: “object”,
“properties”: {
“sandbox_path”: {
“type”: “string”,
“description”: “Ruta al archivo o directorio en el sandbox.”,
},
“path_on_user_sync_folder”: {
“type”: “string”,
“description”: “Ruta relativa donde se colocará el archivo o directorio dentro de la carpeta de sincronización del usuario. Por ejemplo, ‘/hola.txt’ se coloca directamente en la carpeta de sincronización, mientras que ‘/ejecucion1/hola.txt’ se colocará en una subcarpeta ‘ejecucion1’ dentro de la carpeta de sincronización.”,
},
},
“required”: [“sandbox_path”, “path_on_user_sync_folder”],
},
},
},

delete_from_user_sync_folder, que toma solo un parámetro de entrada:
- path_on_user_sync_folder: la ruta al archivo o directorio que el agente quiere eliminar de la carpeta de sincronización del usuario.

Ten en cuenta que esto sigue las mismas suposiciones de ruta que en sync_with_user.


{ “type”: “function”, “function”: { “name”: “delete_from_user_sync_folder”, “description”: “Elimina un archivo o directorio de la carpeta de sincronización del usuario en su computadora”, “parameters”: { “type”: “object”, “properties”: { “path_on_user_sync_folder”: { “type”: “string”, “description”: “Ruta relativa al archivo o directorio en la carpeta de sincronización del usuario. Por ejemplo, ‘/hola.txt’ lo eliminará directamente de la carpeta de sincronización, mientras que ‘/ejecucion1/hola.txt’ lo eliminará directamente de la subcarpeta ‘ejecucion1’ dentro de la carpeta de sincronización.”, } }, “required”: [“path_on_user_sync_folder”], }, }, },

{
“type”: “function”,
“function”: {
“name”: “delete_from_user_sync_folder”,
“description”: “Elimina un archivo o directorio de la carpeta de sincronización del usuario en su computadora”,
“parameters”: {
“type”: “object”,
“properties”: {
“path_on_user_sync_folder”: {
“type”: “string”,
“description”: “Ruta relativa al archivo o directorio en la carpeta de sincronización del usuario. Por ejemplo, ‘/hola.txt’ lo eliminará directamente de la carpeta de sincronización, mientras que ‘/ejecucion1/hola.txt’ lo eliminará directamente de la subcarpeta ‘ejecucion1’ dentro de la carpeta de sincronización.”,
}
},
“required”: [“path_on_user_sync_folder”],
},
},
},

Implementación de las funciones disponibles:

Primero, tenemos una clase SandboxEDA que toma los siguientes parámetros:

sandbox: la instancia del sandbox.
model_base_url y model_api_key: para conectarse al modelo de Novita.
max_consecutive_function_calls_allowed: el valor predeterminado es 30 para evitar bucles infinitos de llamadas a funciones por parte del agente, como veremos más adelante.


… # código existente debajo 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 )

Luego, run_python_code como método de la clase SandboxEDA. El método toma el código Python como entrada y usa la instancia del sandbox para ejecutarlo. Cuando se devuelven las salidas, cualquier salida de imagen (nota: están codificadas en base64) se guarda en el directorio temp_image_output. Finalmente, el método devuelve un diccionario que contiene las salidas de imagen, otras salidas, registros y cualquier error.


… # código existente debajo class SandboxEDA: … # código existente debajo def run_python_code(self, python_code: str) -> dict: “”“ Ejecuta el código python en el sandbox, y si hay alguna imagen la guarda localmente. Args: python_code (str): El código python a ejecutar. Devuelve: dict: Contiene las salidas de imagen en base64 y otras salidas (stdout, registros, error, etc). ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # Itera a través de las imágenes codificadas en base64 y las guarda en un archivo con el formato de nombre: temp-{marca_de_tiempo}.png en el directorio ./temp_image_output for b64_image in image_outputs: timestamp = int(time.time_ns()) image_filename = Path(f"./temp_image_output/temp-{timestamp}.png") # Crea el directorio temp_image_output si no existe ya. 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, }, }

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def run_python_code(self, python_code: str) -> dict:
“”“
Ejecuta el código python en el sandbox, y si hay alguna imagen la guarda localmente.

Args:
python_code (str): El código python a ejecutar.

Devuelve:
dict: Contiene las salidas de imagen en base64 y otras salidas (stdout, registros, error, etc).
”“”
execution = self.sandbox.run_code(python_code, language=“python”)

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

# Itera a través de las imágenes codificadas en base64 y las guarda en un archivo con el formato de nombre: temp-{marca_de_tiempo}.png en el directorio ./temp_image_output
for b64_image in image_outputs:
timestamp = int(time.time_ns())
image_filename = Path(f"./temp_image_output/temp-{timestamp}.png")

# Crea el directorio temp_image_output si no existe ya.
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,
},
}

A continuación, el método run_on_command_line. Este igualmente ejecutará el comando en la instancia del sandbox, luego devolverá un diccionario con las salidas. Si la ejecución falla, devuelve un diccionario con None para las salidas y establece “error de ejecución” en el mensaje de error.


… # código existente debajo class SandboxEDA: … # código existente debajo def run_on_command_line(self, command: str) -> dict: “”“ Ejecuta el comando en el sandbox. Args: command (str): El comando a ejecutar. Devuelve: dict: Contiene la salida del comando y el error de ejecución si lo hay. ”“” 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)}

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def run_on_command_line(self, command: str) -> dict:
“”“
Ejecuta el comando en el sandbox.

Args:
command (str): El comando a ejecutar.

Devuelve:
dict: Contiene la salida del comando y el error de ejecución si lo hay.
”“”

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

Y el método sync_with_user. Como se describió anteriormente, tomará sandbox_path y path_on_user_sync_folder. Si la ruta del sandbox apunta a un archivo, descarga el archivo desde el sandbox a las ubicaciones correspondientes de la carpeta de sincronización. Si es un directorio, el método recorre recursivamente todo el contenido descendente y descarga cada archivo a sus ubicaciones correspondientes. Si tiene éxito, devolvemos “Sincronización exitosa” de lo contrario devolvemos el mensaje de excepción.


… # código existente debajo class SandboxEDA: … # código existente debajo def sync_with_user(self, sandbox_path, path_on_user_sync_folder): “”“ Descarga un archivo o directorio del sandbox a la carpeta de sincronización del usuario. Args: sandbox_path (str): La ruta del archivo o directorio a sincronizar en el sandbox. path_on_user_sync_folder (str): La ruta de destino relativa del archivo o directorio en la carpeta de sincronización del usuario. Devuelve: str: “Sincronización exitosa” si el archivo o directorio se sincronizó correctamente, de lo contrario un mensaje de error. ”“” try: path_info = self.sandbox.files.get_info(sandbox_path) if path_info.type == FileType.DIR: # Si es un directorio, recorre el contenido y lo descarga. 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: # Asegura que el archivo siempre esté dentro de ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Haz que la ruta sea relativa eliminando cualquier componente raíz o de unidad relative_path = sandbox_path_obj.relative_to( sandbox_path_obj.anchor or “.” ) # Ruta final dentro de sync_folder file_path = Path(“sync_folder”) / relative_path # Crea cualquier directorio en la ruta que no exista ya. file_path.parent.mkdir(parents=True, exist_ok=True) # Descarga el archivo a la carpeta de sincronización. file_content = self.sandbox.files.read(sandbox_path, “bytes”) with open(file_path, “wb”) as f: f.write(file_content) return “Sincronización exitosa” except Exception as e: return str(e)

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def sync_with_user(self, sandbox_path, path_on_user_sync_folder):
“”“
Descarga un archivo o directorio del sandbox a la carpeta de sincronización del usuario.

Args:
sandbox_path (str): La ruta del archivo o directorio a sincronizar en el sandbox.
path_on_user_sync_folder (str): La ruta de destino relativa del archivo o directorio en la carpeta de sincronización del usuario.

Devuelve:
str: “Sincronización exitosa” si el archivo o directorio se sincronizó correctamente, de lo contrario un mensaje de error.
”“”

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

if path_info.type == FileType.DIR:
# Si es un directorio, recorre el contenido y lo descarga.
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:
# Asegura que el archivo siempre esté dentro de ./sync_folder.
sandbox_path_obj = Path(path_on_user_sync_folder)

# Haz que la ruta sea relativa eliminando cualquier componente raíz o de unidad
relative_path = sandbox_path_obj.relative_to(
sandbox_path_obj.anchor or “.”
)

# Ruta final dentro de sync_folder
file_path = Path(“sync_folder”) / relative_path

# Crea cualquier directorio en la ruta que no exista ya.
file_path.parent.mkdir(parents=True, exist_ok=True)

# Descarga el archivo a la carpeta de sincronización.
file_content = self.sandbox.files.read(sandbox_path, “bytes”)
with open(file_path, “wb”) as f:
f.write(file_content)

return “Sincronización exitosa”

except Exception as e:
return str(e)

Finalmente, el método delete_from_user_sync_folder. También, como ya se describió, toma path_on_user_sync_folder. Luego eliminamos el archivo o directorio correspondiente si existe y devolvemos “Eliminación exitosa” de lo contrario devolvemos el mensaje de excepción.


… # código existente debajo class SandboxEDA: … # código existente debajo def delete_from_user_sync_folder(self, path_on_user_sync_folder): “”“ Elimina un archivo o directorio de la carpeta de sincronización del usuario. Args: path_on_user_sync_folder (str): La ruta del archivo o directorio a eliminar en la carpeta de sincronización del usuario. Devuelve: str: “Eliminación exitosa” si el archivo o directorio se eliminó correctamente, de lo contrario un mensaje de error. ”“” # Asegura que el archivo siempre esté dentro de ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Haz que la ruta sea relativa eliminando cualquier componente raíz o de unidad relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”) # Ruta final dentro de sync_folder delete_path = Path(“sync_folder”) / relative_path try: if not delete_path.exists(): raise Exception( f"El archivo o directorio no existe en {path_on_user_sync_folder} en la carpeta de sincronización." ) if delete_path.is_file(): delete_path.unlink() elif delete_path.is_dir(): shutil.rmtree(str(delete_path)) return “Eliminación exitosa” except Exception as e: return str(e)

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def delete_from_user_sync_folder(self, path_on_user_sync_folder):
“”“
Elimina un archivo o directorio de la carpeta de sincronización del usuario.

Args:
path_on_user_sync_folder (str): La ruta del archivo o directorio a eliminar en la carpeta de sincronización del usuario.

Devuelve:
str: “Eliminación exitosa” si el archivo o directorio se eliminó correctamente, de lo contrario un mensaje de error.
”“”
# Asegura que el archivo siempre esté dentro de ./sync_folder.
sandbox_path_obj = Path(path_on_user_sync_folder)

# Haz que la ruta sea relativa eliminando cualquier componente raíz o de unidad
relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”)

# Ruta final dentro de sync_folder
delete_path = Path(“sync_folder”) / relative_path

try:
if not delete_path.exists():
raise Exception(
f"El archivo o directorio no existe en {path_on_user_sync_folder} en la carpeta de sincronización."
)

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

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

return “Eliminación exitosa”

except Exception as e:
return str(e)

Otras funciones del sandbox:

1. Método para subir archivos al sandbox.


… # código existente debajo class SandboxEDA: … # código existente debajo def upload_files_to_sandbox( self, file_paths: list[str], file_names_in_sandbox: list[str] ): “”“ Sube archivos al sandbox. Args: file_paths (list[str]): Rutas de los archivos a subir (por ejemplo, [”./Download/data.csv", “./Download/data2.csv”]). file_names_in_sandbox (list[str]): Los nombres que tomarán los archivos en el sandbox (por ejemplo, [“data.csv”, “data2.csv”]). Nota: Los archivos se subirán al directorio /home/user del sandbox (por ejemplo, ./home/user/data.csv, ./home/user/data2.csv). “”“ console.print( f”[yellow]Subiendo archivo(s) en {file_paths} al 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]Archivo(s) {file_paths} subidos al Sandbox[/bold cyan] (id: {self.sandbox.sandbox_id})" )

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def upload_files_to_sandbox(
self, file_paths: list[str], file_names_in_sandbox: list[str]
):
“”“
Sube archivos al sandbox.

Args:
file_paths (list[str]): Rutas de los archivos a subir (por ejemplo, [”./Download/data.csv", “./Download/data2.csv”]).
file_names_in_sandbox (list[str]): Los nombres que tomarán los archivos en el sandbox (por ejemplo, [“data.csv”, “data2.csv”]).

Nota:
Los archivos se subirán al directorio /home/user del sandbox (por ejemplo, ./home/user/data.csv, ./home/user/data2.csv).
“”“

console.print(
f”[yellow]Subiendo archivo(s) en {file_paths} al 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]Archivo(s) {file_paths} subidos al Sandbox[/bold cyan] (id: {self.sandbox.sandbox_id})"
)

2. Método para listar el contenido del directorio principal del sandbox (/home/user).


… # código existente debajo class SandboxEDA: … # código existente debajo def list_files_in_sandbox_main_dir(self) -> list[str]: return [i.name for i in self.sandbox.files.list(“/home/user”)]

Construcción de la interacción con el agente:

Ahora veremos un método para interactuar con el agente. Antes de eso, veamos la instrucción del system prompt parametrizado en prompts/system_prompt.py.


SYSTEM_PROMPT = “”“ Eres un agente de Análisis Exploratorio de Datos (EDA) y tienes acceso a un sandbox (con acceso a internet) donde puedes: - Ejecutar código python usando la llamada a la función run_python_code. - Básicamente puedes hacer cualquier cosa que puedas hacer en una máquina linux mediante la llamada a la función run_on_command_line o run_python_code. - Sincronizar cualquier directorio (puede ser preferible por estructura, por ejemplo un sitio web) o archivo que hayas creado, escrito o actualizado a la carpeta de sincronización del usuario en su máquina local mediante la llamada a la función sync_with_user. - Eliminar cualquier directorio o archivo de la carpeta de sincronización del usuario en su máquina local mediante la llamada a la función delete_from_user_sync_folder. Tu directorio de trabajo actual (PWD) es ‘/home/user’ y a continuación se muestran los archivos que contiene. {list_sandbox_files} Nota: - El sandbox ya viene preinstalado con los paquetes habituales de análisis de datos, pero si hay un paquete del que no estés seguro de que exista o tu código tuvo un error de importación por un paquete faltante, puedes verificar si está instalado y si no, instalarlo. - Para salidas de imagen (por ejemplo, de visualización de datos) asegúrate de que esté en formato png. Pautas para llamadas a funciones: - Usa siempre run_python_code para realizar cualquier tarea a menos que absolutely necesites usar run_on_command_line (por ejemplo, para instalar paquetes, etc.) - Encadena llamadas a funciones cuando sea necesario: después de recibir resultados de una llamada a función, realiza inmediatamente llamadas adicionales si se requiere más información - Recopila primero solo la información necesaria: Responde al usuario solo cuando tengas al menos suficiente información de las llamadas a función para proporcionar una buena respuesta - Sé eficiente: Aunque hay un límite máximo de {max_consecutive_function_calls_allowed} llamadas consecutivas a funciones, intenta hacer la menor cantidad de llamadas posible para obtener solo la información suficiente. - No asumas que el usuario leerá la salida de la llamada a la herramienta, respóndele con tu respuesta. Sé un asistente útil para el usuario que probablemente está intentando realizar un EDA en archivos de conjuntos de datos ({downloaded_dataset_names}) en el directorio (/home/user/). Puedes realizar las siguientes llamadas a funciones: {available_function_calls_schema} ”“”

SYSTEM_PROMPT = “”“
Eres un agente de Análisis Exploratorio de Datos (EDA) y tienes acceso a un sandbox (con acceso a internet) donde puedes:

- Ejecutar código python usando la llamada a la función run_python_code.
- Básicamente puedes hacer cualquier cosa que puedas hacer en una máquina linux mediante la llamada a la función run_on_command_line o run_python_code.
- Sincronizar cualquier directorio (puede ser preferible por estructura, por ejemplo un sitio web) o archivo que hayas creado, escrito o actualizado a la carpeta de sincronización del usuario en su máquina local mediante la llamada a la función sync_with_user.
- Eliminar cualquier directorio o archivo de la carpeta de sincronización del usuario en su máquina local mediante la llamada a la función delete_from_user_sync_folder.

Tu directorio de trabajo actual (PWD) es ‘/home/user’ y a continuación se muestran los archivos que contiene.
{list_sandbox_files}

Nota:
- El sandbox ya viene preinstalado con los paquetes habituales de análisis de datos, pero si hay un paquete del que no estés seguro de que exista o tu código tuvo un error de importación por un paquete faltante, puedes verificar si está instalado y si no, instalarlo.

- Para salidas de imagen (por ejemplo, de visualización de datos) asegúrate de que esté en formato png.

Pautas para llamadas a funciones:
- Usa siempre run_python_code para realizar cualquier tarea a menos que absolutely necesites usar run_on_command_line (por ejemplo, para instalar paquetes, etc.)
- Encadena llamadas a funciones cuando sea necesario: después de recibir resultados de una llamada a función, realiza inmediatamente llamadas adicionales si se requiere más información
- Recopila primero solo la información necesaria: Responde al usuario solo cuando tengas al menos suficiente información de las llamadas a función para proporcionar una buena respuesta
- Sé eficiente: Aunque hay un límite máximo de {max_consecutive_function_calls_allowed} llamadas consecutivas a funciones, intenta hacer la menor cantidad de llamadas posible para obtener solo la información suficiente.
- No asumas que el usuario leerá la salida de la llamada a la herramienta, respóndele con tu respuesta.

Sé un asistente útil para el usuario que probablemente está intentando realizar un EDA en archivos de conjuntos de datos ({downloaded_dataset_names}) en el directorio (/home/user/).

Puedes realizar las siguientes llamadas a funciones:
{available_function_calls_schema}
”“”

Ahora pasamos al método eda_chat en la clase SandboxEDA. Este método toma los nombres de los archivos que se han subido al sandbox (manejaremos la subida antes de iniciar el chat, como veremos más adelante) y el nombre del modelo de Novita a usar.

1. Primero, configuramos el cliente de OpenAI para apuntar a Novita a través de la URL base, e inicializamos la conversación con el system prompt como primer mensaje.


… # código existente debajo class SandboxEDA: … # código existente debajo def eda_chat( self, downloaded_dataset_names: list[str], model_for_eda: str, ): “”“ Sesión EDA interactiva con agente de IA capaz de ejecutar código y comandos de terminal Args: downloaded_dataset_names (list[str]): Los nombres de los conjuntos de datos descargados. model_for_eda (str, opcional): El modelo subyacente a usar. ”“” console.print( Panel( “[bold green]Sesión EDA iniciada[/bold green]\ Escribe ‘quit()’ para salir.”, title=“Análisis Exploratorio de Datos”, border_style=“green”, ) ) client = OpenAI( base_url=self.model_api_base_url, api_key=self.model_api_key, ) # Inicializa la conversación con el system prompt 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, ), } ]

… # código existente debajo

class SandboxEDA:

… # código existente debajo

def eda_chat(
self,
downloaded_dataset_names: list[str],
model_for_eda: str,
):
“”“
Sesión EDA interactiva con agente de IA capaz de ejecutar código y comandos de terminal

Args:
downloaded_dataset_names (list[str]): Los nombres de los conjuntos de datos descargados.
model_for_eda (str, opcional): El modelo subyacente a usar.
”“”

console.print(
Panel(
“[bold green]Sesión EDA iniciada[/bold green]\ Escribe ‘quit()’ para salir.”,
title=“Análisis Exploratorio de Datos”,
border_style=“green”,
)
)

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

# Inicializa la conversación con el system prompt
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. A continuación, el bucle principal del chat es un bucle while, en el que tomamos el mensaje del usuario y lo agregamos al historial de mensajes existente. Y para evitar llamadas consecutivas infinitas a herramientas, dentro del bucle while hay un bucle for hasta el límite, lanzando una excepción cuando se alcanza. Luego, el modelo es consultado con los mensajes.


… # código existente debajo # Bucle principal del chat while True: user_input = Prompt.ask(“\ [bold yellow]>>> Mensaje del usuario[/bold yellow]”) if user_input.lower().strip() == “quit()”: break messages.append({“role”: “user”, “content”: user_input}) # Maneja posibles llamadas consecutivas a herramientas con un límite de seguridad para evitar bucles infinitos for i in range(self.max_consecutive_function_calls_allowed + 1): if i == self.max_consecutive_function_calls_allowed: raise Exception( f"Las llamadas consecutivas a herramientas del agente no deben exceder {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

… # código existente debajo

# Bucle principal del chat
while True:
user_input = Prompt.ask(“\ [bold yellow]>>> Mensaje del usuario[/bold yellow]”)
if user_input.lower().strip() == “quit()”:
break

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

# Maneja posibles llamadas consecutivas a herramientas con un límite de seguridad para evitar bucles infinitos
for i in range(self.max_consecutive_function_calls_allowed + 1):

if i == self.max_consecutive_function_calls_allowed:
raise Exception(
f"Las llamadas consecutivas a herramientas del agente no deben exceder {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. A continuación, verificamos si el modelo decidió hacer una llamada a una herramienta. Si lo hizo, ejecutamos la herramienta usando los argumentos proporcionados por el modelo, imprimimos la salida en la terminal para el usuario (o mostramos una imagen si corresponde) y luego devolvemos la salida al modelo, comenzando con la llamada a la herramienta run_python_code.


… # código existente debajo tool_calls = response_message.tool_calls if tool_calls: messages.append( response_message ) # Agrega el mensaje del asistente que activó las llamadas a herramientas # Ejecuta cada llamada a herramienta solicitada 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=“Agente ejecutando código 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, # Si hay salidas de imagen (por ejemplo, visualización de datos), como aún no es posible devolver imágenes # desde una llamada a herramienta, solo informa al agente que la imagen se ha mostrado al usuario. “content”: [ { “type”: “text”, “text”: ( f"LAS IMÁGENES YA SE HAN MOSTRADO AL USUARIO EN LA TERMINAL Y SE HAN GUARDADO EN ARCHIVOS TEMPORALES, por ejemplo temp-{{timestamp}}.png en la computadora del usuario en el directorio ./temp_image_output, LAS OTRAS SALIDAS ESTÁN DEBAJO\ {code_result[‘other_outputs’]}“ if code_result[“image_outputs”] else f”{code_result[‘other_outputs’]}" ), } ], } ) display_sandbox_code_output(code_result)

… # código existente debajo

tool_calls = response_message.tool_calls

if tool_calls:

messages.append(
response_message
) # Agrega el mensaje del asistente que activó las llamadas a herramientas

# Ejecuta cada llamada a herramienta solicitada
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=“Agente ejecutando código 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,
# Si hay salidas de imagen (por ejemplo, visualización de datos), como aún no es posible devolver imágenes
# desde una llamada a herramienta, solo informa al agente que la imagen se ha mostrado al usuario.
“content”: [
{
“type”: “text”,
“text”: (
f"LAS IMÁGENES YA SE HAN MOSTRADO AL USUARIO EN LA TERMINAL Y SE HAN GUARDADO EN ARCHIVOS TEMPORALES, por ejemplo temp-{{timestamp}}.png en la computadora del usuario en el directorio ./temp_image_output, LAS OTRAS SALIDAS ESTÁN DEBAJO\ {code_result[‘other_outputs’]}“
if code_result[“image_outputs”]
else f”{code_result[‘other_outputs’]}"
),
}
],
}
)

display_sandbox_code_output(code_result)

3b. Llamada a la herramienta run_on_command_line.


… # código existente debajo elif name == “run_on_command_line”: console.print( Panel( args[“command”], title=“Agente ejecutando comando en la terminal”, border_style=“blue”, ) ) command_result = self.run_on_command_line(args[“command”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # Indica que este mensaje proviene del uso de una herramienta “name”: name, “content”: str(command_result), } ) display_sandbox_command_output(command_result)

… # código existente debajo

elif name == “run_on_command_line”:
console.print(
Panel(
args[“command”],
title=“Agente ejecutando comando en la terminal”,
border_style=“blue”,
)
)

command_result = self.run_on_command_line(args[“command”])
messages.append(
{
“tool_call_id”: tool_call.id,
“role”: “tool”, # Indica que este mensaje proviene del uso de una herramienta
“name”: name,
“content”: str(command_result),
}
)

display_sandbox_command_output(command_result)

3c. Llamada a la herramienta sync_with_user


… # código existente debajo elif name == “sync_with_user”: console.print( Panel( f"[bold yellow]Agente iniciando sincronización de {args[‘sandbox_path’]} a la carpeta de sincronización del usuario ({args[‘path_on_user_sync_folder’]})[/bold yellow]", title=“Sincronización de archivos”, 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”, # Indica que este mensaje proviene del uso de una herramienta “name”: name, “content”: sync_result, } ) … # omitido por brevedad

… # código existente debajo

elif name == “sync_with_user”:
console.print(
Panel(
f"[bold yellow]Agente iniciando sincronización de {args[‘sandbox_path’]} a la carpeta de sincronización del usuario ({args[‘path_on_user_sync_folder’]})[/bold yellow]",
title=“Sincronización de archivos”,
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”, # Indica que este mensaje proviene del uso de una herramienta
“name”: name,
“content”: sync_result,
}
)

… # omitido por brevedad

3d. Llamada a la herramienta delete_from_user_sync_folder y lanzar un error desconocido si hay llamadas a funciones que no existen.


… # código existente debajo elif name == “delete_from_user_sync_folder”: console.print( Panel( f"[bold yellow]Agente eliminando archivo(s) de la carpeta de sincronización del usuario ({args[‘path_on_user_sync_folder’]})[/bold yellow]“, title=“Sincronización de archivos”, 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”, # Indica que este mensaje proviene del uso de una herramienta “name”: name, “content”: delete_result, } ) … # omitido por brevedad else: raise ValueError(f"Llamada a función desconocida: {name}”)

… # código existente debajo

elif name == “delete_from_user_sync_folder”:
console.print(
Panel(
f"[bold yellow]Agente eliminando archivo(s) de la carpeta de sincronización del usuario ({args[‘path_on_user_sync_folder’]})[/bold yellow]“,
title=“Sincronización de archivos”,
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”, # Indica que este mensaje proviene del uso de una herramienta
“name”: name,
“content”: delete_result,
}
)

… # omitido por brevedad

else:
raise ValueError(f"Llamada a función desconocida: {name}”)

4. Si la última respuesta del agente no es una llamada a una herramienta, imprimimos su respuesta al usuario y salimos del bucle de límite de llamadas consecutivas a herramientas.


… # código existente debajo else: # No hay llamadas a herramientas, solo muestra la respuesta del asistente después de agregarla a los mensajes. messages.append( {“role”: “assistant”, “content”: response_message.content} ) console.print( f"[bold green]>>> Respuesta del asistente: {response_message.content} [/]" ) break

Orquestación del flujo del agente:

Finalmente, tenemos main.py que sirve como punto de entrada de nuestra aplicación, uniendo todo. Dentro, el método start_eda inicia una nueva sesión de sandbox. El parámetro sandbox_timeout determina cuánto tiempo permanece activo el sandbox antes de que se termine automáticamente; para nuestra demostración, lo estableceremos en 900 segundos (≈15 minutos).

Después de que se crea el sandbox, subimos archivos a él, luego lanzamos el método eda_chat para comenzar a interactuar con el agente.


… # código existente debajo 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 iniciado[/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]------ Sesión EDA completada para el Sandbox (id: {sandbox.sandbox_id}) ------[/]“ ) finally: console.print( f”[bold cyan]----- Sandbox cerrado (id: {sandbox.sandbox_id})-----[/bold cyan]\ " )

… # código existente debajo

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 iniciado[/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]------ Sesión EDA completada para el Sandbox (id: {sandbox.sandbox_id}) ------[/]“
)
finally:
console.print(
f”[bold cyan]----- Sandbox cerrado (id: {sandbox.sandbox_id})-----[/bold cyan]\ "
)

A continuación se muestra el método main que es el punto de inicio de nuestra aplicación:


… # código existente debajo 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: # Banner de bienvenida console.print( Panel( “[bold white]Bienvenido al Análisis Exploratorio de Datos Agéntico[/bold white]\ \ “ ”[grey]¿Cómo te gustaría proceder?:[/grey]\ “ ”[grey]1.[/grey] Descargar un conjunto de datos primero.\ “ ”[grey]2.[/grey] Proceder con un conjunto de datos ya descargado.\ “ ”[grey]3.[/grey] Salir”, title=“MENÚ PRINCIPAL”, border_style=“green”, width=70, ) ) choice = Prompt.ask( “\ [bold yellow]Ingresa tu elección[/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 # El usuario volvió al menú 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 # ya que el usuario hizo clic para volver al menú principal. elif choice == “3”: break # Inicia la sesión 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, )

… # código existente debajo

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:

# Banner de bienvenida
console.print(
Panel(
“[bold white]Bienvenido al Análisis Exploratorio de Datos Agéntico[/bold white]\ \ “
”[grey]¿Cómo te gustaría proceder?:[/grey]\ “
”[grey]1.[/grey] Descargar un conjunto de datos primero.\ “
”[grey]2.[/grey] Proceder con un conjunto de datos ya descargado.\ “
”[grey]3.[/grey] Salir”,
title=“MENÚ PRINCIPAL”,
border_style=“green”,
width=70,
)
)

choice = Prompt.ask(
“\ [bold yellow]Ingresa tu elección[/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 # El usuario volvió al menú 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 # ya que el usuario hizo clic para volver al menú principal.

elif choice == “3”:
break

# Inicia la sesión 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,
)

Debido a que main.py se ejecutará como un script, agregamos el siguiente código, pasando también nuestras variables de entorno, el tiempo de espera del sandbox y los modelos de Novita que pretendemos usar:


… # código existente debajo 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 es verdadero, asegúrate de que el modelo del agente de navegador tenga capacidades de visión. ) NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct” NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900 segundos (15 minutos), la instancia del sandbox se eliminará automáticamente despué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, ) )

… # código existente debajo

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 es verdadero, asegúrate de que el modelo del agente de navegador tenga capacidades de visión.
)
NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct”
NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900 segundos (15 minutos), la instancia del sandbox se eliminará automáticamente despué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,
)
)

Probar la ejecución de nuestro eda-agent:

Ejecuta el siguiente comando en la terminal


uv run main.py

Conclusión

¡Felicidades por construir tu agente de análisis exploratorio de datos! Ahora puedes pedirle que cree sitios web, presentaciones de PowerPoint, etc., usando análisis/conocimientos de cualquier archivo de conjunto de datos, y los resultados se sincronizarán directamente en tu computadora local.

Para un resumen rápido, en este artículo aprendiste a construir un agente que puede tomar instrucciones, luego usar el navegador para navegar por la web y descargar archivos, ejecutar código y comandos en el Novita Sandbox, y sincronizar archivos y directorios a tu computadora local.

Esto es solo la punta del iceberg, puedes extender tu agente para conectarte a bases de datos, integrarte con herramientas como Google Docs a través de MCP y mucho más. ¡Dirígete a Novita para dar vida a tus ideas!

Novita AI es una plataforma de IA en la nube que ofrece a los desarrolladores una forma sencilla de implementar modelos de IA usando nuestra API simple, además de proporcionar la nube de GPU asequible y confiable para construir y escalar.

Cómo construir un agente con Novita AI Sandbox, productos LLM y uso del navegador.

¿Qué es un agente de IA?

Estos son los componentes típicos de un agente de IA:

Descripción general de Novita AI Sandbox, productos LLM y uso del navegador