Novita AI Sandbox、LLM製品、Browser Useを使用したエージェントの構築方法

最近、大規模言語モデル（LLM）、視覚言語モデルなどを含む人工知能の分野が指数的に成長しており、特に最近ではAIエージェントが注目されています。従来のルールベースのアプリケーションとは異なり、これらのエージェントは、推論と意思決定機能を活用して、複雑なタスクを自律的、または最小限のユーザー介入で実行することができます。これを行うために、AIエージェントは多くの場合、コードを動的に生成して実行したり、仮想マシン全体を制御したりする必要があり、Novita AI sandboxはこれらの機能を提供します。

このチュートリアルでは、データアナリストとして機能するAIエージェントを作成します。人間のアナリストと同様に、指示に従ってブラウザを使用してデータセットを見つけてダウンロードし、必要に応じて分析、可視化、コードを実行して、得られた洞察を回答として返すことができます。

AIエージェントとは？

AIエージェントは、人工知能を使用して推論し、計画を立て、特定の目標を達成するために行動を実行するソフトウェアプログラムです。これらのエージェントは、API、コード実行、Web検索などのツールを使用して、情報を収集したりタスクを実行したりすることができます。場合によっては、特にDeep Researchのようなより複雑な目標のために、他のエージェントと協力することもあります。

AIエージェントの典型的な構成要素：

• 推論/計画: 目標が与えられると、エージェントはそれを達成するための段階的な計画を作成します。このプロセスには、サブタスクの完了や次の行動を導くための追加情報の収集が含まれる場合があります。これは特に、時間がかかる複雑なタスクで一般的であり、推論モデルはこれらの状況で適切に機能します。
• ツールの使用: AIエージェントは外部ツールやサービスを呼び出すことができます。これらには、機能、API、または機能を拡張するその他のリソースが含まれる場合があります。
• 調整: 複数のエージェントが責任を分担して共同で作業し、それぞれが共有された複雑または長期的な目標の達成に貢献することができます。

Novita AI Sandbox、LLM製品、Browser Useの概要

Novita AI Sandbox

サンドボックスとは何か、そしてなぜ重要なのか

サンドボックスは、信頼されていないコードをホストシステムに影響を与えずに実行できる安全で分離されたランタイム環境です。基本的には、AIエージェントがコード、コマンドを実行し、ファイルを作成するための軽量な仮想コンピュータです。

Novita AIは、このサンドボックスをクラウド上で提供し、エージェントがオンデマンドで迅速にアクセスできるようにし、使用されたリソースに基づいた柔軟な秒単位の課金を実現しています。

Novitaサンドボックスの主な機能：

• 安全な分離: 各サンドボックスは分離されたファイルシステムと環境を持ち、データを保護し、意図しない相互作用を防止します。
• 高速起動: サンドボックスインスタンスは平均して約200ms未満で起動し、低遅延のシナリオに理想的です。
• マルチ言語サポート: Python、JavaScript、TypeScriptなど、複数のプログラミング言語でコードを実行できます。
• クイック一時停止と再開: いつでもサンドボックスを一時停止し、必要に応じて再開でき、ファイルシステムとプロセス状態が完全に復元されます。
• バックグラウンド実行: バックグラウンドタスクの実行をサポートし、結果を待つ必要があるシナリオに適しています。

Novita Model 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の概要:

Browser Useは、AIエージェントが自然言語コマンド（例：「今日のNYの天気を確認」）を使用してWebブラウザと対話できるようにするオープンソースのPythonライブラリです。複雑なルールベースのセレクタを記述する代わりに、AIが要素の検索と対話を処理します。また、NovitaのOpenAI互換モデルAPIが提供されているため、任意のLLMまたはVLM（視覚言語モデル）を使用してBrowser Useを動作させることができます。

開発環境のセットアップ

開始するには、GitHubリポジトリをクローンし、クリーンなPython環境をセットアップし、必要なすべての依存関係をインストールし、Novita AIキーを取得します。

GitHubリポジトリのクローンとuvを使用した依存関係のインストール

1. uv（軽量なPythonパッケージマネージャー）をインストールします。

pip install uv

2. リポジトリ（GitHub repo）をクローンし、そのディレクトリに移動します。

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 Useによるデータセット（またはファイル）のダウンロード：

人間のアナリストと同様に、エージェントが指示を受け取り、ブラウザを使用してデータセットを取得できるようにしたいと考えます。そのため、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=“ユーザータスクから生成されたファイル（スクレイピングまたは調査データなど）、該当する場合” )

以下は、エージェントブラウザがどのように動作するかのサンプルです。ブラウザのWebナビゲーションには大規模言語モデル（LLM）を使用します。これはほとんどのユースケースをカバーするはずです。ただし、より複雑なUIタスク（例：CAPTCHAの解決）の場合は、視覚言語モデル（VLM）が推奨されます。VLMを使用する場合、視覚の詳細レベル（‘high’、‘low’、または’auto’）を設定して、トークンコストと視覚の明確さのバランスを取ることができます。

