Как создать агента с помощью Novita AI Sandbox, продуктов LLM и Browser Use.

В последнее время мы наблюдаем экспоненциальный рост в области искусственного интеллекта с появлением больших языковых моделей (LLM), визуально-языковых моделей и, совсем недавно, ИИ-агентов. В отличие от традиционных приложений, основанных на правилах, эти агенты способны использовать свои возможности рассуждения и принятия решений для выполнения сложных задач автономно или с минимальным вмешательством пользователя. Для этого ИИ-агентам часто необходимо динамически генерировать и выполнять код, а иногда даже управлять всей виртуальной машиной, и Novita AI sandbox предоставляет такие возможности.

В этом руководстве мы создадим ИИ-агента, который будет выступать в роли нашего аналитика данных, и, как и человеческий аналитик, он сможет находить и загружать наборы данных с помощью браузера по нашим инструкциям, а по запросу анализировать, визуализировать и выполнять код, чтобы предоставить нам готовые выводы.

Что такое ИИ-агент?

ИИ-агент — это программное обеспечение, которое использует искусственный интеллект для рассуждений, планирования и выполнения действий для достижения конкретных целей. Эти агенты могут использовать инструменты, такие как API, выполнение кода и веб-поиск, для сбора информации или выполнения задач. Иногда они также сотрудничают с другими агентами, особенно для решения более сложных задач, например, глубокого исследования (Deep Research).

Типичные компоненты ИИ-агента:

• Рассуждение/планирование: Получив цель, агент создает пошаговый план для её достижения. Этот процесс может включать выполнение подзадач и сбор дополнительной информации для управления последующими действиями. Это особенно характерно для сложных задач, на выполнение которых уходит много времени, и модели рассуждений хорошо подходят для таких ситуаций.
• Использование инструментов: ИИ-агенты могут вызывать внешние инструменты и сервисы. К ним могут относиться функции, API или другие ресурсы, которые расширяют их возможности.
• Координация: Несколько агентов могут работать вместе, распределяя обязанности, при этом каждый из них помогает достичь общей, сложной или долгосрочной цели.

Обзор Novita AI Sandbox, продуктов LLM и Browser Use

Novita AI Sandbox

Что такое песочница и почему это важно

Песочница — это безопасная изолированная среда выполнения, в которой ненадежный код может быть выполнен без влияния на хост-систему. По сути, это легкий виртуальный компьютер, который ваш ИИ-агент может использовать для выполнения кода, команд, создания файлов и т.д.

Novita AI предоставляет эту песочницу в облаке, чтобы ваш агент мог быстро получать к ней доступ по требованию, с гибкой поминутной тарификацией в зависимости от используемых ресурсов.

Ключевые особенности песочницы Novita:

• Безопасная изоляция: Каждая песочница имеет собственную изолированную файловую систему и среду, что защищает данные и предотвращает нежелательные взаимодействия.
• Быстрый запуск: Экземпляры песочницы запускаются в среднем менее чем за ~200 мс, что делает их идеальными для сценариев с низкой задержкой.
• Поддержка нескольких языков: Вы можете запускать код на нескольких языках программирования, включая Python, JavaScript, TypeScript и другие.
• Быстрая пауза и возобновление: Вы можете приостановить работу песочницы в любое время и возобновить её при необходимости, при этом состояние файловой системы и процессов полностью восстанавливается.
• Выполнение в фоновом режиме: Поддерживает выполнение задач в фоновом режиме и подходит для сценариев, требующих ожидания результата.

API моделей Novita:

Novita AI Models

Novita предлагает обширную библиотеку открытых ИИ-моделей от ведущих исследовательских лабораторий, таких как OpenAI, Google, DeepSeek, Qwen и других. В неё входят модели для работы с языком, зрением, аудио, видео и эмбеддингами. Наши языковые модели также полностью совместимы с SDK OpenAI, поэтому переход с OpenAI на Novita требует только обновления базового URL и ключа API в вашем клиенте, после чего вы можете выбрать модель Novita.

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

Обзор Browser Use:

Browser Use — это открытая библиотека Python, которая позволяет ИИ-агентам взаимодействовать с веб-браузерами с помощью команд на естественном языке (например, «Узнай погоду в Нью-Йорке сегодня»). Вместо написания сложных селекторов на основе правил ИИ самостоятельно находит элементы и взаимодействует с ними. А благодаря совместимому с OpenAI API моделей Novita вы можете использовать любую LLM или VLM (визуально-языковую модель) для работы Browser Use.

Настройка среды разработки

Для начала мы клонируем репозиторий GitHub, настроим чистую среду Python, установим все необходимые зависимости и получим ключи Novita AI.

Клонирование репозитория GitHub и установка зависимостей с помощью uv

1. Установите uv (легкий менеджер пакетов Python)

pip install uv

2. Клонируйте репозиторий (репозиторий GitHub) и перейдите в его каталог.

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

3. Создайте и активируйте виртуальное окружение 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. Установите зависимости проекта

