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ом решает это через:
- Изоляция процесса: Предотвращение выхода из sandbox на host-систему
- Ограничения ресурсов: Принудительное применение лимитов CPU, памяти, диска и времени
- Ограничения возможностей: Блокировка системных вызовов (mount, ptrace, kernel modules)
- Границы файловой системы: Контроль доступа read/write к файлам host
- Сетевые политики: Ограничение или блокировка внешних соединений
- Санитизация выходных данных: Предотвращение атак инъекции через захваченный 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.