Como Construir um Agente com o Novita AI Sandbox, Produtos LLM e Uso de Navegador.

Recentemente, temos visto um crescimento exponencial no campo da inteligência artificial com Large Language Models (LLM), Vision Language Models, etc, e mais recentemente, agentes de IA. Ao contrário das aplicações tradicionais baseadas em regras, esses agentes são capazes de utilizar suas capacidades de raciocínio e tomada de decisão para realizar tarefas complexas de forma autônoma ou com intervenção mínima do usuário. Para fazer isso, os agentes de IA muitas vezes precisam gerar e executar código dinamicamente, ou até mesmo controlar toda a máquina virtual, e o Novita AI sandbox fornece essas capacidades.

Neste tutorial, criaremos um agente de IA que servirá como nosso analista de dados, e assim como um analista humano, ele será capaz de encontrar e baixar um conjunto de dados usando o Navegador de acordo com nossas instruções, e quando solicitado, analisar, visualizar e executar código para nos responder com insights extraídos.

O que é um Agente de IA?

Um agente de IA é um programa de software que usa inteligência artificial para raciocinar, fazer planos e executar ações para atingir objetivos específicos. Esses agentes podem usar ferramentas, como APIs, execução de código e pesquisas na web, para coletar informações ou realizar tarefas. Às vezes, eles também colaboram com outros agentes, especialmente para objetivos mais complexos, como Deep Research.

Aqui estão os componentes típicos de um agente de IA:

• Raciocínio/Planejamento: Quando recebe um objetivo, o agente cria planos passo a passo para atingi-lo. Esse processo pode envolver a conclusão de subtarefas e a coleta de informações adicionais para orientar suas próximas ações. Isso é especialmente comum para tarefas complexas que demoram mais, e os modelos de raciocínio funcionam bem nessas situações.
• Uso de Ferramentas: Os agentes de IA podem chamar ferramentas e serviços externos. Isso pode incluir funções, APIs ou outros recursos que estendem suas capacidades.
• Coordenação: Vários agentes podem trabalhar juntos compartilhando responsabilidades, com cada um ajudando a atingir um objetivo compartilhado, complexo ou de longo prazo.

Visão Geral do Novita AI Sandbox, Produtos LLM e Uso de Navegador

Novita AI Sandbox

O que é um Sandbox e Por Que Ele Importa

Um sandbox é um ambiente de execução seguro e isolado onde código não confiável pode ser executado sem afetar o sistema host. É basicamente um computador virtual leve para o seu agente de IA executar código, comandos, criar arquivos, etc.

A Novita AI fornece esse sandbox na nuvem para o seu agente acessar rapidamente sob demanda, com faturamento flexível por segundo com base nos recursos usados.

Principais recursos do sandbox Novita:

• Isolamento Seguro: Cada sandbox recebe seu sistema de arquivos e ambiente isolados, protegendo dados e evitando interações não intencionais.
• Inicialização Rápida: As instâncias do sandbox são iniciadas em menos de ~200ms em média, tornando-o ideal para cenários de baixa latência.
• Suporte a Múltiplas Linguagens: Você pode executar código em várias linguagens de programação, incluindo Python, JavaScript, TypeScript e mais.
• Pausa e Retomada Rápidas: Pause o sandbox a qualquer momento e retome quando necessário, com o estado do sistema de arquivos e dos processos totalmente restaurado.
• Execução em Segundo Plano: Suporta execução de tarefas em segundo plano e é adequado para cenários que exigem espera por um resultado.

API de Modelos Novita:

Novita AI Models

A Novita oferece uma vasta biblioteca de modelos de IA de código aberto de laboratórios de pesquisa líderes como OpenAI, Google, DeepSeek e Qwen, etc. Isso inclui modelos para linguagem, visão, áudio, vídeo e embeddings. Nossos modelos de linguagem também são totalmente compatíveis com o SDK OpenAI, então mudar do OpenAI para a Novita só requer atualizar a URL base e a chave de API no seu cliente, depois selecionar um modelo Novita.

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

Visão Geral do Uso de Navegador:

O Browser Use é uma biblioteca Python de código aberto que permite que agentes de IA interajam com navegadores da web usando comandos de linguagem natural (ex: “Verifique o clima em NY hoje”). Em vez de escrever seletores baseados em regras complexas, a IA lida com a localização e interação de elementos. E dada a API de modelos compatível com OpenAI da Novita, você pode usar qualquer LLM ou VLM (modelo de linguagem visual) para alimentar o Browser Use.

Configurando Seu Ambiente de Desenvolvimento

Para começar, clonaremos o repositório GitHub, configuraremos um ambiente Python limpo, instalaremos todas as dependências necessárias e obteremos as chaves da Novita AI.

Clone o repositório GitHub e instale as dependências com o uv.

1. Instale o uv (um gerenciador de pacotes Python leve)

pip install uv

2. Clone o repositório (repositório GitHub) e navegue até ele.

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

3. Crie e ative o ambiente virtual uv

# Creates a virtual environment uv venv # Activate the virtual environment source .venv/bin/activate # For Mac/Linux # or .venv\Scripts\activate # For Windows

4. Instale as dependências do projeto

# Install dependencies uv sync

Criando uma conta Novita AI e obtendo uma chave de API

1. Cadastre-se em novita.ai.

2. No painel, passe o mouse sobre o ícone de perfil de usuário e clique em API Keys no pop-up.

3. Na página de Gerenciamento de Chaves, clique em Adicionar Nova Chave. No pop-up, insira um nome para sua chave, clique em Confirmar e depois copie a chave gerada.

4. Agora, dentro do diretório do projeto, crie um arquivo .env e cole o conteúdo abaixo.

NOVITA_API_KEY=“” NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1”

Adicione créditos à conta Novita AI.

Para usar o sandbox Novita, você precisa adicionar créditos à sua conta. Na aba do painel, clique em ‘Faturamento’. Depois, na página de faturamento, adicione um método de pagamento e adicione pelo menos $10 em créditos.

Construindo o Agente Analista de Dados Exploratórios (EDA)

Agora que nosso ambiente está configurado e temos nossa chave de API, vamos iniciar nosso Agente

Baixando conjuntos de dados (ou arquivos) via Uso de Navegador do Agente:

Assim como um analista humano, queremos que nosso agente seja capaz de receber instruções e usar um navegador para obter os conjuntos de dados. Então, como visto em browser_agent.py, primeiro criamos uma estrutura de dados Pydantic para a saída final do agente, que receberá os nomes dos arquivos baixados e os resultados das tarefas concluídas para serem escritos em um arquivo.

… # below existing code class TaskFile(BaseModel): “”“Represents a file generated as part of a task result e.g. scraped data or researched data.”“” filename: str = Field(…, description=“Name of the file including extension”) content: str = Field(…, description=“Text content to be written into the file”) class AgentOutput(BaseModel): “”“Final aggregated output of the browser agent execution.”“” downloaded_files: Optional[list[str]] = Field( None, description=“List of downloaded file names (with extension), if any” ) task_files: Optional[list[TaskFile]] = Field( None, description=“Files generated from user tasks (e.g., scraped or researched data), if any” )

Abaixo está um exemplo de como o navegador do agente funcionará. Usaremos um Large Language Model (LLM) para navegação na web, o que deve cobrir a maioria dos casos de uso. No entanto, para tarefas de interface de usuário mais complexas (ex: resolução de captcha), um Vision-Language Model (VLM) é recomendado. Ao usar um VLM, você pode definir o nível de detalhe de visão (‘alto’, ‘baixo’ ou ‘automático’) para equilibrar o custo de tokens com a clareza da visão.

Depois, criamos uma sessão de navegador, configuramos o perfil com um caminho de download (./Download) e user_data_dir (definido como None para o modo anônimo), e definimos nosso modelo Pydantic como controlador para obter as saídas do agente estruturadas. Então, iniciaremos o agente com await agent.run(), e os resultados finais são analisados para obter os nomes dos arquivos baixados.