# Install dependencies uv sync

Создание аккаунта Novita AI и получение ключа API

1. Зарегистрируйтесь на novita.ai.

2. В панели управления наведите курсор на иконку профиля пользователя и нажмите API Keys во всплывающем меню.

3. На странице управления ключами нажмите Add New Key. Во всплывающем окне введите имя для вашего ключа, нажмите Confirm, а затем скопируйте сгенерированный ключ.

4. Теперь в каталоге проекта создайте файл .env и вставьте в него следующее содержимое:

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

Пополните баланс аккаунта Novita AI.

Для использования песочницы Novita вам необходимо пополнить баланс вашего аккаунта. На вкладке панели управления нажмите ‘Billing’. Затем на странице биллинга добавьте способ оплаты и пополните баланс как минимум на 10 долларов.

Создание агента для разведочного анализа данных (EDA)

Теперь, когда наша среда настроена, а у нас есть ключ API, давайте запустим нашего агента

Загрузка наборов данных (или файлов) с помощью Browser Use агента:

Так же, как и человеческий аналитик, мы хотим, чтобы наш агент мог получать инструкции и использовать браузер для получения наборов данных. Поэтому, как видно в файле browser_agent.py, мы сначала создаем структуру данных Pydantic для конечного вывода агента, которая будет содержать имена загруженных файлов и результаты выполненных задач для записи в файл.

… # 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” )

Ниже приведен пример работы агента браузера. Мы будем использовать большую языковую модель (LLM) для навигации по веб-страницам в браузере, что должно покрывать большинство сценариев использования. Однако для более сложных задач с пользовательским интерфейсом (например, решение капчи) рекомендуется использовать визуально-языковую модель (VLM). При использовании VLM вы можете задать уровень детализации зрения («высокий», «низкий» или «авто»), чтобы сбалансировать стоимость токенов и четкость восприятия.

Затем мы создаем сессию браузера, настраиваем профиль с путем загрузки (./Download) и user_data_dir (установлен в None для режима инкогнито), а также задаем нашу модель Pydantic в качестве контроллера, чтобы получить структурированные выходные данные агента. После этого мы запускаем агента с помощью await agent.run(), а конечные результаты парсятся для получения имен загруженных файлов.

# 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.

(Примечание: Запуск агента браузера на вашем локальном компьютере запустит реальный экземпляр браузера, поэтому вы сможете наблюдать за его работой.) Полный метод будет собирать имена загруженных файлов, записывать результаты задач в файл и возвращать все пути к файлам для последующей загрузки.

… # 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)

Вызов агента обернут в блок try-except, поэтому при возникновении ошибки выбрасывается исключение, а переменной file_results присваивается значение None. В итоге метод возвращает как путь download_dir_path, так и пути из file_results.

Определение схем доступных инструментов для нашего агента:

Теперь, когда мы закончили работу с Browser Use, пора определить схемы инструментов, доступных нашему агенту EDA. Мы предоставим четыре инструмента:

• run_python_code: позволяет агенту выполнять код Python внутри песочницы.
• run_on_command_line: позволяет агенту запускать команды в терминале песочницы (например, устанавливать пакеты Python).
• sync_with_user: позволяет агенту синхронизировать созданные или обновленные файлы и каталоги из песочницы в вашу локальную папку синхронизации.
• delete_from_user_sync_folder: позволяет агенту удалять файлы или каталоги из локальной папки синхронизации.

Вместе все эти инструменты дают агенту полный контроль над выполнением кода, использованием терминала и синхронизацией файлов между песочницей и вашей локальной системой.

В файле sandbox_eda.py мы видим схемы для этих инструментов:

• run_python_code, который просто принимает код Python для выполнения в качестве входного параметра и возвращает результат, если он есть.

{ “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, который также просто принимает команду для выполнения в качестве входного параметра и возвращает результат, если он есть.

{ “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, который принимает два входных параметра:
  • sandbox_path: путь к файлу или каталогу внутри песочницы.
  • path_on_user_sync_folder: путь, который агент хочет присвоить файлу или каталогу внутри папки синхронизации пользователя.

Этот второй путь будет иметь форму (например, /new.txt), при условии, что папка синхронизации является родительской. Например, позже мы преобразуем /new.txt в sync_folder/new.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, который принимает только один входной параметр:
  • path_on_user_sync_folder: путь к файлу или каталогу, который агент хочет удалить из папки синхронизации пользователя.

Обратите внимание, что здесь используются те же предположения о путях, что и в 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”], }, }, },

Реализация доступных функций:

Сначала у нас есть класс SandboxEDA, который принимает следующие параметры:

• sandbox: экземпляр песочницы.
• model_base_url и model_api_key: для подключения к модели Novita.
• max_consecutive_function_calls_allowed: по умолчанию равен 30, чтобы предотвратить бесконечные циклы вызовов функций агентом, как мы увидим позже.

… # 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 )

