如何训练一个原生 Computer Use 模型

import json
from pathlib import Path
from dataclasses import dataclass, field

@dataclass
class DataQualityFilter:
    min_resolution: tuple = (1024, 768)
    max_elements: int = 200
    coordinate_range: tuple = (0, 1)
    required_fields: list = field(default_factory=lambda: [
        "screenshot", "instruction", "action", "coordinates"
    ])
    
    def validate_sample(self, sample: dict) -> tuple[bool, str]:
        for field_name in self.required_fields:
            if field_name not in sample:
                return False, f"missing field: {field_name}"
        
        x, y = sample["coordinates"]
        if not (self.coordinate_range[0] <= x <= self.coordinate_range[1]):
            return False, f"x coordinate {x} out of range"
        if not (self.coordinate_range[0] <= y <= self.coordinate_range[1]):
            return False, f"y coordinate {y} out of range"
        
        if sample["action"] not in ["click", "type", "scroll", "key", "drag", "wait", "finish"]:
            return False, f"unknown action: {sample['action']}"
        
        return True, "ok"
    
    def filter_dataset(self, samples: list[dict]) -> list[dict]:
        valid = []
        for i, sample in enumerate(samples):
            ok, reason = self.validate_sample(sample)
            if ok:
                valid.append(sample)
            else:
                print(f"sample {i} rejected: {reason}")
        return valid

data_filter = DataQualityFilter()
raw_dataset = json.loads(Path("raw_samples.json").read_text())
clean_dataset = data_filter.filter_dataset(raw_dataset)
print(f"filtered {len(raw_dataset)} -> {len(clean_dataset)} samples")

{
  "conversations": [
    {
      "role": "user",
      "content": [
        {"type": "image", "image": "screenshot_001.png"},
        {"type": "text", "text": "在当前页面中，点击'设置'按钮进入设置页面"}
      ]
    },
    {
      "role": "assistant", 
      "content": "```json\n{\"action\": \"click\", \"coordinates\": [0.82, 0.15], \"reasoning\": \"设置按钮位于右上角，图标为齿轮形状\"}\n```"
    }
  ]
}

from transformers import TrainingArguments

training_config = TrainingArguments(
    output_dir="./model_checkpoints/sft",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,
    learning_rate=2e-5,
    weight_decay=0.01,
    warmup_ratio=0.03,
    lr_scheduler_type="cosine",
    bf16=True,
    gradient_checkpointing=True,
    dataloader_num_workers=8,
    save_strategy="steps",
    save_steps=500,
    logging_steps=10,
    max_grad_norm=1.0,
    report_to="wandb",
)

