DeltaBox 毫秒级沙箱：有状态 Agent 基础设施解析

# 对比: 传统沙箱在 Agent 场景下的延迟分解
import time

def measure_traditional_sandbox_overhead():
    """测量传统沙箱在 Agent 场景下的总延迟"""
    
    # Docker 容器方案
    docker_steps = {
        "容器启动": 300,        # ms
        "环境初始化": 200,      # ms (安装依赖、配置网络)
        "状态恢复": 500,        # ms (从数据库加载历史)
        "代码执行": 100,        # ms
        "结果返回": 50,         # ms
    }
    # 容器方案: 隔离弱, 但总延迟 ~1150ms
    
    # Firecracker 方案
    firecracker_steps = {
        "VM 冷启动": 200,       # ms
        "内核启动": 150,        # ms
        "环境初始化": 200,      # ms
        "状态恢复": 2000,       # ms (从快照加载完整内存)
        "代码执行": 100,        # ms
        "结果返回": 50,         # ms
    }
    # Firecracker 方案: 隔离强, 但总延迟 ~2700ms
    
    total_docker = sum(docker_steps.values())
    total_fc = sum(firecracker_steps.values())
    
    print(f"Docker 总延迟: {total_docker}ms")
    print(f"Firecracker 总延迟: {total_fc}ms")
    print(f"Agent 用户体验阈值: < 200ms")
    print(f"Docker {'达标' if total_docker < 200 else '超标'}")
    print(f"Firecracker {'达标' if total_fc < 200 else '超标'}")

measure_traditional_sandbox_overhead()

# DeltaBox 预热 VM 池 + 分层状态管理实现
import time
import threading
from dataclasses import dataclass
from typing import Dict, Optional

@dataclass
class AgentState:
    hot: dict          # 热状态: 当前执行上下文
    warm: dict         # 温状态: 近期对话摘要
    cold_ref: str      # 冷状态: 持久化存储引用

class PreWarmedVMPool:
    """预热微 VM 池: 毫秒级沙箱启动"""
    
    def __init__(self, pool_size=10, max_pool_size=20):
        self.pool_size = pool_size
        self.max_pool_size = max_pool_size
        self.available_vms = []
        self.lock = threading.Lock()
        self._warm_up()
    
    def _warm_up(self):
        """预启动一批最小化 VM"""
        print(f"正在预热 {self.pool_size} 个微 VM...")
        for i in range(self.pool_size):
            vm = self._create_minimal_vm()
            vm.pause()  # 暂停而不是关闭
            self.available_vms.append(vm)
        print(f"预热完成: {len(self.available_vms)} 个 VM 就绪")
    
    def _create_minimal_vm(self):
        """创建最小化 VM: 只加载内核和基础运行时"""
        return MicroVM(kernel_preloaded=True, minimal_runtime=True)
    
    def acquire_vm(self) -> "MicroVM":
        """获取一个预热的 VM"""
        with self.lock:
            if self.available_vms:
                vm = self.available_vms.pop(0)
                vm.resume()  # 从暂停状态恢复: 5-20ms
                return vm
            else:
                # 池空了: 现场创建（较慢 100-500ms）
                return self._create_minimal_vm()
    
    def release_vm(self, vm: "MicroVM"):
        """释放 VM 回到池中"""
        with self.lock:
            if len(self.available_vms) < self.max_pool_size:
                vm.reset()
                vm.pause()
                self.available_vms.append(vm)
            else:
                vm.terminate()

class LayeredStateManager:
    """分层状态管理器"""
    
    def __init__(self, fast_storage, persistent_storage):
        self.fast_storage = fast_storage      # Redis / Memcached
        self.persistent_storage = persistent_storage  # PostgreSQL / S3
    
    def save_state(self, agent_id: str, state: AgentState):
        """保存分层状态"""
        # 热状态: 直接保存在本地内存（与 VM 同生命周期）
        # 温状态: 保存到快速存储
        self.fast_storage.set(
            f"agent:{agent_id}:warm",
            state.warm,
            expire=3600  # 1 小时过期
        )
        # 冷状态: 保存到持久化存储
        cold_ref = self.persistent_storage.store(state.cold_ref)
        return cold_ref
    
    def load_state(self, agent_id: str) -> AgentState:
        """恢复分层状态"""
        warm = self.fast_storage.get(f"agent:{agent_id}:warm") or {}
        cold_ref = self.persistent_storage.get_latest_ref(agent_id)
        return AgentState(hot={}, warm=warm, cold_ref=cold_ref)

