Агенты выполнения кода: E2B, Modal и Docker Sandbox


title: "Агенты выполнения кода: E2B, Modal и Docker Sandbox" slug: code-execution-agents-e2b-modal-2026-ru date: 2026-02-21 lang: ru tags: [code-execution, e2b, modal, docker, sandbox, agents, security] description: "Production-руководство по агентам выполнения кода: E2B Sandbox с постоянной файловой системой, Modal serverless GPU-контейнеры, Docker-sandbox с изоляцией, итеративные циклы выполнения и поддержка нескольких языков."

Агенты выполнения кода: E2B, Modal и Docker Sandbox

Ключевые факты

  • E2B версия пакета: e2b==2.14.0, e2b-code-interpreter==2.4.1 (PyPI, март 2026)
  • Modal версия пакета: modal==1.3.4 (PyPI, март 2026)
  • E2B время холодного старта Sandbox: Обычно 2-5 секунд для базовых образов; пользовательские шаблоны 3-8 секунд
  • E2B тайм-аут по умолчанию: 60 секунд на выполнение (настраивается до 600 секунд)
  • E2B время жизни сессии: Sandbox'ы существуют до явного уничтожения или тайм-аута 1 час
  • E2B модель ценообразования: Оплата за минуту использования sandbox'а; бесплатный уровень включает 100 часов в месяц
  • Modal холодный старт: 1-3 секунды для slim-образов; 5-15 секунд для GPU-контейнеров
  • Modal ценообразование: Pricing за compute-minute; CPU ~$0.000231/second, GPU T4 ~$0.000231/second, A10G ~$0.001155/second (варьируется по регионам)
  • E2B изоляция безопасности: Firecracker microVMs с kernel-level изоляцией
  • Modal изоляция безопасности: gVisor для контейнеризации
  • Docker изоляция: Namespace-изоляция, seccomp-профили, отказ от capabilities (настраиваемые уровни безопасности)
  • E2B поддерживаемые языки: Python (основной), Node.js, Bash через code interpreter
  • Modal поддерживаемые языки: Python-native; любой язык через пользовательские образы контейнеров
  • Docker поддержка нескольких языков: Python, JavaScript, TypeScript, Ruby, Go, Rust, Bash, R, Julia через выбор образа
  • E2B файловая система: Постоянна при выполнении в пределах одной сессии sandbox
  • Modal файловая система: Временная по умолчанию; постоянство требует Dict или Volume primitives
  • E2B установка пакетов: Runtime pip install возможна но добавляет latency; пользовательские шаблоны рекомендованы для production
  • Modal установка пакетов: Предустановлены в определениях Image при deploy-времени
  • Docker лимиты памяти: Настраиваемые (по умолчанию 512MB-2GB); требуются через cgroups
  • Максимальный размер выходных данных: E2B усекает на ~1MB; Modal и Docker настраиваемы но рекомендуется <5MB для LLM контекста
  • Сетевой доступ: E2B разрешает по умолчанию; Docker настраиваемый (флаг network_disabled); Modal разрешает сетевой доступ

Что такое безопасное выполнение кода

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

Безопасное выполнение кода с sandbox'ingом решает это через:

  1. Изоляция процесса: Предотвращение выхода из sandbox на host-систему
  2. Ограничения ресурсов: Принудительное применение лимитов CPU, памяти, диска и времени
  3. Ограничения возможностей: Блокировка системных вызовов (mount, ptrace, kernel modules)
  4. Границы файловой системы: Контроль доступа read/write к файлам host
  5. Сетевые политики: Ограничение или блокировка внешних соединений
  6. Санитизация выходных данных: Предотвращение атак инъекции через захваченный stdout/stderr

Уровень изоляции определяет надежность: firecracker microVMs (E2B) обеспечивают near-VM-level разделение; gVisor (Modal) перехватывает syscalls с user-space kernel; Docker namespaces предлагают более легкую изоляцию, подходящую для доверенных или низкорисковых рабочих нагрузок.

Production-системы требуют sandbox'inga потому что:

  • Агенты итерируют непредсказуемо: Одна задача может генерировать десятки выполнений кода
  • Состояния ошибок — это данные: Failed executions со stack traces информируют обучение агента
  • Таймауты должны быть принудительными: Бесконечные циклы или неэффективные алгоритмы нуждаются в жестких cutoff
  • Observability критична: Каждое выполнение должно быть залогировано для дебаггинга и аудита

