Qwen-UI-Agent深度解析：阿里开源真机GUI智能体，AI终于长出了能操作手机电脑的手

# Batched Actions 示例：模型单次推理输出的动作序列
actions = [
    {"type": "gui_click", "x": 450, "y": 320, "desc": "打开文件管理器"},
    {"type": "gui_type", "text": "/home/user/data", "desc": "输入路径"},
    {"type": "gui_key_combo", "keys": ["Ctrl", "Enter"], "desc": "确认进入"},
    {"type": "cli_bash", "command": "ls -la *.csv | wc -l", "desc": "统计CSV文件数"},
    {"type": "cli_bash", "command": "head -n 5 report.csv", "desc": "预览数据前5行"},
]
# 以上6个动作通过一次模型推理输出，无需逐步等待

def sliding_window_training(trajectory, window_size=4096, stride=2048):
    """
    滑动窗口训练：将长轨迹切分为多个重叠窗口
    
    Args:
        trajectory: 完整轨迹，包含 (screen_obs, action, reward) 序列
        window_size: 窗口大小（token数）
        stride: 滑动步长
    """
    windows = []
    tokens = tokenize_trajectory(trajectory)
    
    # 将长轨迹切分为重叠窗口
    for start in range(0, len(tokens) - window_size + 1, stride):
        window = tokens[start:start + window_size]
        windows.append(window)
    
    # 对每个窗口独立计算损失
    for window in windows:
        # 只对窗口内后半部分（动作token）计算损失
        action_mask = get_action_mask(window)
        loss = cross_entropy_loss(
            model_output=model(window),
            targets=window,
            reduction_mask=action_mask
        )
        loss.backward()
    
    # 梯度累积后更新参数
    optimizer.step()
    optimizer.zero_grad()

class OnlineRLTrainer:
    """
    万级并发的在线强化学习训练器
    """
    def __init__(self, num_envs=10000, max_trajectory_length=150):
        self.num_envs = num_envs
        self.max_trajectory_length = max_trajectory_length
        self.environments = self._init_environments(num_envs)
        self.curriculum = AdaptiveCurriculum()
    
    def train_step(self, model, num_iterations=1000):
        for iteration in range(num_iterations):
            # 1. 从课程学习中采样任务
            tasks = self.curriculum.sample_tasks(batch_size=self.num_envs)
            
            # 2. 并行执行rollout
            rollouts = []
            for env, task in zip(self.environments, tasks):
                trajectory = env.run_episode(
                    model=model,
                    task=task,
                    max_steps=self.max_trajectory_length
                )
                rollouts.append(trajectory)
            
            # 3. 验证器自动评分
            rewards = []
            for trajectory in rollouts:
                verifier_score = self.auto_verifier(trajectory)
                rewards.append(verifier_score)
            
            # 4. 模型自适应课程学习
            # 成功率适中的任务获得更高采样权重
            # 已掌握的任务被替换为更难的任务
            self.curriculum.update(rollouts, rewards)
            
            # 5. GRPO策略优化
            loss = compute_grpo_loss(model, rollouts, rewards)
            loss.backward()
            optimizer.step()
        
        return model

import torch
from transformers import AutoModelForCausalLM, AutoProcessor

# 加载模型（以7B版本为例）
model_path = "Qwen/Qwen-UI-Agent-7B"
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True
)
processor = AutoProcessor.from_pretrained(model_path, trust_remote_code=True)

def run_gui_agent(task_prompt: str, screenshot_path: str, max_steps: int = 50):
    """
    运行GUI智能体执行任务
    
    Args:
        task_prompt: 用户任务描述，如"帮我查一下北京到杭州最早的高铁"
        screenshot_path: 当前屏幕截图路径
        max_steps: 最大执行步数
    """
    # 构建模型输入
    messages = [
        {
            "role": "user",
            "content": [
                {"type": "image", "image": screenshot_path},
                {"type": "text", "text": task_prompt}
            ]
        }
    ]
    
    # 转换为模型输入格式
    inputs = processor.apply_chat_template(
        messages, 
        return_tensors="pt",
        add_generation_prompt=True
    ).to(model.device)
    
    # 生成动作
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1024,
            temperature=0.1,
            do_sample=True,
            top_p=0.9
        )
    
    # 解析动作
    response = processor.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
    actions = parse_actions(response)
    
    return actions