class MicroVM:
    def __init__(self, kernel_preloaded=False, minimal_runtime=False):
        self.kernel_preloaded = kernel_preloaded
        self.minimal_runtime = minimal_runtime
        self.state = {}
    
    def pause(self): self.state["running"] = False
    def resume(self): self.state["running"] = True
    def reset(self): self.state = {"running": False}
    def terminate(self): pass

# 使用示例
pool = PreWarmedVMPool(pool_size=5)
vm = pool.acquire_vm()
print(f"获取 VM 耗时: < 20ms")
pool.release_vm(vm)

# DeltaBox 架构: 四层设计的核心接口
from abc import ABC, abstractmethod
from typing import Dict, Optional

# === 基础设施层 ===
class InfrastructureLayer(ABC):
    @abstractmethod
    def allocate_vm(self, resources: dict) -> str:
        """分配一个微 VM，返回 VM ID"""
        pass
    
    @abstractmethod
    def get_prewarmed_vm(self) -> str:
        """从预热池获取一个 VM"""
        pass

# === 沙箱管理层 ===
class SandboxManager(ABC):
    @abstractmethod
    def create_sandbox(self, agent_id: str, vm_id: str, config: dict) -> str:
        """在 VM 上创建沙箱，返回沙箱 ID"""
        pass
    
    @abstractmethod
    def destroy_sandbox(self, sandbox_id: str):
        """销毁沙箱"""
        pass
    
    @abstractmethod
    def set_resource_quota(self, sandbox_id: str, cpu: float, memory_mb: int):
        """设置资源配额"""
        pass

# === 状态管理层 ===
class StateManager(ABC):
    @abstractmethod
    def save_hot_state(self, sandbox_id: str, state: dict):
        """保存热状态（内存中）"""
        pass
    
    @abstractmethod
    def save_warm_state(self, agent_id: str, state: dict, ttl: int = 3600):
        """保存温状态到快速存储"""
        pass
    
    @abstractmethod
    def save_cold_state(self, agent_id: str, state: dict) -> str:
        """保存冷状态到持久化存储，返回引用"""
        pass
    
    @abstractmethod
    def restore_state(self, agent_id: str) -> dict:
        """恢复分层状态"""
        pass

# === 调度层 ===
class Scheduler(ABC):
    @abstractmethod
    def schedule_agent(self, agent_request: dict) -> str:
        """调度 Agent 请求，返回沙箱 ID"""
        pass
    
    @abstractmethod
    def rebalance(self):
        """重新平衡负载"""
        pass

# DeltaBox 协调器: 将四层组合在一起
class DeltaBoxOrchestrator:
    def __init__(self, infra, sandbox_mgr, state_mgr, scheduler):
        self.infra = infra
        self.sandbox_mgr = sandbox_mgr
        self.state_mgr = state_mgr
        self.scheduler = scheduler
    
    def handle_agent_request(self, request: dict) -> dict:
        """处理 Agent 请求的完整流程"""
        # 1. 调度层: 决定分配策略
        sandbox_id = self.scheduler.schedule_agent(request)
        
        # 2. 状态管理层: 恢复 Agent 状态
        state = self.state_mgr.restore_state(request["agent_id"])
        
        # 3. 沙箱管理层: 注入状态并执行
        result = self.sandbox_mgr.execute(sandbox_id, request["code"], state)
        
        # 4. 状态管理层: 保存更新后的状态
        self.state_mgr.save_hot_state(sandbox_id, result["new_hot_state"])
        self.state_mgr.save_warm_state(request["agent_id"], result["new_warm_state"])
        
        return result

# DeltaBox 状态管理完整实现
import json
import time
from datetime import datetime
from typing import List, Dict, Optional
import hashlib

class ToolCallRecord:
    """工具调用记录"""
    def __init__(self, tool_name: str, input_data: str, output: str, duration_ms: int):
        self.tool_name = tool_name
        self.input_data = input_data
        self.output = output[:500]  # 截断输出，避免过大
        self.duration_ms = duration_ms
        self.timestamp = datetime.now().isoformat()

