以下の内容はhttps://touch-sp.hatenablog.com/entry/2025/04/14/141505より取得しました。

【SmolAgents】汎用性のあるサンドボックス用Dockerイメージを作成する

言語モデル

はじめに

今までは「Dockerfile」にuvやnode.jpのインストールとかファイルのコピーとかを記述していました。

使い方が変わるたびにDockerイメージを作り直していたのですが面倒くさいことに気づきました。

イメージからコンテナを作成するときにインストールとかファイルのコピーとかを実行するように変更しました。

各種ファイル

Dockerfile

ごくごくシンプルなものにしました。

FROM python:3.12-bullseye

# 必要な依存関係のインストールと環境設定
RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        build-essential net-tools

RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir 'smolagents[litellm,openai,gradio,mcp]'

RUN apt-get clean && \
    rm -rf /var/lib/apt/lists/*

# プロジェクト設定
WORKDIR /app

# 起動設定
CMD ["python", "-c", "print('Container ready')"]

sandbox.py

ここにいろいろ記述するようにしました。

# コンテナ内にディレクトリを作成
self.container.exec_run("mkdir -p /app/mcp-project/")

# filesystem.pyを/app/mcp-projectにコピー
tar_stream1 = io.BytesIO()
with tarfile.open(fileobj=tar_stream1, mode='w') as tar:
    local_path1 = '/home/hoge/gemini-works/filesystem.py'
    arcname1 = 'filesystem.py'
    tar.add(local_path1, arcname=arcname1)

tar_stream1.seek(0)
self.container.put_archive('/app/mcp-project/', tar_stream1.read())
print("filesystem.pyを/app/mcp-projectにコピーしました")

# .envを/appにコピー
tar_stream2 = io.BytesIO()
with tarfile.open(fileobj=tar_stream2, mode='w') as tar:
    local_path2 = '/home/hoge/gemini-works/.env'
    arcname2 = '.env'
    tar.add(local_path2, arcname=arcname2)

tar_stream2.seek(0)
self.container.put_archive('/app/', tar_stream2.read())
print(".envを/appにコピーしました")

# 仮想環境のセットアップコマンドを実行
setup_commands = [
    "pip install --no-cache-dir --root-user-action=ignore uv",
    "uv init mcp-project",
    "cd mcp-project && uv venv",
    "cd mcp-project && uv add 'mcp[cli]'"
]

for cmd in setup_commands:
    result = self.container.exec_run(
        ["/bin/sh", "-c", cmd],
        workdir="/app",  # コマンドを実行するディレクトリ
    )
    
    exit_code = result.exit_code
    # 終了コードが0以外の場合のみエラーとして扱う
    if exit_code != 0:
        raise Exception(f"コマンド '{cmd}' の実行に失敗しました。終了コード: {exit_code}")
    else:
        print(f"コマンド '{cmd}' は正常に実行されました")
        
print("仮想環境のセットアップが完了しました")

最終的にはこのようになっています。

import docker
import time
import io
import tarfile

class DockerSandbox:
    def __init__(self, image_name="agent-sandbox"):
        self.client = docker.from_env()
        self.container = None
        self.image_name = image_name

    def create_container(self):
        try:
            # コンテナを作成（ポートマッピングを追加）
            self.container = self.client.containers.run(
                self.image_name,
                command="tail -f /dev/null",  # コンテナを実行状態に保つ
                detach=True,
                tty=True,
                extra_hosts={"host.docker.internal": "host-gateway"},
                network_mode="bridge",
                ports={'7860/tcp': 7860},  # Gradioのデフォルトポート
                volumes={
                    "/home/hoge/data": {"bind": "/app/data", "mode": "rw"}
                }
            )

            # コンテナ内にディレクトリを作成
            self.container.exec_run("mkdir -p /app/mcp-project/")
            
            # filesystem.pyを/app/mcp-projectにコピー
            tar_stream1 = io.BytesIO()
            with tarfile.open(fileobj=tar_stream1, mode='w') as tar:
                local_path1 = '/home/hoge/gemini-works/filesystem.py'
                arcname1 = 'filesystem.py'
                tar.add(local_path1, arcname=arcname1)
        
            tar_stream1.seek(0)
            self.container.put_archive('/app/mcp-project/', tar_stream1.read())
            print("filesystem.pyを/app/mcp-projectにコピーしました")

            # .envを/appにコピー
            tar_stream2 = io.BytesIO()
            with tarfile.open(fileobj=tar_stream2, mode='w') as tar:
                local_path2 = '/home/hoge/gemini-works/.env'
                arcname2 = '.env'
                tar.add(local_path2, arcname=arcname2)
        
            tar_stream2.seek(0)
            self.container.put_archive('/app/', tar_stream2.read())
            print(".envを/appにコピーしました")

            # 仮想環境のセットアップコマンドを実行
            setup_commands = [
                "pip install --no-cache-dir --root-user-action=ignore uv",
                "uv init mcp-project",
                "cd mcp-project && uv venv",
                "cd mcp-project && uv add 'mcp[cli]'"
            ]
            
            for cmd in setup_commands:
                result = self.container.exec_run(
                    ["/bin/sh", "-c", cmd],
                    workdir="/app",  # コマンドを実行するディレクトリ
                )
                
                exit_code = result.exit_code
                # 終了コードが0以外の場合のみエラーとして扱う
                if exit_code != 0:
                    raise Exception(f"コマンド '{cmd}' の実行に失敗しました。終了コード: {exit_code}")
                else:
                    print(f"コマンド '{cmd}' は正常に実行されました")
                    
            print("仮想環境のセットアップが完了しました")

            print(f"コンテナを作成しました (ID: {self.container.id[:8]}...)")
        except Exception as e:
            raise Exception(f"コンテナ作成エラー: {e}")

    def gradio_run(self, code: str) -> None:
        if not self.container:
            self.create_container()
        
        # バックグラウンドでPythonスクリプトを実行
        self.container.exec_run(
            cmd=["python", "-c", code],
            detach=True
        )
        
        # ポート待機確認
        print("Gradioサーバーを起動中...", end="", flush=True)
        max_attempts = 10
        for attempt in range(max_attempts):
            time.sleep(1)
            print(".", end="", flush=True)
            
            # netstatを使用してポートのリスニング状態を確認
            netstat_result = self.container.exec_run(
                cmd=["bash", "-c", "netstat -tulpn 2>/dev/null | grep 7860 || echo ''"]
            )
            
            if netstat_result.output:
                print(" 完了!")
                print("\n✅ Gradioアプリが起動しました")
                print("📊 http://localhost:7860 でアクセスできます")
                return None
        
        print("\n❌ サーバー起動に失敗しました")
        return None
    
    def _safe_decode(self, data, encoding='utf-8', errors='strict'):
        """バイト列か文字列かを判断して適切に処理する"""
        if isinstance(data, bytes):
            return data.decode(encoding, errors=errors)
        return data
            
    def cleanup(self):
        if self.container:
            try:
                self.container.stop()
                self.container.remove()
                print("Container stopped and removed successfully")
            except Exception as e:
                print(f"エラー: {e}")
            finally:
                self.container = None
    
    def get_logs(self):
        """コンテナ内のプロセス状態とログを取得"""
        if not self.container:
            return "コンテナが起動していません"
            
        # プロセス確認
        ps_cmd = "ps aux | grep python | grep -v grep"
        ps_result = self.container.exec_run(cmd=["bash", "-c", ps_cmd])
        ps_output = self._safe_decode(ps_result.output).strip()
        
        # ポート確認
        port_cmd = "netstat -tulpn 2>/dev/null | grep 7860 || echo 'ポートが開いていません'"
        port_result = self.container.exec_run(cmd=["bash", "-c", port_cmd])
        port_output = self._safe_decode(port_result.output).strip()
        
        return f"プロセス状態:\n{ps_output}\n\nポート状態:\n{port_output}"
        
    def exec_command(self, command):
        """コンテナ内でコマンドを実行"""
        if not self.container:
            return "コンテナが起動していません"
        
        result = self.container.exec_run(cmd=["bash", "-c", command])
        return self._safe_decode(result.output, errors='ignore')

agent_runner.py

from sandbox import DockerSandbox

# DockerSandboxのインスタンスを作成
sandbox = DockerSandbox()

agent_code = """
try:
    from smolagents import CodeAgent, GradioUI, ToolCollection
    from mcp import StdioServerParameters
    import os
    from dotenv import load_dotenv
    from smolagents import LiteLLMModel

    load_dotenv()
    api_key = os.environ.get("GOOGLE_API_KEY")
    
    model = LiteLLMModel(
        "gemini/gemini-2.0-flash",
        api_key=api_key
    )

    server_parameters = StdioServerParameters(
        command="/usr/local/bin/uv",
        args=[
            "--directory",
            "/app/mcp-project",
            "run",
            "filesystem.py"
        ]
    )

    with ToolCollection.from_mcp(server_parameters, trust_remote_code=True) as tool_collection:
        agent = CodeAgent(
            model=model,
            tools=[*tool_collection.tools],
            additional_authorized_imports=["PIL", "os", "json"]
        )

        # エージェントの実行
        GradioUI(agent).launch(server_name='0.0.0.0', server_port=7860, share=False)
except Exception as e:
    print(f"エラーが発生しました: {str(e)}")
    with open('/tmp/error.log', 'w') as f:
        f.write(f"スタートアップエラー: {str(e)}\\n")
"""

try:
    # エージェントコンテナに関する情報を確認
    print("\n⚙️ コンテナ環境を確認しています...")
    
    # コンテナを作成して基本的な情報を確認
    sandbox.create_container()
    
    # Pythonとパッケージの確認
    print("\nPython環境:")
    print(sandbox.exec_command("which python || which python3 || echo 'Pythonが見つかりません'"))
    print("\nPythonバージョン:")
    print(sandbox.exec_command("python --version || python3 --version || echo 'バージョン情報を取得できません'"))
    
    # 必要なパッケージの確認
    print("\n必要なパッケージ確認:")
    print(sandbox.exec_command("pip list | grep -E 'smolagents|gradio' || echo 'パッケージが見つかりません'"))
    
    # Gradioアプリを起動
    print("\n🚀 Gradioアプリを起動します...")
    sandbox.gradio_run(agent_code)
    
    # ユーザーが終了するまで待機
    print("\nアプリ実行中... Ctrl+C で終了します")
    
    while True:
        try:
            cmd = input("\n> ")
            if cmd.lower() == "exit" or cmd.lower() == "quit":
                break
            elif cmd.lower() == "status":
                print("\n" + sandbox.get_logs())
            elif cmd.lower() == "exec":
                command = input("実行するコマンド: ")
                print("\n" + sandbox.exec_command(command))
            elif cmd.lower() == "help":
                print("\nコマンド一覧:")
                print("  status - サーバー状態を確認")
                print("  exec   - コンテナ内でコマンドを実行")
                print("  exit   - アプリを終了")
                print("  help   - このヘルプを表示")
            elif cmd.strip() == "":
                pass
            else:
                print(f"不明なコマンド: {cmd}. 'help'と入力してコマンド一覧を表示")
        except KeyboardInterrupt:
            print("\n終了します...")
            break
    
except Exception as e:
    print(f"エラーが発生しました: {e}")
finally:
    # 終了処理
    sandbox.cleanup()

実行時の画面

⚙️ コンテナ環境を確認しています...
filesystem.pyを/app/mcp-projectにコピーしました
.envを/appにコピーしました
コマンド 'pip install --no-cache-dir --root-user-action=ignore uv' は正常に実行されました
コマンド 'uv init mcp-project' は正常に実行されました
コマンド 'cd mcp-project && uv venv' は正常に実行されました
コマンド 'cd mcp-project && uv add 'mcp[cli]'' は正常に実行されました
仮想環境のセットアップが完了しました
コンテナを作成しました (ID: 05fde52c...)

Python環境:
/usr/local/bin/python


Pythonバージョン:
Python 3.12.9


必要なパッケージ確認:
gradio                    5.25.0
gradio_client             1.8.0
smolagents                1.13.0


🚀 Gradioアプリを起動します...
Gradioサーバーを起動中.... 完了!

✅ Gradioアプリが起動しました
📊 http://localhost:7860 でアクセスできます

アプリ実行中... Ctrl+C で終了します



ランキング参加中
プログラミング



以上の内容はhttps://touch-sp.hatenablog.com/entry/2025/04/14/141505より取得しました。
このページはhttp://font.textar.tv/のウェブフォントを使用してます
不具合報告/要望等はこちらへお願いします。
モバイルやる夫Viewer Ver0.14