def parse_actions(model_output: str) -> list:
    """
    解析模型输出的动作序列
    """
    actions = []
    for line in model_output.strip().split("\n"):
        line = line.strip()
        if not line:
            continue
        
        # 支持的动作格式：GUI操作、CLI命令、Batched Actions
        if line.startswith("gui_click"):
            # gui_click(x=450, y=320, desc="打开文件管理器")
            import re
            match = re.match(r"gui_click\(x=(\d+),\s*y=(\d+)(?:,\s*desc=\"([^\"]*)\")?\)", line)
            if match:
                actions.append({
                    "type": "gui_click",
                    "x": int(match.group(1)),
                    "y": int(match.group(2)),
                    "desc": match.group(3) or ""
                })
        elif line.startswith("gui_type"):
            # gui_type(text="hello", desc="输入文本")
            match = re.match(r"gui_type\(text=\"([^\"]*)\"(?:,\s*desc=\"([^\"]*)\")?\)", line)
            if match:
                actions.append({
                    "type": "gui_type",
                    "text": match.group(1),
                    "desc": match.group(2) or ""
                })
        elif line.startswith("cli_bash"):
            # cli_bash(command="ls -la", desc="列出文件")
            match = re.match(r"cli_bash\(command=\"([^\"]*)\"(?:,\s*desc=\"([^\"]*)\")?\)", line)
            if match:
                actions.append({
                    "type": "cli_bash",
                    "command": match.group(1),
                    "desc": match.group(2) or ""
                })
        elif line.startswith("batch_execute"):
            # batch_execute([action1, action2, ...])
            # 批量执行多个动作
            pass
    
    return actions

class GUIExecutor:
    """
    GUI操作执行器：将模型输出的动作指令转换为实际设备操作
    """
    
    def __init__(self, device_type="android"):
        self.device_type = device_type
        if device_type == "android":
            import uiautomator2 as u2
            self.device = u2.connect()  # 连接真机或模拟器
        elif device_type == "desktop":
            import pyautogui
            self.pyautogui = pyautogui
        
    def execute(self, action: dict):
        """执行单个动作"""
        action_type = action["type"]
        
        if action_type == "gui_click":
            self._click(action["x"], action["y"])
        elif action_type == "gui_type":
            self._type(action["text"])
        elif action_type == "gui_scroll":
            self._scroll(action["dx"], action["dy"])
        elif action_type == "gui_key_combo":
            self._key_combo(action["keys"])
        elif action_type == "cli_bash":
            self._exec_bash(action["command"])
        elif action_type == "batch_execute":
            # 批量执行
            for sub_action in action["actions"]:
                self.execute(sub_action)
    
    def _click(self, x: int, y: int):
        if self.device_type == "android":
            self.device.click(x, y)
        else:
            self.pyautogui.click(x, y)
    
    def _type(self, text: str):
        if self.device_type == "android":
            self.device.send_keys(text)
        else:
            self.pyautogui.typewrite(text)
    
    def _exec_bash(self, command: str):
        import subprocess
        result = subprocess.run(
            command, shell=True, capture_output=True, text=True, timeout=30
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode
        }
    
    def _key_combo(self, keys: list):
        import pyautogui
        # 支持组合键，如 Ctrl+C, Alt+Tab
        pyautogui.hotkey(*keys)
    
    def execute_batch(self, actions: list):
        """执行批量动作序列"""
        results = []
        for action in actions:
            try:
                result = self.execute(action)
                results.append({"action": action, "status": "success", "result": result})
            except Exception as e:
                results.append({"action": action, "status": "failed", "error": str(e)})
                # 批量执行中某个动作失败，根据策略决定是否继续
                if self._recoverable(e):
                    continue
                else:
                    break
        return results

class RealDeviceTestFramework:
    """
    真机测试框架：在真实设备上评测GUI智能体
    """
    
    def __init__(self, devices: list, apps: list):
        self.devices = devices  # 设备列表
        self.apps = apps        # 应用列表
        self.auto_judge = AutoJudge()
    
    def run_evaluation(self, model, tasks: list, num_episodes=3):
        """
        在真机上运行评测
        """
        results = []
        
        for task in tasks:
            task_results = []
            for episode in range(num_episodes):
                # 重置设备到初始状态
                device = self._reset_device(task.required_apps)
                
                # 运行智能体
                trajectory = []
                screenshot = device.screenshot()
                done = False
                step = 0
                
                while not done and step < task.max_steps:
                    # 调用模型生成动作
                    actions = run_gui_agent(task.instruction, screenshot)
                    
                    # 执行动作
                    for action in actions:
                        step += 1
                        trajectory.append({
                            "step": step,
                            "screenshot": screenshot,
                            "action": action,
                            "timestamp": time.time()
                        })
                        
                        try:
                            observation = device.execute(action)
                        except Exception as e:
                            observation = {"error": str(e)}
                        
                        # 检查任务是否完成
                        done = self._check_completion(trajectory, task)
                        if done:
                            break
                    
                    screenshot = device.screenshot()
                
                # 自动评分
                judge_result = self.auto_judge.evaluate(trajectory, task)
                task_results.append({
                    "episode": episode,
                    "success": judge_result["success"],
                    "steps": step,
                    "trajectory": trajectory,
                    "judge_detail": judge_result
                })
            
            results.append({
                "task": task,
                "episodes": task_results,
                "avg_success": sum(r["success"] for r in task_results) / num_episodes
            })
        
        return results
    
    def _reset_device(self, required_apps):
        """重置设备到基准状态"""
        device = random.choice(self.devices)
        device.reset_to_home()
        for app in required_apps:
            device.launch_app(app)
            device.kill_app(app)  # 清除缓存状态
        return device

class SafetyController:
    """
    GUI智能体安全控制器
    """
    
    # 敏感操作分类
    SENSITIVE_OPERATIONS = {
        "payment": {
            "keywords": ["支付", "付款", "转账", "充值", "下单", "购买"],
            "action": "pause_and_confirm"
        },
        "data_deletion": {
            "keywords": ["删除", "移除", "清空", "卸载", "格式化"],
            "action": "pause_and_confirm"
        },
        "authorization": {
            "keywords": ["授权", "登录", "权限", "同意", "允许"],
            "action": "pause_and_confirm"
        },
        "illegal": {
            "keywords": ["破解", "入侵", "盗取", "诈骗", "赌博"],
            "action": "reject_immediately"
        }
    }
    
    def __init__(self):
        self.sensitive_detector = self._build_detector()
    
    def _build_detector(self):
        """构建敏感操作检测器"""
        from transformers import pipeline
        classifier = pipeline(
            "text-classification",
            model="Qwen/Qwen-UI-Agent-Safety",
            device="cuda"
        )
        return classifier
    
    def check_action(self, action: dict, context: dict) -> dict:
        """
        检查动作的安全性
        
        Returns:
            {
                "safe": bool,
                "level": "safe" | "warning" | "dangerous" | "illegal",
                "action": "proceed" | "pause" | "reject",
                "message": str
            }
        """
        # 1. 关键词快速检测
        action_text = self._action_to_text(action)
        for category, config in self.SENSITIVE_OPERATIONS.items():
            for keyword in config["keywords"]:
                if keyword in action_text:
                    if config["action"] == "reject_immediately":
                        return {
                            "safe": False,
                            "level": "illegal",
                            "action": "reject",
                            "message": f"检测到违禁操作，已拒绝执行"
                        }
                    elif config["action"] == "pause_and_confirm":
                        return {
                            "safe": False,
                            "level": "dangerous",
                            "action": "pause",
                            "message": f"检测到敏感操作（{category}），请确认是否继续？"
                        }
        
        # 2. 语义级安全检测（覆盖关键词无法匹配的场景）
        safety_score = self.sensitive_detector(
            f"Action: {action_text}\nContext: {context.get('app_name', '')} - {context.get('screen_text', '')}"
        )
        
        if safety_score["label"] == "DANGEROUS" and safety_score["score"] > 0.9:
            return {
                "safe": False,
                "level": "warning",
                "action": "pause",
                "message": "系统检测到潜在风险，请确认操作意图"
            }
        
        return {
            "safe": True,
            "level": "safe",
            "action": "proceed",
            "message": ""
        }
    
    def _action_to_text(self, action: dict) -> str:
        """将动作转为文本用于安全检测"""
        action_type = action.get("type", "")
        if action_type == "gui_click":
            return f"点击位置({action.get('x')}, {action.get('y')})"
        elif action_type == "gui_type":
            return f"输入文本: {action.get('text', '')}"
        elif action_type == "cli_bash":
            return f"执行命令: {action.get('command', '')}"
        return str(action)
    
    def pause_and_confirm(self, action: dict, user_callback) -> bool:
        """
        暂停并等待用户确认
        """
        print(f"\n⚠️ 安全提醒：即将执行以下敏感操作：")
        print(f"  - 操作类型：{action.get('type')}")
        print(f"  - 操作描述：{action.get('desc', '无描述')}")
        print(f"\n请输入 Y 确认执行，或 N 拒绝：")
        
        # 超时机制：30秒无响应自动放弃
        import signal
        def timeout_handler(signum, frame):
            raise TimeoutError("用户确认超时")
        
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(30)
        
        try:
            user_input = input().strip().upper()
            signal.alarm(0)
            return user_input == "Y"
        except TimeoutError:
            print("⏰ 确认超时，已自动取消操作")
            return False