lora_config = {
    "r": 128,
    "lora_alpha": 256,
    "lora_dropout": 0.05,
    "target_modules": ["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
    "task_type": "CAUSAL_LM",
}

import torch
import math

def compute_gui_grounding_reward(
    predicted_coords: tuple[float, float],
    ground_truth_coords: tuple[float, float],
    bbox: tuple[float, float, float, float] | None = None,
    action_type: str = "click"
) -> dict:
    pred_x, pred_y = predicted_coords
    gt_x, gt_y = ground_truth_coords
    
    distance = math.sqrt((pred_x - gt_x) ** 2 + (pred_y - gt_y) ** 2)
    
    position_reward = max(0, 1 - distance / 0.1)
    
    if bbox:
        bx, by, bw, bh = bbox
        inside = (bx <= pred_x <= bx + bw) and (by <= pred_y <= by + bh)
        containment_bonus = 1.0 if inside else -0.3
    else:
        containment_bonus = 0.0
    
    exact_match = 1.0 if distance < 0.005 else 0.0
    
    total_reward = (
        0.6 * position_reward + 
        0.3 * containment_bonus + 
        0.1 * exact_match
    )
    
    return {
        "total_reward": total_reward,
        "position_reward": position_reward,
        "containment_bonus": containment_bonus,
        "exact_match": exact_match,
        "distance": distance,
    }

def grpo_group_loss(
    rewards: list[float],
    log_probs: list[float],
    ref_log_probs: list[float],
    beta: float = 0.04,
    clip_range: float = 0.2,
) -> torch.Tensor:
    mean_r = sum(rewards) / len(rewards)
    std_r = max(sum((r - mean_r) ** 2 for r in rewards) / len(rewards), 1e-8) ** 0.5
    
    advantages = [(r - mean_r) / (std_r + 1e-8) for r in rewards]
    
    loss = torch.tensor(0.0)
    for adv, lp, ref_lp in zip(advantages, log_probs, ref_log_probs):
        ratio = torch.exp(lp - ref_lp)
        clipped = torch.clamp(ratio, 1 - clip_range, 1 + clip_range)
        loss = loss - torch.min(ratio * adv, clipped * adv)
    
    kl_penalty = beta * sum(lp - ref_lp for lp, ref_lp in zip(log_probs, ref_log_probs)) / len(log_probs)
    loss = loss / len(rewards) + kl_penalty
    
    return loss

SENSITIVE_ACTIONS = {
    "send_email": {"risk": "high", "confirm": True},
    "delete_file": {"risk": "high", "confirm": True},
    "financial_transaction": {"risk": "critical", "confirm": True},
    "install_software": {"risk": "medium", "confirm": True},
    "open_website": {"risk": "low", "confirm": False},
    "scroll_page": {"risk": "low", "confirm": False},
    "type_text": {"risk": "low", "confirm": False},
}

def should_confirm_action(action: dict) -> bool:
    action_type = action.get("action", "unknown")
    risk_config = SENSITIVE_ACTIONS.get(action_type, {"risk": "high", "confirm": True})
    return risk_config["confirm"]

def execute_with_safety(action: dict, env) -> dict:
    if should_confirm_action(action):
        confirmation = request_human_confirmation(
            action=action,
            description=f"模型请求执行: {action['action']} - {action.get('reasoning', '')}"
        )
        if not confirmation:
            return {"status": "rejected", "reason": "human_rejected"}
    
    return env.execute(action)

┌─────────────────────────────────────────────────────────────────────┐
│                    Computer Use 模型训练流程                          │
│                                                                      │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │  Phase 0: 数据准备                                             │  │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐  │  │
│  │  │ 公开数据集  │  │ 合成数据   │  │ 人工标注   │  │ OmniParser   │  │  │
│  │  │ 收集      │  │ 生成      │  │ 采购      │  │ 质量过滤     │  │  │
│  │  └─────┬────┘  └─────┬────┘  └─────┬────┘  └──────┬───────┘  │  │
│  │        └──────────────┴─────────────┴───────────────┘          │  │
│  │                            ▼                                    │  │
│  │                    数据清洗 + 难度分级                            │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                            ▼                                         │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │  Phase 1: SFT (监督微调)                                       │  │
│  │  基座: Qwen2.5-VL-7B → LoRA r=128 → 3 epochs                 │  │
│  │  数据: ~100K-500K 高质量指令-操作对                               │  │
│  │  目标: 基础感知 + 坐标预测 + 动作词汇                              │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                            ▼                                         │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │  Phase 2: RL (强化学习)                                         │  │
│  │  算法: GRPO → 交互环境: OSWorld / WebArena / 自建               │  │
│  │  奖励: 位置奖励 + 任务完成奖励 + 轨迹奖励                          │  │
│  │  目标: 错误恢复 + 多步推理 + 泛化能力                              │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                            ▼                                         │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │  Phase 3: 安全对齐                                              │  │
│  │  拒绝训练 + Human-in-the-Loop + 沙箱验证                         │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                            ▼                                         │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │  Phase 4: 部署                                                  │  │
│  │  vLLM/TGI 推理服务 + 截图获取 + 动作执行 + 环境交互              │  │
│  └───────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────┘