次に、ブラウザセッションを作成し、ダウンロードパス（./Download）とuser_data_dir（シークレットモードの場合はNoneに設定）でプロファイルを構成し、Pydanticモデルをコントローラとして設定してエージェント出力を構造化します。次に、**await agent.run()**でエージェントを開始し、最終結果を解析してダウンロードされたファイル名を取得します。

# ブラウザエージェントの動作サンプル 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”, # 利用可能なオプション [‘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エージェントが利用可能なツールのスキーマの定義：

ブラウザの使用が完了したので、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は、2つの入力パラメータを受け取ります：
  • sandbox_path: サンドボックス内のファイルまたはディレクトリのパス。
  • path_on_user_sync_folder: エージェントがユーザーの同期フォルダ内でファイルまたはディレクトリに持たせたいパス。

この2番目のパスは、例えば（/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は、1つの入力パラメータのみを受け取ります：
  • 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画像出力とその他の出力（stdout、ログ、エラーなど）を含みます。 ”“” execution = self.sandbox.run_code(python_code, language=“python”) image_outputs = [result.png for result in execution.results if result.png] # base64エンコードされた画像を反復処理し、./temp_image_outputディレクトリにtemp-{timestamp}.pngの名前形式で保存します。 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]ファイル(s) {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]ファイル(s) {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関数呼び出しを使用して、作成、書き込み、更新したディレクトリ（構造を優先する場合はWebサイトなど）またはファイルを、ユーザーのローカルマシンの同期フォルダに同期できます。 - delete_from_user_sync_folder関数呼び出しを使用して、ユーザーのローカルマシンの同期フォルダからディレクトリまたはファイルを削除できます。 現在のPWDは’/home/user’であり、以下はその中のファイルです。 {list_sandbox_files} 注意: - サンドボックスには通常のデータ分析パッケージがプリインストールされていますが、インストールされているかどうか不明なパッケージや、インポートエラーで不足しているパッケージがある場合は、インストールされているか確認し、インストールされていない場合はインストールできます。 - 画像出力（データ可視化など）はPNG形式にしてください。 関数呼び出しガイドライン: - run_on_command_line（例：パッケージのインストールなど）をどうしても使用する必要がある場合を除き、常にrun_python_codeを使用してタスクを実行してください。 - 必要に応じて関数呼び出しを連鎖させます：1つの関数呼び出しから結果を受信した後、さらに情報が必要な場合はすぐに追加の呼び出しを行ってください。 - 最初に必要な情報だけを収集します：関数呼び出しから十分な情報が得られた場合にのみユーザーに回答してください。 - 効率的に：{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, ): “”“ コード実行とターミナルコマンドを備えたAIエージェントとの対話型EDAセッション 引数: downloaded_dataset_names (list[str]): ダウンロードされたデータセットの名前。 model_for_eda (str, optional): 使用する基盤モデル。 ”“” console.print( Panel( “[bold green]EDAセッションが開始されました[/bold green]\ 'quit()'と入力して終了します。”, title=“探索的データ分析”, 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=“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, # 画像出力（データ可視化など）がある場合、ツール呼び出しから画像を返すことがまだ不可能なため、 # 画像がユーザーに表示されたことをエージェントに通知します。 “content”: [ { “type”: “text”, “text”: ( f"画像は既にターミナルでユーザーに表示され、temp-{{timestamp}}.pngなどの一時ファイルとしてユーザーのコンピュータの./temp_image_outputディレクトリに保存されています。他の出力は以下です\ {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=“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”, # このメッセージがツール使用からのものであることを示します。 “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=“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”, # このメッセージがツール使用からのものであることを示します。 “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=“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”, # このメッセージがツール使用からのものであることを示します。 “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=“メインメニュー”, 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

結論

探索的データアナリストエージェントの構築おめでとうございます。これで、任意のデータセットファイルから分析/洞察を使用してWebサイトの作成、PowerPointの作成などを依頼できるようになりました。結果はローカルコンピュータに直接同期されます。

簡単にまとめると、この記事では、指示を受け取り、ブラウザを使用してWebをナビゲートしてファイルをダウンロードし、Novita Sandboxでコードとコマンドを実行し、ファイルとディレクトリをローカルコンピュータに同期するエージェントを構築する方法を学びました。

これはほんの一部に過ぎません。エージェントを拡張してデータベースに接続したり、MCPを介してGoogle Docsなどのツールと統合したり、さらに多くのことができます。Novitaにアクセスして、あなたのアイデアを実現しましょう！

Novita AIは、開発者がシンプルなAPIを使用してAIモデルをデプロイする簡単な方法を提供すると同時に、構築とスケーリングのための手頃で信頼性の高いGPUクラウドを提供するAIクラウドプラットフォームです。