如何使用Novita AI沙盒、LLM产品与浏览器功能构建智能体

近年来，随着大语言模型（LLM）、视觉语言模型等技术的出现，人工智能领域经历了指数级增长，而近期AI智能体的发展尤为迅猛。与传统基于规则的应用不同，这类智能体能够运用推理与决策能力，在极少人工干预下自主完成复杂任务。要实现这一点，AI智能体通常需要动态生成并执行代码，甚至控制整个虚拟机，而Novita AI沙盒恰好提供了这些能力。

在本教程中，我们将创建一个AI智能体作为数据分析师，和人类分析师一样，它能够根据我们的指令使用浏览器查找并下载数据集，在收到提示时分析数据、生成可视化结果、执行代码，并最终为我们输出有价值的洞察。

什么是AI智能体？

AI智能体是一种利用人工智能进行推理、规划并执行操作以达成特定目标的软件程序。这类智能体可以使用API、代码执行、网页搜索等工具来收集信息或完成任务，部分场景下还会与其他智能体协作，以完成深度研究等更复杂的目标。

AI智能体的典型组成部分包括：

• 推理/规划能力：当接收到目标后，智能体会制定分步计划来达成目标。该过程可能涉及完成子任务、收集额外信息以指导后续操作，这类能力在耗时较长的复杂任务中尤为常见，推理模型非常适合处理这类场景。
• 工具调用能力：AI智能体可以调用外部工具和服务，包括函数、API或其他可扩展其能力的资源。
• 协同能力：多个智能体可以通过分工协作，共同完成共享的、复杂的或长期目标。

Novita AI沙盒、LLM产品与浏览器功能概览

Novita AI Sandbox

什么是沙盒及其重要性

沙盒是一种安全、隔离的运行时环境，可在其中执行不受信任的代码，且不会影响主机系统。它本质上是一台轻量级虚拟计算机，可供AI智能体执行代码、运行命令、创建文件等操作。

Novita AI提供云端沙盒服务，智能体可按需快速调用，并根据实际使用的资源灵活按秒计费。

Novita沙盒的核心特性：

• 安全隔离：每个沙盒都拥有独立的文件系统和运行环境，可保护数据安全，避免非预期的交互。
• 快速启动：沙盒实例平均启动时间不到200毫秒，非常适合低延迟场景。
• 多语言支持：支持运行多种编程语言的代码，包括Python、JavaScript、TypeScript等。
• 快速暂停与恢复：可随时暂停沙盒，需要时再恢复，文件系统和进程状态会完全保留。
• 后台执行：支持后台任务运行，适用于需要等待结果的场景。

Novita模型API：

Novita AI Models

Novita提供了来自OpenAI、Google、DeepSeek、Qwen等顶尖实验室的大量开源AI模型库，涵盖语言、视觉、音频、视频和嵌入模型。我们的语言模型完全兼容OpenAI SDK，从OpenAI迁移到Novita仅需更新客户端中的基础URL和API密钥，再选择对应的Novita模型即可。

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

浏览器功能概览：

Browser Use是一个开源的Python库，支持AI智能体通过自然语言指令（例如“查询纽约今日天气”）与网页浏览器交互。无需编写复杂的基于规则的选择器，AI会自动处理元素查找与交互操作。结合Novita兼容OpenAI的模型API，你可以使用任意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虚拟环境

# 创建虚拟环境 uv venv # 激活虚拟环境 source .venv/bin/activate # Mac/Linux系统 # 或 .venv\Scripts\activate # Windows系统

4. 安装项目依赖

# 安装依赖 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_agent.py中所示，我们首先为智能体的最终输出创建Pydantic数据结构，用于存储下载的文件名以及需要写入文件的任务执行结果。

… # 以下为已有代码 class TaskFile(BaseModel): “”“表示任务结果生成的文件，例如抓取的数据或研究数据”“” filename: str = Field(…, description=“包含扩展名的文件名”) content: str = Field(…, description=“要写入文件的文本内容”) class AgentOutput(BaseModel): “”“浏览器智能体执行的最终聚合输出”“” downloaded_files: Optional[list[str]] = Field( None, description=“下载的文件名列表（包含扩展名），如果有的话” ) task_files: Optional[list[TaskFile]] = Field( None, description=“用户任务生成的文件（例如抓取或研究的数据），如果有的话” )

以下是智能体浏览器的运行示例。我们将使用大语言模型（LLM）进行网页导航，可覆盖大多数使用场景；而对于更复杂的UI任务（例如验证码识别），建议使用视觉语言模型（VLM）。使用VLM时，可设置视觉细节级别（‘high’、‘low’或’auto’），以平衡令牌成本与视觉清晰度。