# A sample of how the browser agent will look agent = Agent( task=“Go to Hugging Face, search for An-j96/SuperstoreData and open its page, then navigate to the Files tab and download the data csv file.”, llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key), use_vision=False, vision_detail_level=“auto”, # available options [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=“./Download”, user_data_dir=None ) ), # set the download directory path for the browser. controller=Controller( output_model=AgentOutput ), # Get the agent to output the name of the downloaded file at the end of the task. ) all_results = await agent.run() final_output = AgentOutput.model_validate_json( all_results.final_result() ) # parse the final agent result.

(Nota: Executar um agente de navegador na sua máquina local iniciará uma instância real de navegador para você poder ver em ação.) O método completo coletará os nomes dos arquivos baixados, escreverá os resultados das tarefas em um arquivo e retornará todos os caminhos de arquivo para upload posterior.

… # below existing code 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]]: “”“ Will perform the user’s download task via browser use and return download directory path and the downloaded files names. Returns: Tuple of (download_directory, filenames_with_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, # This penalty can slightly affect tool use; keep at 0. ), use_vision=use_vision, vision_detail_level=“auto”, # available options [‘low’, ‘high’, ‘auto’]; browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=download_dir_path, user_data_dir=None, # “./browser_user_data” ) ), # set the download directory path for the browser. controller=Controller( output_model=AgentOutput ), # Have the agent output the task execution result according to the AgentOutput schema. max_failures=5, ) try: # Run the agent and structure its output 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]Writing task results to files…[/bold yellow] {final_output.task_files}“, title=“Task Results”, border_style=“white”, ) ) # Write each task result to a file in the download directory task_result_files: list[str] = [] for task_file in final_output.task_files or []: file_path = Path(task_file.filename) # Prevent path traversal or unsafe absolute paths if file_path.is_absolute() or “…” in file_path.parts: raise ValueError( f"The agent passed an unsafe file path as a filename: {file_path}” ) # Point the file path inside the download directory file_path = Path(download_dir_path) / file_path # Ensure the download directory exists, else create it. file_path.parent.mkdir(parents=True, exist_ok=True) # Write task result content to a file 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]Task results written to files:[/bold green] {task_result_files}“, title=“Task Results”, border_style=“white”, ) ) # Combine downloaded files with task result files file_results = (final_output.downloaded_files or []) + task_result_files if file_results: console.print( Panel( f”[bold green]Files available:[/bold green] {file_results} in {download_dir_path}“, title=“Downloaded Files”, border_style=“green”, ) ) else: raise RuntimeError(“No files were downloaded or written.”) except Exception as e: file_results = None console.print( Panel( f”[bold red]Error:[/bold red] {str(e)}\ ", title=“Execution Error”, border_style=“red”, ) ) return (download_dir_path, file_results)

A chamada do agente está envolta em um bloco try-except, então se ocorrer uma falha, uma exceção é lançada e file_results é definido como None. Finalmente, o método retorna tanto o download_dir_path quanto os caminhos de file_results.

Definindo esquemas das ferramentas disponíveis para nosso agente:

Agora que terminamos o uso do navegador, é hora de definir os esquemas das ferramentas disponíveis para nosso agente EDA. Forneceremos quatro ferramentas:

• run_python_code: permite que o agente execute código Python dentro do sandbox.
• run_on_command_line: permite que o agente execute comandos no terminal do sandbox (ex: instalar pacotes Python).
• sync_with_user: permite que o agente sincronize arquivos e diretórios criados ou atualizados do sandbox para sua pasta de sincronização local.
• delete_from_user_sync_folder: permite que o agente remova arquivos ou diretórios da pasta de sincronização local.

Juntas, todas essas ferramentas dão ao agente controle total sobre a execução de código, uso do terminal e sincronização de arquivos entre o sandbox e seu sistema local.

Em sandbox_eda.py, vemos o esquema das ferramentas:

• run_python_code, que simplesmente recebe o código Python a ser executado como parâmetro de entrada e retorna um resultado, se houver.

{ “type”: “function”, “function”: { “name”: “run_python_code”, “description”: “Runs the python code and returns the result if any.”, “parameters”: { “type”: “object”, “properties”: { “python_code”: { “type”: “string”, “description”: “The Python code to run.”, } }, “required”: [“python_code”], }, }, },

• run_on_command_line, que novamente simplesmente recebe o comando a ser executado como parâmetro de entrada e retorna um resultado, se houver.

{ “type”: “function”, “function”: { “name”: “run_on_command_line”, “description”: “Runs the command on the command line and returns the result if any.”, “parameters”: { “type”: “object”, “properties”: { “command”: { “type”: “string”, “description”: “The command to run on the command line.”, } }, “required”: [“command”], }, }, },

• sync_with_user, que recebe dois parâmetros de entrada:
  • sandbox_path: o caminho para o arquivo ou diretório dentro do sandbox.
  • path_on_user_sync_folder: o caminho que o agente deseja que o arquivo ou diretório tenha dentro da pasta de sincronização do usuário.

Esse segundo caminho terá o formato (ex: /novo.txt), com a premissa de que a pasta de sincronização é a pasta pai. Por exemplo, mais tarde resolveremos /novo.txt para sync_folder/novo.txt.

{ “type”: “function”, “function”: { “name”: “sync_with_user”, “description”: “Will sync a file or directory on sandbox, to the sync folder on the user’s computer”, “parameters”: { “type”: “object”, “properties”: { “sandbox_path”: { “type”: “string”, “description”: “Path to the file or directory on the sandbox.”, }, “path_on_user_sync_folder”: { “type”: “string”, “description”: “Relative path where the file or directory will be placed inside the user’s sync folder. For example, ‘/hello.txt’ goes directly in the sync folder, while ‘/run1/hello.txt’ will be placed in a ‘run1’ subfolder within the sync folder.”, }, }, “required”: [“sandbox_path”, “path_on_user_sync_folder”], }, }, },

• delete_from_user_sync_folder, que recebe apenas um parâmetro de entrada:
  • path_on_user_sync_folder: o caminho para o arquivo ou diretório que o agente deseja excluir da pasta de sincronização do usuário.

Observe que isso segue as mesmas premissas de caminho que em sync_with_user.

{ “type”: “function”, “function”: { “name”: “delete_from_user_sync_folder”, “description”: “Will delete a file or directory from the sync folder on the user’s computer”, “parameters”: { “type”: “object”, “properties”: { “path_on_user_sync_folder”: { “type”: “string”, “description”: “Relative path to the file or directory on the user’s sync folder. For example, ‘/hello.txt’ will delete it directly from the sync folder, while ‘/run1/hello.txt’ will delete it directly from ‘run1’ subfolder within the sync folder.”, } }, “required”: [“path_on_user_sync_folder”], }, }, },

Implementando as funções disponíveis:

Primeiro, temos uma classe SandboxEDA que recebe os seguintes parâmetros:

• sandbox: a instância do sandbox.
• model_base_url e model_api_key: para conectar ao modelo Novita.
• max_consecutive_function_calls_allowed: o padrão é 30 para evitar loops de chamadas de função infinitas por parte do agente, como veremos mais tarde.

… # below existing code 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 )

Depois, o run_python_code como um método da classe SandboxEDA. O método recebe o código Python como entrada e usa a instância sandbox para executá-lo. Quando as saídas são retornadas, quaisquer saídas de imagem (observação: elas são codificadas em base64) são salvas no diretório temp_image_output. Finalmente, o método retorna um dicionário contendo as saídas de imagem, outras saídas, logs e quaisquer erros.

… # below existing code class SandboxEDA: … # below existing code def run_python_code(self, python_code: str) -> dict: “”“ Runs the python code on the sandbox, and if there are any images save them locally. Args: python_code (str): The python code to run. Returns: dict: Containing the base64 image outputs and other outputs (stdout, logs, error, etc). ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # Iterate through the base64 encoded images and save them to a file with name format: temp-{timestamp}.png to ./temp_image_output dir for b64_image in image_outputs: timestamp = int(time.time_ns()) image_filename = Path(f"./temp_image_output/temp-{timestamp}.png") # Will create the temp_image_output directory if it doesn’t exist already. 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, }, }

Próximo, o método run_on_command_line. Da mesma forma, ele executará o comando na instância do sandbox, depois retornará um dicionário com as saídas. Se a execução falhar, ele retorna um dicionário com None para as saídas e define “erro de execução” como a mensagem de erro.

… # below existing code class SandboxEDA: … # below existing code def run_on_command_line(self, command: str) -> dict: “”“ Runs the command on the sandbox. Args: command (str): The command to run. Returns: dict: Containing the output of the command and the execution error if any. ”“” 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)}

E o método sync_with_user. Como descrito anteriormente, ele receberá o sandbox_path e o path_on_user_sync_folder. Se o caminho do sandbox apontar para um arquivo, ele baixa o arquivo do sandbox para os locais correspondentes da pasta de sincronização. Se for um diretório, o método percorre recursivamente todo o conteúdo descendente e baixa cada arquivo para seus locais correspondentes. Se for bem-sucedido, retornamos “Sincronização Concluída”, caso contrário retornamos a mensagem da exceção.

… # below existing code class SandboxEDA: … # below existing code def sync_with_user(self, sandbox_path, path_on_user_sync_folder): “”“ Downloads a file or directory from the sandbox to the user’s sync folder. Args: sandbox_path (str): The path of the file or directory to sync in the sandbox. path_on_user_sync_folder (str): The relative destination path of the file or directory in the user’s sync folder. Returns: str: “Sync Successful” if the file or directory was synced successfully, otherwise an error message. ”“” try: path_info = self.sandbox.files.get_info(sandbox_path) if path_info.type == FileType.DIR: # If its a directory loop through the contents and download them. 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: # Ensure the file is always inside ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Make the path relative by stripping any root or drive component relative_path = sandbox_path_obj.relative_to( sandbox_path_obj.anchor or “.” ) # Final path inside sync_folder file_path = Path(“sync_folder”) / relative_path # Will create any directory in the path that doesn’t exist already. file_path.parent.mkdir(parents=True, exist_ok=True) # Download the file to sync folder. 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)

Finalmente, o método delete_from_user_sync_folder. Também, como já descrito, recebe path_on_user_sync_folder. Depois excluímos o arquivo ou diretório correspondente se ele existir e retornamos “Exclusão Concluída”, caso contrário retornamos a mensagem da exceção.

… # below existing code class SandboxEDA: … # below existing code def delete_from_user_sync_folder(self, path_on_user_sync_folder): “”“ Deletes a file or directory from the user sync folder. Args: path_on_user_sync_folder (str): The path of the file or directory to delete in the user sync folder. Returns: str: “Deletion Successful” if the file or directory was deleted successfully, otherwise an error message. ”“” # Ensure the file is always inside ./sync_folder. sandbox_path_obj = Path(path_on_user_sync_folder) # Make the path relative by stripping any root or drive component relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”) # Final path inside sync_folder delete_path = Path(“sync_folder”) / relative_path try: if not delete_path.exists(): raise Exception( f"File or Directory does not exist at {path_on_user_sync_folder} in sync folder." ) 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)

Outras funções do Sandbox:

1. Método de Upload de Arquivos para o sandbox.

… # below existing code class SandboxEDA: … # below existing code def upload_files_to_sandbox( self, file_paths: list[str], file_names_in_sandbox: list[str] ): “”“ Uploads files to the sandbox. Args: file_paths (list[str]): File paths of the files to upload (eg [”./Download/data.csv", “./Download/data2.csv”]). file_names_in_sandbox (list[str]): The names the files will take in the sandbox (eg [“data.csv”, “data2.csv”]). Note: The files will be uploaded to the sandbox’s /home/user directory (e.g ./home/user/data.csv, ./home/user/data2.csv). “”“ console.print( f”[yellow]Uploading files(s) at {file_paths} to 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]Files(s) {file_paths} uploaded to Sandbox[/bold cyan] (id: {self.sandbox.sandbox_id})" )

2. Método de listagem de conteúdo do diretório principal do sandbox (/home/user).

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