class DeltaBoxStateManager:
    """DeltaBox 分层状态管理器"""
    
    def __init__(self, fast_storage, persistent_storage, max_warm_rounds=50):
        self.fast_storage = fast_storage
        self.persistent_storage = persistent_storage
        self.max_warm_rounds = max_warm_rounds  # 温状态保留的最近对话轮数
    
    def on_conversation_turn(self, agent_id: str, turn_data: dict):
        """处理一轮对话完成后的状态更新"""
        agent_id_str = str(agent_id)
        
        # 1. 更新热状态（内存中，由 VM 直接管理）
        hot_state = turn_data.get("hot_state", {})
        
        # 2. 更新温状态
        warm_state = self._build_warm_state(turn_data)
        self.fast_storage.set(
            f"deltabox:{agent_id_str}:warm",
            json.dumps(warm_state),
            expire=7200  # 2 小时过期
        )
        
        # 3. 批量更新冷状态
        cold_delta = self._build_cold_delta(turn_data)
        self.persistent_storage.append_delta(agent_id_str, cold_delta)
        
        # 4. 检查是否需要做完整快照
        if self._should_snapshot(agent_id_str):
            self._create_full_snapshot(agent_id_str)
    
    def _build_warm_state(self, turn_data: dict) -> dict:
        """构建温状态"""
        return {
            "recent_messages": turn_data.get("recent_messages", [])[-self.max_warm_rounds:],
            "tool_calls": [
                {
                    "tool": tc.tool_name,
                    "duration_ms": tc.duration_ms,
                    "timestamp": tc.timestamp,
                }
                for tc in turn_data.get("tool_calls", [])
            ],
            "last_updated": datetime.now().isoformat(),
        }
    
    def _build_cold_delta(self, turn_data: dict) -> dict:
        """构建冷状态增量"""
        return {
            "full_messages": turn_data.get("recent_messages", []),
            "full_tool_calls": turn_data.get("tool_calls", []),
            "file_changes": turn_data.get("file_changes", {}),
            "timestamp": datetime.now().isoformat(),
        }
    
    def restore(self, agent_id: str) -> dict:
        """恢复 Agent 状态"""
        agent_id_str = str(agent_id)
        
        # 从温状态恢复（快速）
        warm_json = self.fast_storage.get(f"deltabox:{agent_id_str}:warm")
        warm_state = json.loads(warm_json) if warm_json else {}
        
        # 从冷状态获取引用（后台加载）
        cold_ref = self.persistent_storage.get_latest_ref(agent_id_str)
        
        return {
            "hot": {},  # 热状态由 VM 重建
            "warm": warm_state,
            "cold_ref": cold_ref,
            "restore_time_ms": time.time() * 1000,
        }
    
    def _should_snapshot(self, agent_id: str) -> bool:
        """检查是否需要创建完整快照"""
        delta_count = self.persistent_storage.get_delta_count(agent_id)
        return delta_count > 100  # 超过 100 个增量就做完整快照
    
    def _create_full_snapshot(self, agent_id: str):
        """创建完整快照"""
        all_deltas = self.persistent_storage.get_all_deltas(agent_id)
        snapshot = self._merge_deltas(all_deltas)
        snapshot_ref = self.persistent_storage.store_snapshot(agent_id, snapshot)
        self.persistent_storage.clear_deltas(agent_id, snapshot_ref)

# DeltaBox 多租户调度器实现
import time
import threading
from typing import Dict, List, Optional
from dataclasses import dataclass, field

@dataclass
class TenantQuota:
    """租户资源配额"""
    cpu_cores: float
    memory_mb: int
    bandwidth_mbps: float
    storage_mb: int
    max_concurrent_agents: int

@dataclass
class PriorityLevel:
    """优先级配置"""
    name: str
    target_latency_ms: int
    availability: float
    quota: TenantQuota
    prewarm_priority: int  # 预热池分配优先级（越高越优先）

class TokenBucket:
    """令牌桶: 资源配额管理"""
    def __init__(self, rate: float, capacity: float):
        self.rate = rate          # 每秒生成的令牌数
        self.capacity = capacity  # 最大令牌数
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = threading.Lock()
    
    def consume(self, count: float) -> bool:
        """尝试消费令牌"""
        with self.lock:
            now = time.time()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now
            
            if self.tokens >= count:
                self.tokens -= count
                return True
            return False