Фреймворк принятия решений

Используйте E2B когда:

  • Требуется постоянное состояние файловой системы при множественных выполнениях кода
  • Multi-step анализ данных или итеративные debugging workflows
  • Python/Node.js code interpreter pattern с установкой пакетов
  • Управляемая инфраструктура предпочтительнее self-hosted
  • Бюджет позволяет pay-per-minute cloud pricing
  • Холодный старт <5 секунд приемлем
  • Требуется максимальная изоляция безопасности (Firecracker microVMs)

Используйте Modal когда:

  • Требуется GPU ускорение для ML inference или training
  • Computationally intensive задачи (large-scale обработка данных, симуляции)
  • Python-first workflows с комплексными зависимостями
  • Требуется serverless масштабирование (0 to N контейнеров автоматически)
  • Предпочитаются pre-built образы вместо runtime установки пакетов
  • Команда уже использует Modal для других рабочих нагрузок
  • Оптимизация стоимости через per-second billing для intermittent рабочих нагрузок

Используйте Docker (self-hosted) когда:

  • Требуется on-premises или air-gapped окружение
  • Regulatory compliance запрещает cloud execution (GDPR, HIPAA data residency)
  • Требуется multi-language поддержка за пределами Python (R, Julia, Go, Rust)
  • Требуется полный контроль над security profiles
  • Cost constraints благоприятствуют self-hosted compute
  • Интеграция с существующей Kubernetes/container инфраструктурой
  • Пользовательские требования networking или volume mount

Используйте Daytona когда:

  • Требуется full persistent development environment (не просто code execution)
  • Long-running agent сессии (часы и дни)
  • IDE интеграция или interactive debugging требуются
  • Team collaboration на agent-generated artifacts

Гибридный подход:

  • E2B для начального prototyping и итеративного анализа
  • Modal для финального production рабочих нагрузок требующих GPU/scale
  • Docker для regulated data processing pipelines

Таблица справки параметров

Параметр E2B значение Modal значение Docker значение Примечания
Timeout timeout=60 (по умолчанию), макс 600 timeout=300 (по умолчанию), макс 86400 timeout_seconds=30 (рекомендуется) E2B timeout — per-execution; Modal — per-function invocation
Memory Limit Не прямо настраиваемый (sandbox-level) memory=4096 (MB) mem_limit="512m" E2B имеет типы instance; Modal/Docker имеют per-container лимиты
CPU Limit Instance-based (не user-configurable) cpu=2.0 (cores) cpu_quota=50000 (50% от 1 CPU) Docker использует cpu_quota (microseconds за 100ms период)
GPU Не поддерживается gpu="T4" или gpu="A10G" device_requests=[...] с nvidia-docker Modal GPU типы: T4, A10G, A100, H100
API Key Env E2B_API_KEY MODAL_TOKEN_ID, MODAL_TOKEN_SECRET N/A Modal использует modal token new для setup
Session Persistence Sandbox() или AsyncSandbox() context Dict.from_name() для state Container pooling pattern E2B sandbox сохраняет filesystem; Modal требует explicit state management
Network Access Включен по умолчанию Включен по умолчанию network_disabled=True (рекомендуется) Отключите network для untrusted кода
Read-Only FS Не настраиваемый Не прямо настраиваемый read_only=True Docker read-only с tmpfs для /tmp
PID Limit Не exposed Не exposed pids_limit=64 Предотвращает fork bombs
Working Directory /home/user /root или custom /workspace (mounted) E2B имеет persistent home directory

Типовые ошибки

Ошибка 1: Package Installation Latency в E2B

Влияние: 10-30 секундных задержек на выполнение при установке пакетов во время runtime.

Неправильно (runtime package install):

from e2b_code_interpreter import Sandbox

with Sandbox() as sandbox:
    # This runs on EVERY execution - wastes 15+ seconds
    sandbox.run_code("import subprocess; subprocess.run(['pip', 'install', 'pandas', 'numpy', 'scikit-learn'])")
    sandbox.run_code("import pandas as pd; df = pd.read_csv('data.csv')")

Правильно (используйте custom template):

# e2b.Dockerfile
FROM e2bdev/code-interpreter:latest
RUN pip install --no-cache-dir pandas==2.1.4 numpy==1.26.3 scikit-learn==1.4.0
# Build once, reuse forever
e2b template build -d e2b.Dockerfile -n data-science-v1
from e2b_code_interpreter import Sandbox

with Sandbox(template="data-science-v1") as sandbox:
    # Packages already installed - starts immediately
    sandbox.run_code("import pandas as pd; df = pd.read_csv('data.csv')")

Ошибка 2: Unbounded Output Truncation

Влияние: LLM получает неполный выход, принимает неправильные решения, agent loop fails.

Неправильно (нет output handling):

execution = sandbox.run_code("for i in range(100000): print(i)")
# execution.text is 1MB+, exceeds LLM context, gets truncated randomly
agent_context = f"Output: {execution.text}"

Правильно (structured truncation):

def truncate_output(text: str, max_chars: int = 4000) -> str:
    if len(text) <= max_chars:
        return text
    return text[:max_chars] + f"\n... [truncated {len(text) - max_chars} chars, total {len(text)}]"

execution = sandbox.run_code("for i in range(100000): print(i)")
agent_context = f"Output:\n{truncate_output(execution.text)}"

Ошибка 3: Игнорирование статуса выполнения в Agent Loop

Влияние: Агент продолжает при failed executions, производит неправильные результаты.

Неправильно (нет error handling):

result = sandbox.run_code("import nonexistent_module")
# result.error exists but is ignored
next_code = generate_next_step(result.text)  # Proceeds with empty/error output

Правильно (error-aware iteration):

result = sandbox.run_code("import nonexistent_module")

if result.error:
    error_context = f"Error: {result.error.name}: {result.error.value}"
    next_prompt = f"Previous execution failed:\n{error_context}\n\nHow to fix this?"
    next_code = llm.generate(next_prompt)
else:
    next_code = generate_next_step(result.text)

Ошибка 4: Docker Security Misconfiguration

Влияние: Уязвимости container escape, privilege escalation, host compromise.

Неправильно (запуск как root с capabilities):

container = client.containers.run(
    'python:3.11-slim',
    command='python -c "import os; os.system(\'rm -rf /host/*\')"',
    volumes={'/': {'bind': '/host', 'mode': 'rw'}},  # CATASTROPHIC
    user='root',  # Unnecessary privilege
)

Правильно (minimal privileges):

container = client.containers.run(
    'python:3.11-slim',
    command=['python', '-c', 'print("safe code")'],
    volumes={'/tmp/sandbox_workspace': {'bind': '/workspace', 'mode': 'ro'}},
    user='nobody',  # Non-root user
    cap_drop=['ALL'],  # Drop all capabilities
    security_opt=['no-new-privileges'],
    network_disabled=True,
    read_only=True,
    tmpfs={'/tmp': 'size=64m'},
)

Ошибка 5: Modal Image Rebuilds на каждый Deploy

Влияние: 5-10 минутные задержки deployment для неизменных зависимостей.

Неправильно (inline pip install):

import modal

app = modal.App("my-agent")

@app.function(timeout=300)
def execute_code(code: str):
    import subprocess
    # This reinstalls packages on every cold start
    subprocess.run(['pip', 'install', 'pandas', 'numpy'])
    exec(code)

Правильно (pre-built Image):

import modal

app = modal.App("my-agent")

# Build image once at deploy time
image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install(["pandas==2.1.4", "numpy==1.26.3"])
)

@app.function(image=image, timeout=300)
def execute_code(code: str):
    # Packages already installed in image
    exec(code)

E2B реализация

Базовый Code Interpreter

from e2b_code_interpreter import Sandbox

# Synchronous execution
with Sandbox() as sandbox:
    execution = sandbox.run_code("print('Hello from E2B')")
    print(execution.text)  # "Hello from E2B"

    if execution.error:
        print(f"Error: {execution.error.name}: {execution.error.value}")

Async выполнение с Timeout

import asyncio
from e2b_code_interpreter import AsyncSandbox

async def run_analysis():
    async with AsyncSandbox() as sandbox:
        execution = await sandbox.run_code(
            "import time; time.sleep(5); print('done')",
        )
        return execution.text

# Run with timeout
result = asyncio.wait_for(run_analysis(), timeout=10)