我们随后会创建浏览器会话，配置包含下载路径**（./Download）和user_data_dir**（设置为None以开启无痕模式）的配置文件，并将Pydantic模型设置为控制器以获取结构化的智能体输出。之后我们会通过**await agent.run()**启动智能体，最终解析结果获取下载的文件名。

# 智能体浏览器的运行示例 agent = Agent( task=“前往Hugging Face，搜索An-j96/SuperstoreData并打开其页面，然后切换到Files标签页下载数据csv文件。”, llm=ChatOpenAI(base_url=novita_base_url, model=novita_model, api_key=novita_api_key), use_vision=False, vision_detail_level=“auto”, # 可选值 [‘low’, ‘high’, ‘auto’] browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=“./Download”, user_data_dir=None ) ), # 设置浏览器的下载目录路径 controller=Controller( output_model=AgentOutput ), # 让智能体在任务结束时输出下载的文件名 ) all_results = await agent.run() final_output = AgentOutput.model_validate_json( all_results.final_result() ) # 解析最终智能体结果

（注意：在本地运行浏览器智能体时会启动真实的浏览器实例，你可以实时观察其运行过程。）完整的方法会收集下载的文件名，将任务结果写入文件，并返回所有文件路径供后续上传使用。

… # 以下为已有代码 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]]: “”“ 通过浏览器功能执行用户的下载任务，返回下载目录路径和下载的文件名。 返回值： 元组 (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, # 该惩罚可能轻微影响工具使用，保持为0 ), use_vision=use_vision, vision_detail_level=“auto”, # 可选值 [‘low’, ‘high’, ‘auto’] browser_session=BrowserSession( browser_profile=BrowserProfile( downloads_path=download_dir_path, user_data_dir=None, # “./browser_user_data” ) ), # 设置浏览器的下载目录路径 controller=Controller( output_model=AgentOutput ), # 让智能体按照AgentOutput模式输出任务执行结果 max_failures=5, ) try: # 运行智能体并结构化其输出 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]正在将任务结果写入文件…[/bold yellow] {final_output.task_files}“, title=“Task Results”, border_style=“white”, ) ) # 将每个任务结果写入下载目录的文件 task_result_files: list[str] = [] for task_file in final_output.task_files or []: file_path = Path(task_file.filename) # 防止路径遍历或不安全的绝对路径 if file_path.is_absolute() or “…” in file_path.parts: raise ValueError( f"智能体传入了不安全的文件路径作为文件名: {file_path}” ) # 将文件路径指向下载目录内 file_path = Path(download_dir_path) / file_path # 确保下载目录存在，不存在则创建 file_path.parent.mkdir(parents=True, exist_ok=True) # 将任务结果内容写入文件 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]任务结果已写入文件:[/bold green] {task_result_files}“, title=“Task Results”, border_style=“white”, ) ) # 合并下载的文件和任务结果文件 file_results = (final_output.downloaded_files or []) + task_result_files if file_results: console.print( Panel( f”[bold green]可用文件:[/bold green] {file_results}，位于 {download_dir_path}“, title=“Downloaded Files”, border_style=“green”, ) ) else: raise RuntimeError(“未下载或写入任何文件。”) except Exception as e: file_results = None console.print( Panel( f”[bold red]错误:[/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两个路径。

定义智能体可用工具的模式

现在浏览器功能部分已完成，接下来为EDA智能体定义可用工具的模式。我们将提供4个工具：

• 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”: “运行Python代码，若有返回值则返回结果。”, “parameters”: { “type”: “object”, “properties”: { “python_code”: { “type”: “string”, “description”: “要运行的Python代码。”, } }, “required”: [“python_code”], }, }, },

• run_on_command_line，同样仅需传入待执行的命令作为输入参数，若有返回值则返回结果。