class MultiTenantScheduler:
    """多租户调度器"""
    
    def __init__(self):
        self.priority_configs = {
            "premium": PriorityLevel("Premium", 50, 0.9999,
                TenantQuota(4, 4096, 100, 1000, 100), 3),
            "standard": PriorityLevel("Standard", 100, 0.999,
                TenantQuota(2, 2048, 50, 500, 50), 2),
            "free": PriorityLevel("Free", 200, 0.99,
                TenantQuota(1, 1024, 10, 100, 10), 1),
        }
        self.tenant_buckets: Dict[str, TokenBucket] = {}
        self.node_load: Dict[str, dict] = {}
    
    def register_tenant(self, tenant_id: str, priority: str):
        """注册租户并分配配额"""
        config = self.priority_configs[priority]
        self.tenant_buckets[tenant_id] = TokenBucket(
            rate=config.quota.cpu_cores,
            capacity=config.quota.cpu_cores * 2,
        )
    
    def schedule(self, tenant_id: str, request: dict) -> dict:
        """调度 Agent 请求"""
        # 1. 检查配额
        bucket = self.tenant_buckets.get(tenant_id)
        if not bucket or not bucket.consume(0.1):
            return {"error": "资源配额已用完，请稍后重试"}
        
        # 2. 选择最优节点
        best_node = self._select_best_node(tenant_id)
        if not best_node:
            return {"error": "没有可用的节点"}
        
        # 3. 获取预热 VM
        vm = best_node.get_prewarmed_vm()
        
        # 4. 恢复状态并执行
        result = self._execute_on_node(best_node, vm, tenant_id, request)
        
        return result
    
    def _select_best_node(self, tenant_id: str) -> Optional["Node"]:
        """选择最优节点: 就近 + 负载感知"""
        # 简化版: 选择负载最低的节点
        available = [n for n in self.node_load.values() if n["available_capacity"] > 0.2]
        if not available:
            return None
        return min(available, key=lambda n: n["current_load"])

class Node:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.prewarm_pool = []
        self.current_load = 0.0
    
    def get_prewarmed_vm(self):
        if self.prewarm_pool:
            return self.prewarm_pool.pop(0)
        return None
    
    @property
    def available_capacity(self):
        return max(0, 1.0 - self.current_load)

# 沙箱方案对比测试脚本
import time
import subprocess

def benchmark_sandbox_startup():
    """对比不同沙箱方案的启动时间"""
    results = {}
    
    # 1. Docker 容器启动
    start = time.time()
    subprocess.run(["docker", "run", "--rm", "python:3.11-slim", "echo", "hello"],
                   capture_output=True, timeout=30)
    results["docker"] = (time.time() - start) * 1000
    
    # 2. Firecracker VM 启动（需要 Firecracker 安装）
    start = time.time()
    # firecracker --api-sock /tmp/firecracker.socket --config-file vm.json
    # 简化模拟
    results["firecracker"] = 200  # ms (典型值)
    
    # 3. DeltaBox 预热 VM 恢复
    start = time.time()
    # DeltaBox 从预热池恢复: 5-20ms
    results["deltabox_prewarmed"] = 12  # ms (典型值)
    
    # 4. DeltaBox 冷启动
    results["deltabox_cold"] = 80  # ms (典型值)
    
    print("沙箱启动时间对比:")
    for name, ms in sorted(results.items(), key=lambda x: x[1]):
        print(f"  {name}: {ms:.0f}ms")
    
    return results

def benchmark_state_restore():
    """对比不同沙箱方案的状态恢复时间"""
    results = {
        "docker_external_db": 500,        # ms (从外部数据库加载)
        "firecracker_snapshot": 5000,     # ms (从快照恢复完整 VM)
        "kata_snapshot": 10000,           # ms
        "deltabox_layered": 8,            # ms (分层恢复)
    }
    
    print("状态恢复时间对比:")
    for name, ms in sorted(results.items(), key=lambda x: x[1]):
        print(f"  {name}: {ms:.0f}ms")
    
    return results

benchmark_sandbox_startup()
benchmark_state_restore()

# DeltaBox 完整实现: 从基础设施到 Agent 执行
import asyncio
import time
import json
from typing import Dict, Optional
from dataclasses import dataclass

@dataclass
class AgentRequest:
    agent_id: str
    code: str
    priority: str  # "premium" | "standard" | "free"
    timeout_ms: int = 10000

@dataclass
class AgentResult:
    success: bool
    output: str
    error: str
    duration_ms: float
    sandbox_id: str

