生产部署：监控、成本与可靠性

Docker 容器化生产部署

生产级 Dockerfile

生产环境的容器设计与开发环境有几个关键区别：使用非 root 用户、多阶段构建减小镜像体积、健康检查、资源限制：

# Dockerfile.production - 生产级 Computer Use 容器
FROM ubuntu:22.04 AS base

# 避免交互式安装
ENV DEBIAN_FRONTEND=noninteractive
ENV DISPLAY=:99

# 安装系统依赖（合并 RUN 减少层数）
RUN apt-get update && apt-get install -y --no-install-recommends \
    xvfb x11vnc fluxbox \
    python3.11 python3.11-pip \
    firefox-esr \
    curl wget \
    && rm -rf /var/lib/apt/lists/* \
    && ln -s /usr/bin/python3.11 /usr/local/bin/python3

# 创建专用用户（非 root）
RUN useradd -m -u 1001 -s /bin/bash agent \
    && mkdir -p /app /tmp/agent_output \
    && chown -R agent:agent /app /tmp/agent_output

FROM base AS python-deps

# 安装 Python 依赖（利用缓存层）
COPY requirements.txt /tmp/
RUN pip3 install --no-cache-dir -r /tmp/requirements.txt

FROM python-deps AS production

USER agent
WORKDIR /app

COPY --chown=agent:agent . .

# 健康检查：验证 Xvfb 进程是否运行
HEALTHCHECK --interval=30s --timeout=10s --start-period=15s --retries=3 \
    CMD pgrep Xvfb || exit 1

EXPOSE 5900 8080

ENTRYPOINT ["/app/entrypoint.sh"]

VNC 远程桌面 vs 无头模式对比

import os
import subprocess


class DisplayManager:
    """管理虚拟显示器，支持按需开关 VNC"""

    def __init__(self, display_num: int = 99, width: int = 1280, height: int = 800):
        self.display_num = display_num
        self.width = width
        self.height = height
        self._xvfb_proc = None
        self._vnc_proc = None

    def start(self, enable_vnc: bool = False):
        # 启动虚拟显示器
        self._xvfb_proc = subprocess.Popen([
            "Xvfb", f":{self.display_num}",
            "-screen", "0", f"{self.width}x{self.height}x24"
        ])
        os.environ["DISPLAY"] = f":{self.display_num}"

        if enable_vnc:
            self.start_vnc()

    def start_vnc(self, password: str = None, port: int = 5900):
        """按需启动 VNC 服务器（用于调试）"""
        cmd = [
            "x11vnc",
            "-display", f":{self.display_num}",
            "-rfbport", str(port),
            "-listen", "localhost",   # 只监听本地，配合 SSH 隧道
            "-xkb", "-forever"
        ]
        if password:
            cmd += ["-passwd", password]
        else:
            cmd.append("-nopw")
        self._vnc_proc = subprocess.Popen(cmd)

    def stop_vnc(self):
        """关闭 VNC（节约资源）"""
        if self._vnc_proc:
            self._vnc_proc.terminate()
            self._vnc_proc = None

Kubernetes 弹性伸缩

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: computer-use-agent
  namespace: ai-agents
spec:
  replicas: 3
  selector:
    matchLabels:
      app: computer-use-agent
  template:
    metadata:
      labels:
        app: computer-use-agent
    spec:
      containers:
      - name: agent
        image: your-registry/computer-use-agent:latest
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2000m"
            memory: "4Gi"
        env:
        - name: ANTHROPIC_API_KEY
          valueFrom:
            secretKeyRef:
              name: anthropic-secret
              key: api-key
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 30
---
# 基于队列长度自动伸缩
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: computer-use-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: computer-use-agent
  minReplicas: 1
  maxReplicas: 20
  metrics:
  - type: External
    external:
      metric:
        name: task_queue_length
      target:
        type: AverageValue
        averageValue: "5"  # 每个 Pod 处理 5 个任务

成本控制策略

Token 消耗优化

成本监控实现

from dataclasses import dataclass
from datetime import datetime, timedelta
import json


@dataclass
class CostRecord:
    task_id: str
    model: str
    input_tokens: int
    output_tokens: int
    screenshots_count: int
    timestamp: str

    @property
    def cost_usd(self) -> float:
        """估算美元成本（基于 claude-opus-4-5 定价）"""
        # Claude Opus: $15/M input, $75/M output
        input_cost = self.input_tokens / 1_000_000 * 15
        output_cost = self.output_tokens / 1_000_000 * 75
        return input_cost + output_cost


class CostMonitor:
    """API 成本监控器"""

    def __init__(
        self,
        daily_budget_usd: float = 50.0,
        alert_threshold: float = 0.8  # 80% 触发告警
    ):
        self.daily_budget = daily_budget_usd
        self.alert_threshold = alert_threshold
        self._records: list[CostRecord] = []

    def record(self, record: CostRecord):
        self._records.append(record)
        self._check_budget()

    def daily_cost(self) -> float:
        today = datetime.utcnow().date()
        return sum(
            r.cost_usd for r in self._records
            if r.timestamp.startswith(str(today))
        )

    def _check_budget(self):
        cost = self.daily_cost()
        ratio = cost / self.daily_budget
        if ratio >= 1.0:
            raise RuntimeError(f"每日预算超限！已消耗 ${cost:.2f} / ${self.daily_budget:.2f}")
        elif ratio >= self.alert_threshold:
            print(f"[成本告警] 已消耗每日预算的 {ratio*100:.1f}%: ${cost:.2f}")

    def cost_report(self) -> dict:
        """生成成本报告"""
        return {
            "daily_cost_usd": self.daily_cost(),
            "daily_budget_usd": self.daily_budget,
            "total_tasks": len(self._records),
            "avg_cost_per_task": self.daily_cost() / max(len(self._records), 1),
            "total_screenshots": sum(r.screenshots_count for r in self._records)
        }

监控指标体系

关键监控指标

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time


# Prometheus 监控指标定义
TASK_COUNTER = Counter(
    "computer_use_tasks_total",
    "Total number of tasks",
    ["status", "task_type"]  # labels
)

TASK_DURATION = Histogram(
    "computer_use_task_duration_seconds",
    "Task execution duration",
    buckets=[10, 30, 60, 120, 300, 600, 1800]
)

TOKEN_COUNTER = Counter(
    "computer_use_tokens_total",
    "Total tokens consumed",
    ["type"]  # input/output
)

SCREENSHOT_COUNTER = Counter(
    "computer_use_screenshots_total",
    "Total screenshots taken"
)

ACTIVE_TASKS = Gauge(
    "computer_use_active_tasks",
    "Currently running tasks"
)

TOOL_CALL_COUNTER = Counter(
    "computer_use_tool_calls_total",
    "Tool call counts",
    ["tool_name", "status"]
)


class MetricsMiddleware:
    """在 Agent 执行循环中收集监控指标"""

    def track_task(self, task_type: str):
        """上下文管理器：追踪单个任务的指标"""
        class TaskTracker:
            def __enter__(self_inner):
                ACTIVE_TASKS.inc()
                self_inner.start_time = time.time()
                return self_inner

            def __exit__(self_inner, exc_type, exc_val, exc_tb):
                ACTIVE_TASKS.dec()
                duration = time.time() - self_inner.start_time
                TASK_DURATION.observe(duration)
                status = "error" if exc_type else "success"
                TASK_COUNTER.labels(status=status, task_type=task_type).inc()

        return TaskTracker()

    def record_api_usage(self, usage):
        """记录 Anthropic API 用量"""
        TOKEN_COUNTER.labels(type="input").inc(usage.input_tokens)
        TOKEN_COUNTER.labels(type="output").inc(usage.output_tokens)

    def record_tool_call(self, tool_name: str, success: bool):
        """记录工具调用"""
        status = "success" if success else "error"
        TOOL_CALL_COUNTER.labels(tool_name=tool_name, status=status).inc()
        if tool_name == "computer":
            SCREENSHOT_COUNTER.inc()


# 暴露 Prometheus 指标端点
start_http_server(8080)

可观测性：完整操作录像

任务执行录像系统

import cv2
import numpy as np
import base64
from datetime import datetime


class TaskRecorder:
    """任务执行录像器：将所有截图帧合并为视频"""

    def __init__(
        self,
        output_dir: str,
        fps: int = 2,      # 2帧/秒，足够回放操作过程
        width: int = 1280,
        height: int = 800
    ):
        self.output_dir = output_dir
        self.fps = fps
        self.width = width
        self.height = height
        self._frames: list[tuple[float, bytes]] = []  # (timestamp, jpeg_bytes)
        self._annotations: list[dict] = []

    def add_frame(self, screenshot_b64: str, annotation: str = None):
        """添加一帧截图"""
        frame_bytes = base64.b64decode(screenshot_b64)
        self._frames.append((time.time(), frame_bytes))
        if annotation:
            self._annotations.append({
                "frame_index": len(self._frames) - 1,
                "text": annotation,
                "timestamp": datetime.utcnow().isoformat()
            })

    def save(self, task_id: str) -> str:
        """保存录像为 MP4 文件"""
        output_path = f"{self.output_dir}/{task_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4"
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (self.width, self.height))

        for i, (ts, frame_bytes) in enumerate(self._frames):
            # 解码 JPEG 为 numpy 数组
            nparr = np.frombuffer(frame_bytes, np.uint8)
            img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
            if img is not None:
                # 调整到目标分辨率
                img = cv2.resize(img, (self.width, self.height))
                # 叠加步骤标注
                annotation = next(
                    (a["text"] for a in self._annotations if a["frame_index"] == i),
                    None
                )
                if annotation:
                    cv2.putText(img, annotation[:80], (10, 30),
                               cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
                out.write(img)

        out.release()
        return output_path

错误告警系统

from typing import Optional
import httpx


class AlertManager:
    """生产告警管理器"""

    def __init__(
        self,
        slack_webhook: Optional[str] = None,
        dingtalk_webhook: Optional[str] = None
    ):
        self.slack_webhook = slack_webhook
        self.dingtalk_webhook = dingtalk_webhook

    async def alert(
        self,
        title: str,
        message: str,
        level: str = "warning",  # info/warning/critical
        task_id: str = None
    ):
        emoji = {"info": "ℹ️", "warning": "⚠️", "critical": "🚨"}.get(level, "ℹ️")
        full_message = f"{emoji} [{level.upper()}] {title}\n{message}"
        if task_id:
            full_message += f"\nTask ID: {task_id}"

        if self.slack_webhook:
            async with httpx.AsyncClient() as client:
                await client.post(
                    self.slack_webhook,
                    json={"text": full_message}
                )

        if self.dingtalk_webhook:
            async with httpx.AsyncClient() as client:
                await client.post(
                    self.dingtalk_webhook,
                    json={
                        "msgtype": "text",
                        "text": {"content": full_message}
                    }
                )


# 告警触发示例
async def monitor_and_alert(metrics, alert_manager: AlertManager):
    # 任务成功率低于 90% 触发告警
    success_rate = metrics.get_success_rate(window_minutes=60)
    if success_rate < 0.9:
        await alert_manager.alert(
            title="任务成功率异常",
            message=f"过去1小时任务成功率 {success_rate*100:.1f}%，低于阈值 90%",
            level="critical"
        )

    # 平均任务时长超过 10 分钟触发告警
    avg_duration = metrics.get_avg_duration(window_minutes=60)
    if avg_duration > 600:
        await alert_manager.alert(
            title="任务执行过慢",
            message=f"平均任务时长 {avg_duration:.0f}s，超过预期 600s",
            level="warning"
        )

商业化 Computer Use 服务

主流云端方案对比

生产部署核心原理深讲

容器化的本质：环境一致性与隔离

为什么生产环境必须用 Docker 容器而不是直接在宿主机运行？理解这一点需要理解 Computer Use Agent 的特殊性：

Kubernetes HPA 的伸缩原理

HPA（Horizontal Pod Autoscaler）基于外部指标（如任务队列长度）自动调整 Pod 数量，其核心控制逻辑如下：

目标副本数 = ceil(当前副本数 × (当前指标值 / 目标指标值))

示例：
  当前副本数 = 3
  当前队列长度 = 30
  目标：每个 Pod 处理 5 个任务
  目标副本数 = ceil(3 × (30/5)) = ceil(18) = 18 → 扩容到 18 个 Pod

当队列消耗完：
  当前队列长度 = 2
  目标副本数 = ceil(18 × (2/5)) = ceil(7.2) = 8 → 缩容到 8 个 Pod
  （缩容有 stabilizationWindow 防止抖动，默认 5 分钟）

Prometheus 监控的可观测性三支柱

成本控制的数学模型

理解 Computer Use 的成本构成，才能做出有效的优化决策：

单次任务成本 = Σ(每轮 API 调用成本)

每轮 API 调用成本 =
  input_tokens × $15/M  +  output_tokens × $75/M

input_tokens = 系统提示词 + 消息历史 + 所有截图 token
            = ~500  +  累积消息  +  每张截图600-1500 tokens

典型 10 步任务估算：
  系统提示词：500 tokens
  每步消息历史（累积）：平均 2000 tokens  → 10步共 20,000 tokens
  每步2张截图 × 1000 tokens：20,000 tokens  → 10步共 200,000 tokens
  每步输出：平均 200 tokens  → 10步共 2,000 tokens

  总 input：≈ 220,500 tokens  → $3.31
  总 output：≈ 2,000 tokens   → $0.15
  总成本：≈ $3.46 / 任务

优化后（截图压缩 + 消息历史管理）：
  截图 token 降至 600/张  → 120,000 tokens
  消息历史压缩 50%  → 10,000 tokens
  总 input：≈ 130,500 tokens  → $1.96
  节省 ≈ $1.50 / 任务（43% 降低）

常见生产问题与排查