Затем метод run_python_code как метод класса SandboxEDA. Метод принимает код Python на входе и использует экземпляр sandbox для его выполнения. При получении выходных данных все изображения (обратите внимание: они закодированы в base64) сохраняются в каталог temp_image_output. В итоге метод возвращает словарь, содержащий выходные данные изображений, другие выходные данные, логи и любые ошибки.

… # 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, }, }

Далее метод run_on_command_line. Он аналогичным образом выполняет команду на экземпляре песочницы, а затем возвращает словарь с выходными данными. При возникновении ошибки выполнения он возвращает словарь со значением None для выходных данных и устанавливает «ошибка выполнения» в сообщение об ошибке.

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

И метод sync_with_user. Как описано ранее, он принимает sandbox_path и path_on_user_sync_folder. Если путь в песочнице указывает на файл, он загружает файл из песочницы в соответствующие каталоги папки синхронизации. Если это каталог, метод рекурсивно обходит все вложенные элементы и загружает каждый файл в соответствующие ему расположения. При успешном выполнении возвращается «Синхронизация успешна», в противном случае возвращается сообщение об исключении.

… # 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)

Наконец, метод delete_from_user_sync_folder. Как уже было описано, он также принимает path_on_user_sync_folder. Затем мы удаляем соответствующий файл или каталог, если он существует, и возвращаем «Удаление успешно», в противном случае возвращаем сообщение об исключении.

… # 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)

Другие функции песочницы:

1. Метод загрузки файлов в песочницу.

… # 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. Метод получения содержимого основного каталога песочницы (/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”)]

Реализация взаимодействия с агентом:

Теперь мы рассмотрим метод взаимодействия с агентом. Перед этим давайте посмотрим на параметризированную инструкцию системного промпта в файле 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} ”“”

Теперь перейдем к методу eda_chat в классе SandboxEDA. Этот метод принимает имена файлов, загруженных в песочницу (мы будем обрабатывать загрузку перед началом чата, как увидим позже), и имя модели Novita, которую нужно использовать.

1. Сначала мы настраиваем клиент OpenAI на указание Novita через базовый URL и инициализируем разговор с системным промптом в качестве первого сообщения.

… # 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. Далее основной цикл чата представляет собой цикл while, в котором мы берем сообщение пользователя и добавляем его в существующую историю сообщений. А чтобы предотвратить бесконечные последовательные вызовы инструментов, внутри цикла while есть цикл for, который работает до достижения лимита, выбрасывая исключение при его достижении. Затем модель получает запрос с этими сообщениями.

… # 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. Далее мы проверяем, решила ли модель вызвать инструмент. Если да, мы выполняем инструмент с использованием аргументов, предоставленных моделью, выводим результат в терминал для пользователя (или отображаем изображение, если применимо), а затем возвращаем результат модели, начиная с вызова инструмента 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. Вызов инструмента 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.

… # 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. Вызов инструмента delete_from_user_sync_folder и выброс ошибки при вызове несуществующих функций.

… # 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. Если последний ответ агента не является вызовом инструмента, выводим его ответ пользователю и выходим из цикла ограничения последовательных вызовов инструментов.

… # 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

Оркестрация потока работы агента:

Наконец, у нас есть файл main.py, который является точкой входа в наше приложение и объединяет все компоненты. Внутри него метод start_eda запускает новую сессию песочницы. Параметр sandbox_timeout определяет, как долго песочница остается активной перед автоматическим завершением; для нашей демонстрации мы установим его значение в 900 секунд (≈15 минут).

… # 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})-----[/]\ " )

После создания песочницы мы загружаем в неё файлы, а затем запускаем метод eda_chat для начала взаимодействия с агентом.

… # 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, )

Поскольку main.py будет запускаться как скрипт, мы добавляем нижеприведенный код, также передавая переменные окружения, таймаут песочницы и модели Novita, которые мы планируем использовать:

… # 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, ) )

Тестовый запуск нашего eda-агента:

Выполните следующую команду в терминале

uv run main.py

Заключение

Поздравляем с созданием вашего агента для разведочного анализа данных. Теперь вы можете попросить его создать веб-сайты, презентации PowerPoint и т.д., используя анализ и выводы из любых файлов наборов данных, а результаты будут синхронизированы напрямую с вашим локальным компьютером.

Краткий recap: в этой статье вы научились создавать агента, который может получать инструкции, использовать браузер для навигации по вебу и загрузки файлов, выполнять код и команды в Novita Sandbox, а также синхронизировать файлы и каталоги с вашим локальным компьютером.

Это только верхушка айсберга: вы можете расширить возможности вашего агента, подключив его к базам данных, интегрировав с инструментами вроде Google Docs через MCP и многое другое. Перейдите на сайт Novita, чтобы воплотить ваши идеи в жизнь!

Novita AI — это облачная ИИ-платформа, которая предлагает разработчикам простой способ развертывания ИИ-моделей с помощью нашего простого API, а также предоставляет доступное и надежное облако GPU для разработки и масштабирования.