Construindo a interação com o agente:

Agora veremos um método para interagir com o agente. Antes disso, vamos analisar a instrução de prompt de sistema parametrizada em 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} ”“”

Agora vamos para o método eda_chat na classe SandboxEDA. Esse método recebe os nomes dos arquivos que foram enviados para o sandbox (lidaremos com o upload antes de iniciar o chat, como veremos mais tarde) e o nome do modelo Novita a ser usado.

1. Primeiro, configuramos o cliente OpenAI para apontar para a Novita via a URL base, e inicializamos a conversa com o prompt de sistema como a primeira mensagem.

… # below existing code class SandboxEDA: … # below existing code def eda_chat( self, downloaded_dataset_names: list[str], model_for_eda: str, ): “”“ Interactive EDA session with AI agent capable of code execution and terminal commands Args: downloaded_dataset_names (list[str]): The names of the downloaded datasets. model_for_eda (str, optional): The underlying model to use. ”“” console.print( Panel( “[bold green]EDA Session Started[/bold green]\ Type ‘quit()’ to exit.”, title=“Exploratory Data Analysis”, border_style=“green”, ) ) client = OpenAI( base_url=self.model_api_base_url, api_key=self.model_api_key, ) # Initialize conversation with 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. Em seguida, o loop principal de chat é um loop while, no qual recebemos a mensagem do usuário e a adicionamos ao histórico de mensagens existente. E para evitar chamadas de ferramentas consecutivas infinitas, dentro do loop while há um loop for até o limite, lançando uma exceção quando ele é atingido. O modelo é então solicitado com as mensagens.

… # below existing code # Main chat loop while True: user_input = Prompt.ask(“\ [bold yellow]>>> User Message[/bold yellow]”) if user_input.lower().strip() == “quit()”: break messages.append({“role”: “user”, “content”: user_input}) # Handle potential consecutive tool calls with a safety limit to avoid infinite loops for i in range(self.max_consecutive_function_calls_allowed + 1): if i == self.max_consecutive_function_calls_allowed: raise Exception( f"Consecutive tool calls from the Agent must not exceed {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. Em seguida, verificamos se o modelo decidiu fazer uma chamada de ferramenta. Se sim, executamos a ferramenta usando os argumentos fornecidos pelo modelo, imprimimos a saída no terminal para o usuário (ou exibimos uma imagem, se aplicável) e depois retornamos a saída para o modelo, começando pela chamada de ferramenta run_python_code.

… # below existing code tool_calls = response_message.tool_calls if tool_calls: messages.append( response_message ) # Add assistant message that triggered tool calls # Execute each requested tool call 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 Executing Python Code”, border_style=“blue”, ) ) code_result = self.run_python_code(args[“python_code”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, “name”: name, # If there are any image outputs (e.g data visualization), as it is not yet possible to return images # from a tool call just inform the Agent that the image has been shown to the user. “content”: [ { “type”: “text”, “text”: ( f"THE IMAGES HAS ALREADY BEEN SHOW TO THE USER ON THE TERMINAL AND SAVED TO TEMP FILES eg temp-{{timestamp}}.png on the user’s computer in ./temp_image_output dir, THE OTHER OUTPUTS ARE BELOW\ {code_result[‘other_outputs’]}“ if code_result[“image_outputs”] else f”{code_result[‘other_outputs’]}" ), } ], } ) display_sandbox_code_output(code_result)

3b. Chamada de ferramenta run_on_command_line.

… # below existing code elif name == “run_on_command_line”: console.print( Panel( args[“command”], title=“Agent Executing Command On Terminal”, border_style=“blue”, ) ) command_result = self.run_on_command_line(args[“command”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # Indicates this message is from tool use “name”: name, “content”: str(command_result), } ) display_sandbox_command_output(command_result)

3c. sync_with_user Chamada de ferramenta.

… # below existing code elif name == “sync_with_user”: console.print( Panel( f"[bold yellow]Agent Started Syncing {args[‘sandbox_path’]} To User’s Sync Folder ({args[‘path_on_user_sync_folder’]})[/bold yellow]", title=“File Syncing”, 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”, # Indicates this message is from tool use “name”: name, “content”: sync_result, } ) … # skipped for brevity

3d. Chamada de ferramenta delete_from_user_sync_folder e lança um erro desconhecido se houver chamadas de função inexistentes.

… # below existing code elif name == “delete_from_user_sync_folder”: console.print( Panel( f"[bold yellow]Agent Deleting File(s) From User’s Sync Folder ({args[‘path_on_user_sync_folder’]})[/bold yellow]", title=“File Syncing”, 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”, # Indicates this message is from tool use “name”: name, “content”: delete_result, } ) … # skipped for brevity else: raise ValueError(f"Unknown Function Call: {name}

4. Se a resposta mais recente do agente não for uma chamada de ferramenta, imprima sua resposta para o usuário e saia do loop de limite de chamadas de ferramentas consecutivas.

… # below existing code else: # No tool calls just display assistant response after adding it to the messages. messages.append( {“role”: “assistant”, “content”: response_message.content} ) console.print( f"[bold green]>>> Assistant Response: {response_message.content} [/]" ) break

Orquestrando o fluxo do agente:

Finalmente, temos o main.py servindo como ponto de entrada da nossa aplicação, unindo tudo. Dentro, o método start_eda inicia uma nova sessão de sandbox. O parâmetro sandbox_timeout determina por quanto tempo o sandbox permanece ativo antes de ser encerrado automaticamente; para nossa demonstração, vamos defini-lo como 900 segundos (≈15 minutos).

Depois que o sandbox é criado, enviamos arquivos para ele, depois iniciamos o método eda_chat para começar a interagir com o agente.

… # below existing code 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]Started Sandbox[/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]------ EDA Session Completed for Sandbox (id: {sandbox.sandbox_id}) ------[/]“ ) finally: console.print( f”[bold cyan]----- Closed Sandbox (id: {sandbox.sandbox_id})-----[/]\ " )

Abaixo está o método main ponto de início da nossa aplicação:

… # below existing code 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: # Welcome Banner console.print( Panel( “[bold white]Welcome To Agentic Exploratory Data Analysis[/bold white]\ \ “ ”[grey]How would you like to proceed:[/grey]\ “ ”[grey]1.[/grey] Download a dataset first.\ “ ”[grey]2.[/grey] Proceed with already downloaded dataset.\ “ ”[grey]3.[/grey] Exit”, title=“MAIN MENU”, border_style=“green”, width=70, ) ) choice = Prompt.ask( “\ [bold yellow]Enter your choice[/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 # User returned to main menu 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 # since user click back to main menu. elif choice == “3”: break # Start the EDA session 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, )

Como o main.py será executado como um script, adicionamos o código abaixo, passando também nossas variáveis de ambiente, tempo limite do sandbox e os modelos Novita que pretendemos usar:

… # below existing code 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 # If true make sure the browser agent model has vision capabilities. ) NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct” NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900 seconds (15 minutes), sandbox instance will be killed automatically after. 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, ) )

Testando a execução do nosso eda-agent:

Execute o seguinte comando no terminal

uv run main.py

Conclusão

Parabéns por construir seu agente Analista de Dados Exploratórios. Agora você pode pedir para ele criar sites, PowerPoint etc, usando análises/insights de qualquer arquivo de conjunto de dados, e os resultados serão sincronizados diretamente no seu computador local.

Para um rápido resumo, neste artigo você aprendeu a construir um agente que pode receber instruções, depois usar o navegador para navegar na web e baixar arquivos, executar código e comandos no Novita Sandbox, e sincronizar arquivos e diretórios com o seu computador local.

Isso é apenas a ponta do iceberg, você pode estender seu agente para conectar a bancos de dados, integrar com ferramentas como Google Docs via MCP, e muito mais. Acesse a Novita para dar vida às suas ideias!

Novita AI é uma plataforma de nuvem de IA que oferece aos desenvolvedores uma maneira fácil de implantar modelos de IA usando nossa API simples, além de fornecer a nuvem de GPU acessível e confiável para construir e escalar.