File Upload и Download

from e2b_code_interpreter import Sandbox

with Sandbox() as sandbox:
    # Upload local file to sandbox
    with open('local_data.csv', 'rb') as f:
        sandbox.upload_file(f, '/home/user/data.csv')

    # Process the file
    execution = sandbox.run_code("""
import pandas as pd
df = pd.read_csv('/home/user/data.csv')
summary = df.describe().to_json()
print(summary)
""")

    # Download generated file
    sandbox.run_code("df.to_csv('/home/user/output.csv')")
    output_bytes = sandbox.download_file('/home/user/output.csv')

    with open('local_output.csv', 'wb') as f:
        f.write(output_bytes)

Package Installation в Session

from e2b_code_interpreter import Sandbox

with Sandbox() as sandbox:
    # Install packages (persists for this sandbox session)
    sandbox.run_code("""
import subprocess
subprocess.run(['pip', 'install', '-q', 'matplotlib', 'seaborn'])
""")

    # Use installed packages
    execution = sandbox.run_code("""
import matplotlib.pyplot as plt
import numpy as np

x = np.linspace(0, 10, 100)
plt.plot(x, np.sin(x))
plt.savefig('/home/user/plot.png')
print('Plot saved')
""")

    # Download the plot
    plot_data = sandbox.download_file('/home/user/plot.png')

Production Wrapper с Error Handling

import asyncio
from e2b_code_interpreter import AsyncSandbox
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ExecutionStatus(Enum):
    SUCCESS = "success"
    ERROR = "error"
    TIMEOUT = "timeout"

@dataclass
class ExecutionResult:
    status: ExecutionStatus
    stdout: str
    stderr: str
    execution_time_ms: int
    error_message: Optional[str] = None

class E2BExecutor:
    def __init__(self, api_key: str = None, timeout_seconds: int = 60, template: str = "base"):
        self.api_key = api_key
        self.timeout_seconds = timeout_seconds
        self.template = template
        self._sandbox: Optional[AsyncSandbox] = None

    async def __aenter__(self):
        self._sandbox = await AsyncSandbox.create(
            api_key=self.api_key,
            template=self.template,
            timeout=self.timeout_seconds * 10,
        )
        return self

    async def __aexit__(self, *args):
        if self._sandbox:
            await self._sandbox.kill()

    async def run(self, code: str, timeout: int = None) -> ExecutionResult:
        import time

        if not self._sandbox:
            raise RuntimeError("Sandbox not initialized - use as context manager")

        start = time.time()

        try:
            execution = await asyncio.wait_for(
                self._sandbox.run_code(code),
                timeout=timeout or self.timeout_seconds,
            )

            elapsed_ms = int((time.time() - start) * 1000)

            stdout_parts = [log for log in execution.logs.stdout]
            stderr_parts = [log for log in execution.logs.stderr]

            if execution.error:
                return ExecutionResult(
                    status=ExecutionStatus.ERROR,
                    stdout='\n'.join(stdout_parts),
                    stderr=execution.error.value,
                    execution_time_ms=elapsed_ms,
                    error_message=f"{execution.error.name}: {execution.error.value}",
                )

            return ExecutionResult(
                status=ExecutionStatus.SUCCESS,
                stdout='\n'.join(stdout_parts),
                stderr='\n'.join(stderr_parts),
                execution_time_ms=elapsed_ms,
            )

        except asyncio.TimeoutError:
            return ExecutionResult(
                status=ExecutionStatus.TIMEOUT,
                stdout='',
                stderr='',
                execution_time_ms=int((time.time() - start) * 1000),
                error_message=f"Execution timed out after {timeout or self.timeout_seconds}s",
            )

Custom Template с Pre-Installed Packages

# e2b.Dockerfile
FROM e2bdev/code-interpreter:latest