class DeltaBoxPlatform:
    """DeltaBox 沙箱平台: 完整实现"""
    
    def __init__(self, config: dict):
        # 初始化各层组件
        self.pool = PreWarmedVMPool(
            pool_size=config.get("pool_size", 10),
            max_pool_size=config.get("max_pool_size", 20),
        )
        self.state_mgr = DeltaBoxStateManager(
            fast_storage=config.get("redis_url", "redis://localhost:6379"),
            persistent_storage=config.get("db_url", "postgresql://localhost/deltabox"),
        )
        self.scheduler = MultiTenantScheduler()
        self.running = False
    
    async def start(self):
        """启动平台"""
        print("DeltaBox 平台启动中...")
        self.running = True
        # 启动后台任务: 监控、扩缩容、快照
        asyncio.create_task(self._monitor_loop())
        asyncio.create_task(self._autoscale_loop())
        asyncio.create_task(self._snapshot_loop())
        print("DeltaBox 平台已就绪")
    
    async def execute(self, request: AgentRequest) -> AgentResult:
        """执行 Agent 请求"""
        start_time = time.time()
        sandbox_id = f"sandbox-{int(time.time()*1000)}"
        
        try:
            # 1. 调度: 获取 VM
            vm = self.pool.acquire_vm()
            
            # 2. 恢复状态
            state = self.state_mgr.restore(request.agent_id)
            vm.inject_state(state)
            
            # 3. 执行代码
            result = await vm.execute_code(
                request.code,
                timeout_ms=request.timeout_ms,
            )
            
            # 4. 保存状态
            self.state_mgr.save(
                request.agent_id,
                vm.extract_state(),
            )
            
            duration = (time.time() - start_time) * 1000
            
            # 5. 释放 VM
            self.pool.release_vm(vm)
            
            return AgentResult(
                success=True,
                output=result.output,
                error="",
                duration_ms=duration,
                sandbox_id=sandbox_id,
            )
        
        except TimeoutError:
            return AgentResult(
                success=False, output="", error="执行超时",
                duration_ms=(time.time() - start_time) * 1000,
                sandbox_id=sandbox_id,
            )
        except Exception as e:
            return AgentResult(
                success=False, output="", error=str(e),
                duration_ms=(time.time() - start_time) * 1000,
                sandbox_id=sandbox_id,
            )
    
    async def _monitor_loop(self):
        """后台监控循环"""
        while self.running:
            await asyncio.sleep(5)
            # 收集指标: 预热池使用率、请求延迟、错误率
            metrics = self._collect_metrics()
            self._report_metrics(metrics)
    
    async def _autoscale_loop(self):
        """自动扩缩容循环"""
        while self.running:
            await asyncio.sleep(30)
            usage = self.pool.usage_ratio()
            if usage > 0.7:
                self.pool.expand(5)  # 增加 5 个预热 VM
            elif usage < 0.3:
                self.pool.shrink(3)  # 减少 3 个预热 VM
    
    async def _snapshot_loop(self):
        """增量快照循环"""
        while self.running:
            await asyncio.sleep(300)  # 每 5 分钟
            await self.state_mgr.flush_pending_deltas()

# 使用示例
async def main():
    platform = DeltaBoxPlatform({
        "pool_size": 10,
        "max_pool_size": 20,
        "redis_url": "redis://localhost:6379",
        "db_url": "postgresql://localhost/deltabox",
    })
    await platform.start()
    
    result = await platform.execute(AgentRequest(
        agent_id="user-123-session-456",
        code="print('Hello from DeltaBox!')",
        priority="standard",
    ))
    print(f"执行结果: {result.success}, 耗时: {result.duration_ms:.0f}ms")

# asyncio.run(main())

# DeltaBox Docker Compose 部署配置
version: '3.8'

services:
  # Redis: 温状态存储
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
    volumes:
      - redis-data:/data
  
  # PostgreSQL: 冷状态存储
  postgres:
    image: postgres:16-alpine
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: deltabox
      POSTGRES_USER: deltabox
      POSTGRES_PASSWORD: changeme
    volumes:
      - postgres-data:/var/lib/postgresql/data
  
  # MinIO: S3 兼容对象存储（文件快照）
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: deltabox
      MINIO_ROOT_PASSWORD: changeme
    volumes:
      - minio-data:/data
    command: server /data --console-address ":9001"
  
  # DeltaBox 调度器
  scheduler:
    build: ./scheduler
    ports:
      - "8080:8080"
    environment:
      REDIS_URL: redis://redis:6379
      DB_URL: postgresql://deltabox:changeme@postgres/deltabox
      POOL_SIZE: "10"
      MAX_POOL_SIZE: "20"
    depends_on:
      - redis
      - postgres
    deploy:
      resources:
        limits:
          cpus: "4"
          memory: 8G
        reservations:
          cpus: "2"
          memory: 4G

volumes:
  redis-data:
  postgres-data:
  minio-data: