按 Mage-VL-Cowork 配方训练混合电脑操作模型

这套流程做完，你手里会有一条完整的链路：一个会自己加难度、自己判分的 Harness，一批"截图 + 可交互元素树"双通道的操作轨迹，一个能同时操作浏览器页面和原生桌面窗口的微调模型，以及一份在真实任务上跑出来的评测报告。

具体交付物大致是四样：

• harness/：任务生成器 + 执行器 + 验证器 + 轨迹落盘，能无人值守地连续采集；
• data/sft/*.jsonl：归一化后的混合操作轨迹，每条样本包含一张截图、精简后的元素树、一条指令和一个动作；
• 一个 LoRA 适配器（或全参微调权重），推理时输入当前屏幕，输出"点哪里、输入什么、何时结束"；
• eval/report.md：在留出任务集上的成功率、平均步数、无效动作率。

关键设计只有一句话：同一个策略，在有 DOM 时用元素树辅助定位，在没有 DOM 的原生桌面里只靠像素也能干活。做法是在训练时随机丢掉元素树通道，让模型学出"两条腿走路"。

前置条件清单

先把环境和边界立好，再谈采数据。

• 算力：一台带 GPU 的机器。7B 级别的视觉语言模型做 LoRA，按经验需要 24GB 以上显存，具体取决于分辨率、batch size 和是否开启AI 词典：梯度检查点">梯度检查点，以官方文档当前版本为准。
• 隔离的操作环境：一台能一键快照回滚的虚拟机（推荐），或在容器里用 Xvfb 起一个虚拟显示器。不要用你的工作机采轨迹，Harness 会乱点。
• 账号与安全：所有站点用本地 mock 页或测试账号；禁止让 Agent 触碰支付、删除、发消息这类不可逆操作；给执行器加动作限速（比如每次点击后强制等待若干毫秒）。
• 软件依赖（版本以官方文档当前版本为准）：

```bash

python -m venv .venv && source .venv/bin/activate

pip install playwright pillow pyautogui

pip install "transformers" "peft" "trl" "datasets" "accelerate"

python -m playwright install chromium

```

• 基础模型：任选一个支持图像输入的视觉语言模型（Qwen-VL 系列、InternVL 系列等均可，具体版本与显存需求以官方文档当前版本为准）。
• 种子任务：50～200 条手写的自然语言任务，例如"在本地商城页把购物车里数量为 2 的商品改成 3 并进入结算页"。

第 1 步：搭出 Harness 的目录骨架

先把结构定死，后面每一步都往里填东西。

```text

cowork/

├── configs/

│ ├── tasks_seed.yaml # 手写种子任务

│ └── train.yaml # 训练超参

├── harness/

│ ├── env_browser.py # 浏览器环境

│ ├── env_desktop.py # 原生桌面环境

│ ├── observe.py # 双通道观测

│ ├── actions.py # 动作执行

│ ├── verify.py # 结果验证

│ ├── evolve.py # 任务自演化

│ └── rollout.py # 单条轨迹采集

├── data/

│ ├── raw/ # 原始轨迹

│ ├── sft/ # 训练数据

│ └── eval/ # 留出评测集

├── train/sft_lora.py

└── eval/run_eval.py

```

任务配置用一个最小结构，重点是"任务 + 判分方式"必须成对出现：

```yaml

configs/tasks_seed.yaml

• id: cart_qty_001

env: browser

start_url: "http://localhost:8000/cart"

instruction: "把购物车里数量为 2 的商品数量改成 3"

checks:

• kind: dom_text

selector: "#item-1 .qty"

expect: "3"

max_steps: 15

• id: desktop_rename_001

env: desktop

instruction: "把桌面上的 report.txt 重命名为 report_final.txt"

checks:

• kind: file_exists

path: "~/Desktop/report_final.txt"

max_steps: 20

```

判分只认"最终状态"，不认"过程像不像人"。这是后面能自动化筛选数据的前提。

第 2 步：把观测做成"截图 + 元素树"双通道

浏览器侧用 Playwright 同时拿两样东西：一张 PNG，和一份精简过的可交互元素列表。

```python

harness/observe.py

import base64

from playwright.sync_api import sync_playwright

COLLECT_JS = """

() => {

const sel = 'a,button,input,select,textarea,[role=button],[role=link],[contenteditable=true]';

const out = [];

document.querySelectorAll(sel).forEach((el, i) => {

const r = el.getBoundingClientRect();

if (r.width < 2 || r.height < 2) return;

out.push({

idx: i,

tag: el.tagName.toLowerCase(),

name: (el.innerText || el.value || el.placeholder || el.getAttribute('aria-label') || '').trim().slice(0, 40),

box: [Math.round(r.x), Math.round(r.y), Math.round(r.width), Math.round(r.height)]

});

});

return out.slice(0, 120); // 只保留前若干个可交互节点，防止 prompt 爆炸

}

"""

class BrowserEnv:

def __init__(self, headless=True):

self._pw = sync_playwright().start()

self.browser = self._pw.chromium.launch(headless=headless)

self.ctx = self.browser.new_context(viewport={"width": 1440, "height": 900})

self.page = self.ctx.new_page()

def observe(self):

shot = self.page.screenshot(type="png")

nodes = self.page.evaluate(COLLECT_JS)

return {

"screenshot_b64": base64.b64encode(shot).decode(),

"nodes": nodes,

"url": self.page.url,

"viewport": [1440, 900],

}

def close(self):

self.browser.close()

self._pw.stop()

```

原生桌面侧没有 DOM，退化成"只有截图"，元素树字段留空即可：

```python

harness/env_desktop.py

import pyautogui

class DesktopEnv:

def observe(self):

shot = pyautogui.screenshot()

w, h = shot.size

return {

"screenshot_b64": shot_to_b64(shot),

"nodes": [], # 桌面无元素树，训练时走纯像素分支

"url": None,

"viewport": [w, h],

}

```

如果想把桌面也做成双通道，可以接各平台的辅助功能接口（Windows 的 UI Automation、macOS 的 AX、Linux 的 AT-SPI），封装成和浏览器一致的 nodes 结构，上层代码不用改。具体 API 以各项目官方文档为准。

第 3 步：定义统一动作空间

动作空间越小，模型越好学。起步就用这六种：

```python

harness/actions.py

import time

def execute(env, action):

t = action["type"]

try:

if t == "click":

if action.get("node_idx") is not None and env.observe()["nodes"]:

box = env.observe()["nodes"][action["node_idx"]]["box"]

pt = (box[0] + box[2] / 2, box[1] + box[3] / 2)

else:

pt = tuple(action["point"])

if hasattr(env, "page"):

env.page.mouse.click(*pt)

else:

import pyautogui; pyautogui.click(*pt)

elif t == "type":

if hasattr(env, "page"):

env.page.keyboard.type(action["text"], delay=20)

else:

import pyautogui; pyautogui.typewrite(action["text"], interval=0.02)

elif t == "key":

if hasattr(env, "page"):

env.page.keyboard.press(action["combo"])

else:

import pyautogui; pyautogui.hotkey(*action["combo"].split("+"))

elif t == "scroll":

env.page.mouse.wheel(0, action["dy"])

elif t == "wait":

time.sleep(action["ms"] / 1000)

return {"ok": True}

except Exception as e:

return {"ok": False, "error": str(e)}

finally:

time.sleep(0.3) # 强制限速，给页面和系统留反应时间

```

第 4 步：让 Harness 自演化——生成、执行、验证、回流

"自演化"不是一个玄学词，它就是四件事循环：改任务 → 跑一遍 → 用最终状态判分 → 把成功的轨迹和难度合适的任务留下来。

任务变异器优先做低风险改写：

```python

harness/evolve.py

import copy, random

def mutate(task, k=2):

out = []

for _ in range(k):

t = copy.deepcopy(task)

r = random.random()

if r < 0.4 and "instruction" in t:

t["instruction"] = t["instruction"].replace("3", str(random.choice([4, 5])))

for c in t["checks"]:

if c["kind"] == "dom_text":

c["expect"] = t["instruction"].split("改成")[-1].strip() or c["expect"]

elif r < 0.7:

t["max_steps"] = max(5, t["max_steps"] - 2) # 收紧预算，逼模型少走弯路

else:

t["instruction"] = "先刷新页面，再" + t["instruction"]

t["id"] = f'{task["id"]}_m{random.randint(1000, 9999)}'

out.append(t)

return out

```

主循环按"成功率落在 0.2～0.8 之间"来挑选任务继续变异，太简单的丢掉，太难的回滚：

```python

harness/rollout.py

from collections import defaultdict

def collect(policy, env_factory, seeds, rounds=3, per_round=50):

store = []

stats = defaultdict(lambda: {"ok": 0, "n": 0})

frontier = list(seeds)

for r in range(rounds):

batch = []

for t in frontier:

batch.extend(mutate(t, k=2))

random.shuffle(batch)

batch = batch[:per_round]

for task in batch:

env = env_factory(task["env"])

traj = run_one(policy, env, task) # 逐步记录 obs / action / result

passed = verify(env, task) # 只认最终状态

stats[task["id"]]["n"] += 1

stats[task["id"]]["ok"] += int(passed)

if passed:

store.append(traj) # 成功轨迹入库

task["_passed"] = passed

env.close()

难度筛选：成功率在中间的留作下一轮父本

frontier = [t for t in batch

if 0.2 <= stats[t["id"]]["ok"] / max(1, stats[t["id"]]["n"]) <= 0.8]

if not frontier:

frontier = random.sample(batch, min(5, len(batch)))

return store

```

验证器按 checks 类型分发，写完就复用到评测阶段：

```python

harness/verify.py

import os

def verify(env, task):

for c in task["checks"]:

if c["kind"] == "dom_text":

got = env.page.locator(c["selector"]).inner_text().strip()

if got != c["expect"]:

return False

elif c["kind"] == "file_exists":

if not os.path.exists(os.path.expanduser(c["path"])):

return False

return True

```

一个容易被忽略的点：失败轨迹别扔。把"卡住—尝试别的路径—最终完成"的片段单独抽出来，后面做恢复能力训练很值钱。

第 5 步：把轨迹整理成训练数据

三条清洗规则，缺一条模型就会学歪：

1. 坐标归一化到 0～1000，推理时再按当前屏幕尺寸乘回去，避免换分辨率就废；

2. 元素树做 DOM dropout，每条样本以一定概率清空 nodes，逼模型学会纯像素操作；

3. 动作类型做平衡，click 一般占绝大多数，对 type、done 做过采样。

```python

harness/build_sft.py

import json, random

def to_norm(pt, viewport):

w, h = viewport

return [round(pt[0] / w * 1000), round(pt[1] / h * 1000)]

def build(traj, drop_dom_p=0.3):

samples = []

for step in traj["steps"]:

obs, act = step["obs"], step["action"]

if act["type"] == "click" and act.get("point"):

act = dict(act, point=to_norm(act["point"], obs["viewport"]))

nodes = [] if random.random() < drop_dom_p else obs["nodes"]

prompt = (

f"任务：{traj['instruction']}\n"

f"当前网址：{obs['url']}\n"

f"可交互元素：{json.dumps(nodes, ensure_ascii=False)}\n"

"输出下一步动作，JSON 格式，不要解释。"

)

samples.append({

"messages": [

{"role": "user", "content": [

{"type": "image", "image": f"shots/{step['shot_id']}.png"},

{"type": "text", "text": prompt},

]},

{"role": "assistant", "content": [

{"type": "text", "text": json.dumps(act, ensure_ascii=False)}

]},

],

"meta": {"task_id": traj["task_id"], "env": traj["env"], "has_dom": bool(nodes)},

})

return samples

```

最终每条样本长这样：

```json

{

"messages": [

{"role": "user", "content": [

{"type": "image", "image": "shots/0003.png"},

{"type": "text", "text": "任务：把购物车里数量为 2 的商品数量改成 3\n可交互元素：[{\"idx\":3,\"tag\":\"input\",\"name\":\"qty\",\"box\":[820,410,60,28]}]"}

]},

{"role": "assistant", "content": [

{"type": "text", "text": "{\"type\": \"click\", \"node_idx\": 3, \"point\": [569, 455]}"}

]}

]

}

```

配比建议：混合操作轨迹占大头，掺一部分纯文本的界面问答和通用视觉问答，防止模型在微调后忘记基础能力。

第 6 步：微调

用 LoRA 起步，先把链路跑通再谈效果。

```python

train/sft_lora.py

from datasets import load_dataset

from peft import LoraConfig

from transformers import AutoProcessor, AutoModelForVision2Seq

from trl import SFTConfig, SFTTrainer

model_id = "你的视觉语言模型路径或仓库名" # 以官方文档当前版本为准

processor = AutoProcessor.from_pretrained(model_id)

model = AutoModelForVision2Seq.from_pretrained(model_id, torch_dtype="bfloat16")

peft_cfg = LoraConfig(

r=16, lora_alpha=32, lora_dropout=0.05,

target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], # 按模型结构改

task_type="CAUSAL_LM",

)

args = SFTConfig(

output_dir="out/cowork-lora",

num_train_epochs=2,

per_device_train_batch_size=1,

gradient_accumulation_steps=8,

learning_rate=1e-4,

logging_steps=10,

save_strategy="epoch",

gradient_checkpointing=True,

)

trainer = SFTTrainer(

model=model, args=args,

peft_config=peft_cfg,

train_dataset=load_dataset("json", data_files="data/sft/train.jsonl", split="train"),

processing_class=processor,

)

trainer.train()

trainer.save_model("out/cowork-lora")

```

几个实操要点：

• 截图统一缩放到固定尺寸（例如 1280×800，按模型支持的输入分辨率调整），别让每张图尺寸乱跳；
• 学习率从 1e-4 量级起步，loss 不降就往下调；
• 第一个 epoch 结束就手动跑几条推理看看，别等训完才发现输出格式全错。

第 7 步：在真实桌面任务上评测

评测集必须和训练集不同源：不同的站点、不同的软件、不同的指令措辞。否则分数好看但没用。

指标就四个，够说明问题：

• 成功率：最终状态验证通过的比例；
• 平均步数：越少说明决策越干脆；
• 无效动作率：点了没反应的点击、报错的输入；
• 恢复率：第一次出错后还能完成的比例，最能体现模型是不是真的"会用电脑"。

```python

eval/run_eval.py

import json

def run_suite(policy, env_factory, tasks, repeat=3):

report = []

for task in tasks:

runs = []

for _ in range(repeat):

env = env_factory(task["env"])

traj = run_one(policy, env, task)

runs.append({

"success": verify(env, task),

"steps": len(traj["steps"]),

"invalid": sum(1 for s in traj["steps"] if not s["result"]["ok"]),

})

env.close()

report.append({

"task_id": task["id"],

"success_rate": sum(r["success"] for r in runs) / repeat,

"avg_steps": sum(r["steps"] for r in runs) / repeat,

"invalid_rate": sum(r["invalid"] for r in runs) / (sum(r["steps"] for r in runs) or 1),

})

print(json.dumps(report, ensure_ascii=False, indent=2))

return report

```

跑三次取均值，把每条任务的成功率列成表，重点看成功率为 0 的那几条——它们通常不是模型不行，而是任务设计有歧义，或者环境里有隐藏的等待时机问题。

常见坑与排错

坐标整体偏移。 最常见。原因通常是截图分辨率和实际屏幕分辨率不一致（缩放比例、高分屏、窗口边框）。排查办法：在截图中心画一个红点，用同一坐标去点，看落点偏多少。训练和推理必须用同一套归一化逻辑。

元素树和可点区域对不上。 弹窗遮罩、iframe、Shadow DOM 都会让 box 坐标失真。解决思路是采集时过滤掉 pointer-events: none 和被遮挡的元素；跨 iframe 的任务要么跳过，要么只在顶层页面操作。

动作循环。 模型反复点同一个位置。在推理侧加个简单保护：连续三次相同动作就强制 wait 或换提示。

时序问题。 页面还在加载就截图，模型看到的是半截界面。每个动作后固定等待 + 关键操作后等待某个元素出现，别用固定 sleep 硬扛。

数据泄漏。 训练集和评测集用了同一个站点同一套 DOM 结构，分数虚高。按站点/软件划分，而不是随机划分。

全量 DOM 塞爆上下文。 只保留可交互节点、限制数量、文本截断到几十个字。元素树是辅助，不是主体。

显存不够。 降 batch size、开梯度检查点、降截图分辨率，三招按顺序试。

截图里有敏感信息。 采集环境用假数据，落盘前做一次人工抽检。

下一步建议

链路跑通之后，按投入产出比排序可以往下走：

1. 加反思数据：把失败轨迹配上"错在哪、应该怎么做"的一段说明，和正常样本混着训，恢复率通常涨得比成功率快。

2. 扩任务域：从浏览器扩到文件管理器、表格软件、终端窗口，每个新域先手工写 20 条种子任务，再让变异器去长。

3. 偏好优化：同一状态下采两条轨迹，成功的那条做正例，做一次 DPO，比继续堆 SFT 数据更省事。

4. 在环自演化：把训好的模型放回 Harness 当执行策略，成功率落在中间带的任务自动进下一轮训练集，形成闭环。

5. 工具动作：让模型在纯点击之外学会"直接调脚本"，比如遇到批量重命名时输出一段 shell 而不是点二十次，这类混合动作空间是下一步的分水岭。

最后提醒一句：这套东西的价值不在模型本身，而在那条能不断自我加难、自我判分的数据流水线。模型换个更强的基座就能重训，流水线才是资产。