# System dependencies
RUN apt-get update && apt-get install -y \
    libpq-dev \
    ffmpeg \
    && rm -rf /var/lib/apt/lists/*

# Python packages at build time (no runtime delay)
RUN pip install --no-cache-dir \
    pandas==2.1.4 \
    numpy==1.26.3 \
    scikit-learn==1.4.0 \
    matplotlib==3.8.2 \
    seaborn==0.13.2 \
    duckdb==0.9.2

# Initialization script
COPY init.py /home/user/init.py
RUN python /home/user/init.py
# Build template (run once)
e2b template build -d e2b.Dockerfile -n data-science-v2
# Use the custom template
from e2b_code_interpreter import Sandbox

with Sandbox(template="data-science-v2") as sandbox:
    # pandas, numpy, etc. already available - no wait
    result = sandbox.run_code("import pandas as pd; print(pd.__version__)")
    print(result.text)  # "2.1.4"

Modal реализация

Базовое функциональное выполнение

import modal

app = modal.App("code-execution-agent")

@app.function(timeout=300)
def execute_code(code: str) -> dict:
    import sys
    import io
    import traceback
    import time

    stdout_capture = io.StringIO()
    stderr_capture = io.StringIO()

    namespace = {'__builtins__': __builtins__}

    start = time.time()

    try:
        sys.stdout = stdout_capture
        sys.stderr = stderr_capture

        exec(code, namespace)

        sys.stdout = sys.__stdout__
        sys.stderr = sys.__stderr__

        return {
            'status': 'success',
            'stdout': stdout_capture.getvalue(),
            'stderr': stderr_capture.getvalue(),
            'execution_time_ms': int((time.time() - start) * 1000),
        }

    except Exception as e:
        sys.stdout = sys.__stdout__
        sys.stderr = sys.__stderr__

        return {
            'status': 'error',
            'stdout': stdout_capture.getvalue(),
            'stderr': stderr_capture.getvalue() + traceback.format_exc(),
            'execution_time_ms': int((time.time() - start) * 1000),
            'error': str(e),
        }

GPU-Accelerated Execution

import modal

gpu_image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install([
        "torch==2.2.0",
        "transformers==4.37.0",
        "accelerate",
    ])
)

app = modal.App("gpu-agent")

@app.function(
    image=gpu_image,
    gpu="A10G",  # 24GB VRAM
    timeout=900,
    memory=32768,  # 32GB RAM
)
def run_llm_inference(prompt: str, model_name: str = "mistralai/Mistral-7B-Instruct-v0.2") -> str:
    from transformers import AutoModelForCausalLM, AutoTokenizer
    import torch

    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.float16,
        device_map="auto",
        load_in_4bit=True,
    )

    inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
    outputs = model.generate(**inputs, max_new_tokens=512)

    return tokenizer.decode(outputs[0], skip_special_tokens=True)

Persistent State с Dict

import modal
from modal import Dict

app = modal.App("stateful-agent")
agent_state = Dict.from_name("agent-execution-state", create_if_missing=True)

data_science_image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install(["pandas", "numpy", "scikit-learn"])
)

@app.function(image=data_science_image, timeout=600)
def stateful_execution(session_id: str, code: str) -> dict:
    import pickle

    # Load previous state
    namespace = {}
    if session_id in agent_state:
        state_bytes = agent_state[session_id]
        namespace = pickle.loads(state_bytes)

    # Execute code with preserved variables
    import sys
    import io

    stdout_capture = io.StringIO()
    sys.stdout = stdout_capture

    try:
        exec(code, namespace)
        sys.stdout = sys.__stdout__

        # Persist variables for next execution
        vars_to_save = {
            k: v for k, v in namespace.items()
            if not k.startswith('_') and k != '__builtins__'
        }

        try:
            agent_state[session_id] = pickle.dumps(vars_to_save)
        except Exception:
            pass  # Some objects aren't picklable

        return {
            'status': 'success',
            'stdout': stdout_capture.getvalue(),
        }

    except Exception as e:
        sys.stdout = sys.__stdout__
        return {
            'status': 'error',
            'stderr': str(e),
        }

Client-Side Executor для Agent Loop

import modal
import asyncio
import uuid
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ExecutionStatus(Enum):
    SUCCESS = "success"
    ERROR = "error"

@dataclass
class ExecutionResult:
    status: ExecutionStatus
    stdout: str
    stderr: str
    execution_time_ms: int
    error_message: Optional[str] = None

class ModalExecutor:
    def __init__(self, session_id: str = None):
        self.session_id = session_id or str(uuid.uuid4())
        # Import Modal app (assumes modal app is deployed)
        from modal import Function
        self.execute_fn = Function.lookup("code-execution-agent", "execute_code")

    async def run(self, code: str, timeout: int = 60) -> ExecutionResult:
        loop = asyncio.get_event_loop()

        raw_result = await loop.run_in_executor(
            None,
            lambda: self.execute_fn.remote(code)
        )

        status = ExecutionStatus.SUCCESS if raw_result['status'] == 'success' else ExecutionStatus.ERROR

        return ExecutionResult(
            status=status,
            stdout=raw_result.get('stdout', ''),
            stderr=raw_result.get('stderr', ''),
            execution_time_ms=raw_result.get('execution_time_ms', 0),
            error_message=raw_result.get('error'),
        )

Secrets Management

import modal

app = modal.App("secure-agent")

# Store secrets via Modal dashboard or CLI: modal secret create db-credentials
@app.function(
    secrets=[modal.Secret.from_name("db-credentials")],
    timeout=300,
)
def execute_with_secrets(code: str) -> dict:
    import os

    # Secrets available as environment variables
    db_url = os.environ["DATABASE_URL"]
    api_key = os.environ["API_KEY"]

    namespace = {
        '__builtins__': __builtins__,
        'DB_URL': db_url,
        'API_KEY': api_key,
    }

    exec(code, namespace)

    return {'status': 'success'}

Интеграция с LangGraph/LangChain

LangChain Tool Wrapper для E2B

from langchain.tools import StructuredTool
from e2b_code_interpreter import Sandbox
from pydantic import BaseModel, Field

class CodeExecutionInput(BaseModel):
    code: str = Field(description="Python code to execute")

def execute_python_code(code: str) -> str:
    """Execute Python code in a secure E2B sandbox."""
    with Sandbox() as sandbox:
        execution = sandbox.run_code(code)

        if execution.error:
            return f"Error: {execution.error.name}: {execution.error.value}"

        return execution.text or "Code executed successfully (no output)"

python_executor_tool = StructuredTool.from_function(
    func=execute_python_code,
    name="python_executor",
    description="Execute Python code in a secure sandbox. Use for data analysis, calculations, file processing.",
    args_schema=CodeExecutionInput,
)

# Use in LangChain agent
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-20250514")

agent = create_tool_calling_agent(
    llm=llm,
    tools=[python_executor_tool],
    prompt="You are a data analysis assistant. Use the python_executor tool to run code.",
)

agent_executor = AgentExecutor(agent=agent, tools=[python_executor_tool])

result = agent_executor.invoke({
    "input": "Calculate the sum of squares from 1 to 100"
})

LangGraph Node для Modal Execution

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    code: str
    execution_result: str
    iterations: int

def generate_code_node(state: AgentState) -> AgentState:
    from langchain_anthropic import ChatAnthropic

    llm = ChatAnthropic(model="claude-opus-4-20250514")

    messages = state["messages"]
    response = llm.invoke(messages + [
        {"role": "user", "content": "Generate Python code to solve the task."}
    ])

    return {
        "code": response.content,
        "iterations": state.get("iterations", 0) + 1,
    }

def execute_code_node(state: AgentState) -> AgentState:
    from modal import Function

    execute_fn = Function.lookup("code-execution-agent", "execute_code")
    result = execute_fn.remote(state["code"])

    return {
        "execution_result": result.get("stdout", result.get("stderr", "")),
    }

def should_continue(state: AgentState) -> str:
    if state.get("iterations", 0) >= 5:
        return "end"

    if "error" in state.get("execution_result", "").lower():
        return "generate_code"

    return "end"

# Build LangGraph
workflow = StateGraph(AgentState)

workflow.add_node("generate_code", generate_code_node)
workflow.add_node("execute_code", execute_code_node)

workflow.set_entry_point("generate_code")
workflow.add_edge("generate_code", "execute_code")
workflow.add_conditional_edges(
    "execute_code",
    should_continue,
    {
        "generate_code": "generate_code",
        "end": END,
    }
)

app = workflow.compile()

# Run the agent
result = app.invoke({
    "messages": [{"role": "user", "content": "Calculate factorial of 10"}],
    "iterations": 0,
})

Async E2B Tool для LangChain

from langchain.tools import BaseTool
from e2b_code_interpreter import AsyncSandbox
import asyncio

class AsyncPythonExecutor(BaseTool):
    name = "async_python_executor"
    description = "Execute Python code asynchronously in a secure sandbox"

    async def _arun(self, code: str) -> str:
        async with AsyncSandbox() as sandbox:
            execution = await sandbox.run_code(code)

            if execution.error:
                return f"Error: {execution.error.name}: {execution.error.value}"

            return execution.text or "Success (no output)"

    def _run(self, code: str) -> str:
        return asyncio.run(self._arun(code))

# Use with async LangChain agent
async def run_agent():
    from langchain.agents import AgentExecutor, create_tool_calling_agent
    from langchain_anthropic import ChatAnthropic

    llm = ChatAnthropic(model="claude-opus-4-20250514")
    tools = [AsyncPythonExecutor()]

    agent = create_tool_calling_agent(llm=llm, tools=tools, prompt="...")
    agent_executor = AgentExecutor(agent=agent, tools=tools)

    result = await agent_executor.ainvoke({
        "input": "Analyze the dataset and compute summary statistics"
    })

    return result

# asyncio.run(run_agent())

Performance & Benchmarks

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

Cold Start Performance

Холодный старт sandbox значительно влияет на responsiveness агента в intermittent рабочих нагрузках:

  • E2B с базовым образом: Холодный старт обычно 2-5 секунд; warm reuse <100ms
  • E2B с пользовательским шаблоном: Холодный старт 3-8 секунд в зависимости от размера образа
  • Modal с slim Python образом: Холодный старт 1-3 секунды; warm контейнеры <50ms
  • Modal с GPU образом (T4): Холодный старт 5-15 секунд из-за CUDA инициализации
  • Docker локальное выполнение: Создание контейнера 500ms-2s; pooled контейнеры <100ms

Стратегии оптимизации:

  • E2B: Держите sandbox живым при множественных выполнениях (context manager pattern)
  • Modal: Используйте keep_warm параметр для поддержания горячих контейнеров во время peak hours
  • Docker: Pre-warm container pool (5-10 контейнеров) для sub-100ms latency

Execution Throughput

Concurrency capacity варьируется по платформам:

  • E2B: Обычно обрабатывает десятки concurrent sandboxes за аккаунт; apply rate limits к API calls
  • Modal: Auto-scales к сотням concurrent контейнеров; limited by account quotas
  • Docker self-hosted: Limited by host ресурсами (CPU cores, память); typical single-host capacity 20-50 concurrent контейнеров

Memory и CPU Limits Impact

Resource constraints влияют на success rate и iteration count:

  • 512MB memory limit: Достаточно для большинства data analysis задач с datasets <100MB
  • 2GB memory limit: Поддерживает pandas DataFrames до 500MB, moderate ML inference
  • 4GB+ memory limit: Требуется для large-scale data processing, transformer model inference
  • CPU quota 50%: Адекватно для I/O-bound задач; compute-intensive алгоритмы могут timeout
  • CPU quota 100%+: Рекомендуется для numerical computing, ML training

Undersized лимиты вызывают OOM ошибки требующие agent retry logic; oversized лимиты увеличивают cost без throughput benefit.

Network Latency

API call overhead для remote execution:

  • E2B: Round-trip API latency 50-200ms в зависимости от региона; execution time дополнительный
  • Modal: Function invocation latency 30-100ms; GPU функции имеют выше cold start overhead
  • Docker local: Near-zero network latency; limited по container startup и execution time

Для latency-sensitive приложений (<100ms response requirement), local Docker execution или persistent E2B sandboxes предпочтительны.

Cost Efficiency Comparison

Cost структура варьируется значительно:

  • E2B: Pay-per-minute sandbox usage; бесплатный уровень 100 часов/месяц; typical production cost $0.001-0.005 per execution
  • Modal: Compute-minute pricing; CPU рабочие нагрузки ~$0.01-0.05 per execution; GPU рабочие нагрузки $0.10-0.50 per execution
  • Docker self-hosted: Fixed инфраструктура cost (EC2, GCE); marginal cost per execution near-zero; cost-effective at scale (>10K executions/day)

Для experimentation и low-volume рабочих нагрузок (<1000 executions/day), E2B бесплатный уровень наиболее экономичен. Для GPU-intensive рабочих нагрузок, Modal's per-second billing эффективен для intermittent usage. Для high-volume production (>10K executions/day), self-hosted Docker становится cost-effective несмотря на operational overhead.