def deep_search_with_gui(query: str):
    """
    GUI + DeepSearch 协同工作流
    
    流程：
    1. GUI打开浏览器搜索关键词
    2. 定位数据源URL
    3. CLI下载并分析数据
    4. GUI操作页面完成后续动作
    """
    workflow = [
        # GUI阶段：打开浏览器搜索
        {"type": "gui_click", "x": 100, "y": 50, "desc": "打开浏览器"},
        {"type": "gui_click", "x": 200, "y": 80, "desc": "点击地址栏"},
        {"type": "gui_type", "text": query, "desc": "输入搜索词"},
        {"type": "gui_key_combo", "keys": ["Enter"], "desc": "开始搜索"},
        
        # CLI阶段：下载数据
        {"type": "cli_bash", "command": f"curl -s 'https://api.example.com/search?q={query}' | jq '.results' > data.json", "desc": "API数据获取"},
        {"type": "cli_bash", "command": "cat data.json | python3 -c \"import json,sys; d=json.load(sys.stdin); print(f'找到{len(d)}条结果')\"", "desc": "数据分析"},
        
        # DeepSearch阶段：深度推理
        {"type": "deep_search", "query": f"分析{query}的最新趋势，生成报告"},
        
        # GUI阶段：呈现结果
        {"type": "cli_bash", "command": "python3 generate_report.py", "desc": "生成报告"},
        {"type": "gui_click", "x": 300, "y": 400, "desc": "打开报告文件"},
    ]
    
    return workflow

# 使用vLLM部署Qwen-UI-Agent-7B
pip install vllm qwen-vl-utils

# 启动模型服务
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen-UI-Agent-7B \
    --trust-remote-code \
    --dtype bfloat16 \
    --max-model-len 32768 \
    --gpu-memory-utilization 0.9 \
    --port 8000

class QwenUIAgent:
    """完整的GUI智能体运行循环"""
    
    def __init__(self, model_path: str, device_type: str = "android"):
        self.model = self._load_model(model_path)
        self.executor = GUIExecutor(device_type)
        self.safety = SafetyController()
        self.memory = []  # 短时记忆
        
    def run(self, task: str, max_steps: int = 100):
        """执行任务的完整循环"""
        step = 0
        screenshot = self._capture_screen()
        task_history = [{"role": "system", "content": f"任务: {task}"}]
        
        while step < max_steps:
            # 1. 构建上下文
            messages = task_history + [
                {
                    "role": "user",
                    "content": [
                        {"type": "image", "image": screenshot},
                        {"type": "text", "text": "当前屏幕如上，请输出下一步操作"}
                    ]
                }
            ]
            
            # 2. 模型推理
            actions = self._predict(messages)
            
            # 3. 安全检查
            for action in actions:
                check = self.safety.check_action(
                    action, 
                    {"app_name": self._current_app(), "screen_text": self._ocr(screenshot)}
                )
                
                if check["action"] == "reject":
                    print(f"❌ {check['message']}")
                    return {"status": "rejected", "reason": check["message"]}
                elif check["action"] == "pause":
                    confirmed = self.safety.pause_and_confirm(action)
                    if not confirmed:
                        print(f"⏭️ 用户已拒绝操作")
                        continue
                
                # 4. 执行动作
                result = self.executor.execute(action)
                step += 1
                
                # 5. 更新记忆
                self.memory.append({"action": action, "result": result})
                task_history.append({
                    "role": "assistant", 
                    "content": f"执行: {action}"
                })
                
                # 6. 检查完成
                if self._is_complete(task, screenshot):
                    return {"status": "success", "steps": step}
                
                screenshot = self._capture_screen()
        
        return {"status": "max_steps_reached", "steps": step}

def compute_grpo_loss(model, rollouts, rewards, group_size=8):
    """
    GRPO损失计算：在组内进行相对优势估计
    
    Args:
        rollouts: 采样轨迹列表
        rewards: 每个轨迹的奖励
        group_size: 每组轨迹数
    """
    total_loss = 0
    
    # 将轨迹分组
    for i in range(0, len(rollouts), group_size):
        group_rollouts = rollouts[i:i + group_size]
        group_rewards = rewards[i:i + group_size]
        
        # 组内归一化奖励（优势估计）
        mean_reward = np.mean(group_rewards)
        std_reward = np.std(group_rewards) + 1e-8
        advantages = [(r - mean_reward) / std_reward for r in group_rewards]
        
        # 对每个轨迹计算策略损失
        for rollout, advantage in zip(group_rollouts, advantages):
            log_probs = model.compute_log_probs(rollout)
            # 重要性采样比
            ratio = torch.exp(log_probs - rollout.old_log_probs)
            
            # 裁剪策略梯度（PPO风格）
            clipped_ratio = torch.clamp(ratio, 0.8, 1.2)
            policy_loss = -torch.min(
                ratio * advantage,
                clipped_ratio * advantage
            ).mean()
            
            # KL散度正则化（防止策略偏离过远）
            kl_loss = torch.kl_div(
                log_probs, rollout.old_log_probs, 
                log_target=True
            ).mean()
            
            total_loss += policy_loss + 0.01 * kl_loss
    
    return total_loss / len(rollouts)