{ “type”: “function”, “function”: { “name”: “run_on_command_line”, “description”: “在命令行运行命令，若有返回值则返回结果。”, “parameters”: { “type”: “object”, “properties”: { “command”: { “type”: “string”, “description”: “要在命令行运行的命令。”, } }, “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”: “将沙盒上的文件或目录同步到用户电脑的同步文件夹中”, “parameters”: { “type”: “object”, “properties”: { “sandbox_path”: { “type”: “string”, “description”: “沙盒中要同步的文件或目录的路径。”, }, “path_on_user_sync_folder”: { “type”: “string”, “description”: “文件或目录在用户同步文件夹中的相对目标路径。例如，‘/hello.txt’会直接放在同步文件夹根目录，而’/run1/hello.txt’会放在同步文件夹的’run1’子目录中。”, }, }, “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”: “从用户电脑的同步文件夹中删除文件或目录”, “parameters”: { “type”: “object”, “properties”: { “path_on_user_sync_folder”: { “type”: “string”, “description”: “用户同步文件夹中要删除的文件或目录的相对路径。例如，‘/hello.txt’会直接删除同步文件夹根目录下的该文件，而’/run1/hello.txt’会删除’run1’子目录下的该文件。”, } }, “required”: [“path_on_user_sync_folder”], }, }, },

实现可用功能

首先，我们定义SandboxEDA类，其初始化参数如下：

• sandbox：沙盒实例。
• model_base_url和model_api_key：用于连接Novita模型。
• max_consecutive_function_calls_allowed：默认值为30，用于防止智能体出现无限函数调用循环，后续会具体说明。

… # 以下为已有代码 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 )

接下来是SandboxEDA类的方法run_python_code，该方法接收Python代码作为输入，通过sandbox实例执行代码。返回结果后，会将所有图像输出（注意为base64编码）保存到temp_image_output目录中，最终返回包含图像输出、其他输出、日志及错误的字典。

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def run_python_code(self, python_code: str) -> dict: “”“ 在沙盒中运行Python代码，若有图像输出则保存到本地。 参数： python_code (str)：要运行的Python代码。 返回值： dict：包含base64编码的图像输出及其他输出（标准输出、日志、错误等）。 ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # 遍历base64编码的图像，按格式temp-{时间戳}.png保存到./temp_image_output目录 for b64_image in image_outputs: timestamp = int(time.time_ns()) image_filename = Path(f"./temp_image_output/temp-{timestamp}.png") # 若temp_image_output目录不存在则创建 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、execution error字段为错误信息的字典。

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def run_on_command_line(self, command: str) -> dict: “”“ 在沙盒中运行命令。 参数： command (str)：要运行的命令。 返回值： dict：包含命令输出及执行错误（如果有）。 ”“” 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两个参数。若沙盒路径指向文件，则将该文件从沙盒下载到同步文件夹的对应位置；若路径指向目录，则方法会递归遍历所有子内容，将每个文件下载到对应位置。执行成功时返回**“Sync Successful”**，否则返回异常信息。

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def sync_with_user(self, sandbox_path, path_on_user_sync_folder): “”“ 将沙盒中的文件或目录下载到用户的同步文件夹中。 参数： sandbox_path (str)：沙盒中要同步的文件或目录的路径。 path_on_user_sync_folder (str)：文件或目录在用户同步文件夹中的相对目标路径。 返回值： str：同步成功返回"Sync Successful”，否则返回错误信息。 “”" try: path_info = self.sandbox.files.get_info(sandbox_path) if path_info.type == FileType.DIR: # 如果是目录，遍历所有内容并下载 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: # 确保文件始终在./sync_folder目录内 sandbox_path_obj = Path(path_on_user_sync_folder) # 通过去除根目录或驱动器组件使路径变为相对路径 relative_path = sandbox_path_obj.relative_to( sandbox_path_obj.anchor or “.” ) # sync_folder内的最终路径 file_path = Path(“sync_folder”) / relative_path # 创建路径中所有不存在的目录 file_path.parent.mkdir(parents=True, exist_ok=True) # 将文件下载到同步文件夹 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参数，若目标文件或目录存在则删除，成功时返回**“Deletion Successful”**，否则返回异常信息。

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def delete_from_user_sync_folder(self, path_on_user_sync_folder): “”“ 从用户同步文件夹中删除文件或目录。 参数： path_on_user_sync_folder (str)：用户同步文件夹中要删除的文件或目录的路径。 返回值： str：删除成功返回"Deletion Successful”，否则返回错误信息。 “”“ # 确保文件始终在./sync_folder目录内 sandbox_path_obj = Path(path_on_user_sync_folder) # 通过去除根目录或驱动器组件使路径变为相对路径 relative_path = sandbox_path_obj.relative_to(sandbox_path_obj.anchor or “.”) # sync_folder内的最终路径 delete_path = Path(“sync_folder”) / relative_path try: if not delete_path.exists(): raise Exception( f"同步文件夹中不存在 {path_on_user_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. 上传文件到沙盒的方法

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def upload_files_to_sandbox( self, file_paths: list[str], file_names_in_sandbox: list[str] ): “”“ 上传文件到沙盒。 参数： file_paths (list[str])：要上传的文件路径（例如 [”./Download/data.csv", “./Download/data2.csv”]）。 file_names_in_sandbox (list[str])：文件在沙盒中的名称（例如 [“data.csv”, “data2.csv”]）。 注意： 文件会上传到沙盒的/home/user目录（例如 ./home/user/data.csv, ./home/user/data2.csv）。 “”“ console.print( f”[yellow]正在将文件 {file_paths} 上传到沙盒[/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]文件 {file_paths} 已上传到沙盒[/bold cyan]（ID: {self.sandbox.sandbox_id}）" )

2. 列出沙盒主目录（/home/user）内容的方法

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 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 = “”“ 你是一个探索性数据分析（EDA）智能体，可以访问一个带有互联网连接的沙盒，你可以在其中： - 通过run_python_code函数调用执行Python代码。 - 通过run_on_command_line或run_python_code函数调用，执行Linux机器上的几乎所有操作。 - 通过sync_with_user函数调用，将你创建、写入或更新的任意目录（例如结构化网站）或文件同步到用户本地电脑的同步文件夹中。 - 通过delete_from_user_sync_folder函数调用，从用户本地电脑的同步文件夹中删除任意目录或文件。 你当前的工作目录是’/home/user’，以下是该目录下的文件： {list_sandbox_files} 注意： - 沙盒已预装了常用的数据分析包，如果你不确定某个包是否存在，或代码因缺少包报导入错误，可以检查该包是否已安装，未安装则进行安装。 - 图像输出（例如数据可视化结果）需为png格式。 函数调用指南： - 除非绝对需要使用run_on_command_line（例如安装包等场景），否则始终使用run_python_code执行任务。 - 需要时串联函数调用：收到一个函数调用的结果后，如果需要更多信息，立即发起额外调用。 - 优先收集必要信息：仅当从函数调用中获取到足够信息以提供优质回答时，再回复用户。 - 保持高效：虽然连续函数调用的最大限制为{max_consecutive_function_calls_allowed}次，但请尽量使用最少的调用获取必要信息。 - 不要假设用户会阅读工具调用的输出，请主动将结果回复给用户。 作为用户的助手，帮助用户对(/home/user/)目录下的数据集文件（{downloaded_dataset_names}）执行EDA分析。 你可以执行以下函数调用： {available_function_calls_schema} ”“”

接下来是SandboxEDA类中的eda_chat方法，该方法接收已上传到沙盒的文件名（我们会在启动对话前完成上传，后续会说明）以及要使用的Novita模型名称。

1. 首先，我们将OpenAI客户端的基础URL指向Novita，并以系统提示作为第一条消息初始化对话。

… # 以下为已有代码 class SandboxEDA: … # 以下为已有代码 def eda_chat( self, downloaded_dataset_names: list[str], model_for_eda: str, ): “”“ 支持代码执行和终端命令的交互式EDA会话 参数： downloaded_dataset_names (list[str])：已下载的数据集名称。 model_for_eda (str, 可选)：要使用的底层模型。 ”“” console.print( Panel( “[bold green]EDA会话已启动[/bold green]\ 输入’quit()'退出。”, title=“Exploratory Data Analysis”, border_style=“green”, ) ) client = OpenAI( base_url=self.model_api_base_url, api_key=self.model_api_key, ) # 以系统提示初始化对话 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循环，达到调用上限时会抛出异常，随后将消息输入模型。

… # 以下为已有代码 # 主对话循环 while True: user_input = Prompt.ask(“\ [bold yellow]>>> 用户输入[/bold yellow]”) if user_input.lower().strip() == “quit()”: break messages.append({“role”: “user”, “content”: user_input}) # 处理可能的连续工具调用，设置安全限制避免无限循环 for i in range(self.max_consecutive_function_calls_allowed + 1): if i == self.max_consecutive_function_calls_allowed: raise Exception( f"智能体的连续工具调用次数不能超过 {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工具调用。

… # 以下为已有代码 tool_calls = response_message.tool_calls if tool_calls: messages.append( response_message ) # 添加触发工具调用的助手消息 # 执行每个请求的工具调用 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=“智能体正在执行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, # 如果有图像输出（例如数据可视化结果），由于工具调用目前无法直接返回图像， # 因此告知智能体图像已在终端展示并保存到用户电脑的临时文件中 “content”: [ { “type”: “text”, “text”: ( f"图像已展示在终端，并保存到用户电脑./temp_image_output目录下的临时文件（例如temp-{{timestamp}}.png），其他输出如下\ {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工具调用

… # 以下为已有代码 elif name == “run_on_command_line”: console.print( Panel( args[“command”], title=“智能体正在终端执行命令”, border_style=“blue”, ) ) command_result = self.run_on_command_line(args[“command”]) messages.append( { “tool_call_id”: tool_call.id, “role”: “tool”, # 表示该消息来自工具调用 “name”: name, “content”: str(command_result), } ) display_sandbox_command_output(command_result)

3c. sync_with_user工具调用

… # 以下为已有代码 elif name == “sync_with_user”: console.print( Panel( f"[bold yellow]智能体开始将 {args[‘sandbox_path’]} 同步到用户同步文件夹（{args[‘path_on_user_sync_folder’]）}[/bold yellow]", title=“文件同步”, 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”, # 表示该消息来自工具调用 “name”: name, “content”: sync_result, } ) … # 为简洁省略部分内容

3d. delete_from_user_sync_folder工具调用，若遇到不存在的函数调用则抛出未知错误。

… # 以下为已有代码 elif name == “delete_from_user_sync_folder”: console.print( Panel( f"[bold yellow]智能体正在从用户同步文件夹（{args[‘path_on_user_sync_folder’]）}删除文件[/bold yellow]“, title=“文件同步”, 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”, # 表示该消息来自工具调用 “name”: name, “content”: delete_result, } ) … # 为简洁省略部分内容 else: raise ValueError(f"未知的函数调用: {name}”)

4. 若智能体最新响应不是工具调用，则将其响应打印给用户，并退出连续工具调用限制循环。

… # 以下为已有代码 else: # 无工具调用，将助手响应添加到消息后并展示给用户 messages.append( {“role”: “assistant”, “content”: response_message.content} ) console.print( f"[bold green]>>> 智能体响应: {response_message.content} [/]" ) break

编排智能体流程

最后，main.py作为应用的入口文件，将所有部分整合在一起。其中的start_eda方法会启动新的沙盒会话，sandbox_timeout参数用于设置沙盒在自动终止前的活跃时长，本次演示中我们将其设置为900秒（约15分钟）。

沙盒创建完成后，我们会将文件上传到沙盒中，随后启动eda_chat方法开始与智能体交互。

… # 以下为已有代码 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]沙盒已启动[/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]------ 沙盒（ID: {sandbox.sandbox_id}）的EDA会话已结束 ------[/]“ ) finally: console.print( f”[bold cyan]----- 沙盒（ID: {sandbox.sandbox_id}）已关闭 -----[/]\ " )

以下是应用的入口main方法：

… # 以下为已有代码 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: # 欢迎横幅 console.print( Panel( “[bold white]欢迎使用智能体探索性数据分析工具[/bold white]\ \ “ ”[grey]请选择操作：[/grey]\ “ ”[grey]1.[/grey] 先下载数据集。\ “ ”[grey]2.[/grey] 使用已下载的数据集继续。\ “ ”[grey]3.[/grey] 退出”, title=“MAIN MENU”, border_style=“green”, width=70, ) ) choice = Prompt.ask( “\ [bold yellow]请输入你的选择[/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 # 用户返回主菜单 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 # 用户点击返回主菜单 elif choice == “3”: break # 启动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, )

由于main.py会以脚本形式运行，我们添加以下代码，同时传入环境变量、沙盒超时时间以及我们要使用的Novita模型：

… # 以下为已有代码 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 # 若设为True，请确保浏览器智能体使用的模型支持视觉能力。 ) NOVITA_MODEL_FOR_EDA = “qwen/qwen3-coder-480b-a35b-instruct” NOVITA_SANDBOX_TIMEOUT_SECONDS = 900 # 900秒（15分钟），沙盒实例会在超时后自动终止。 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

总结

恭喜你完成了探索性数据分析智能体的构建！现在你可以要求它基于任意数据集文件的分析结果生成网站、PPT等内容，所有结果会直接同步到你的本地电脑。

快速回顾一下，本文你学会了如何构建一个可接收指令、通过浏览器浏览网页并下载文件、在Novita沙盒中执行代码与命令、并将文件和目录同步到本地电脑的智能体。

这仅仅是冰山一角，你还可以扩展智能体的功能，例如连接数据库、通过MCP集成Google Docs等工具，实现更多可能。欢迎前往Novita将你的想法落地！

Novita AI是一个AI云平台，为开发者提供简单的API来部署AI模型，同时提供高性价比、可靠的GPU云服务，支持AI应用的构建与扩展。