通过桌面录屏实现自动化 RPA 的最佳实践

传统的 RPA（Robotic Process Automation）工具通常需要手动编写脚本或使用可视化编排工具，学习成本高且维护困难。随着多模态 AI 的发展，一种新的范式正在兴起：通过录制用户的桌面操作，让 AI 自动理解并复现这些操作。这种方式大大降低了自动化的门槛，让业务人员也能快速构建自动化流程。

为什么选择录屏驱动的 RPA？

传统 RPA 的痛点

1. 高技术门槛：需要编写代码或学习复杂的可视化工具
2. 维护成本高：UI 变化时脚本经常失效
3. 场景覆盖有限：难以处理动态内容和异常情况
4. 缺乏智能：无法理解操作的语义，只是机械重复

录屏 + AI 的优势

• 零代码上手：只需录制一次操作，AI 自动学习
• 语义理解：AI 理解操作意图，而非死记硬背坐标
• 智能适应：界面变化时能自动调整
• 异常处理：可以识别并处理意外情况

整体架构设计

┌──────────────────────────────────────────────────────────────────┐
│                   Screen Recording RPA System                     │
├──────────────────────────────────────────────────────────────────┤
│  Layer 1: 录制层                                                   │
│  ├── 屏幕录制模块 (视频 + 音频)                                      │
│  ├── 鼠标/键盘事件捕获                                              │
│  └── 时间戳同步                                                    │
├──────────────────────────────────────────────────────────────────┤
│  Layer 2: 理解层                                                   │
│  ├── 视频帧提取与关键帧检测                                          │
│  ├── 多模态 LLM 分析 (GPT-4V / Claude Vision / Gemini)             │
│  └── 操作序列抽象                                                  │
├──────────────────────────────────────────────────────────────────┤
│  Layer 3: 执行层                                                   │
│  ├── 元素定位引擎 (视觉 + 语义)                                      │
│  ├── 动作执行器 (pyautogui / playwright)                           │
│  └── 验证与重试机制                                                 │
├──────────────────────────────────────────────────────────────────┤
│  Layer 4: 编排层                                                   │
│  ├── 流程定义与管理                                                 │
│  ├── 调度与触发器                                                  │
│  └── 监控与日志                                                    │
└──────────────────────────────────────────────────────────────────┘

Step 1: 构建录制模块

1.1 屏幕录制

我们需要同时录制屏幕画面和用户的输入事件。

import cv2
import numpy as np
import pyautogui
from pynput import mouse, keyboard
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
from datetime import datetime
import threading
import json
import time

@dataclass
class InputEvent:
    """表示一个输入事件"""
    timestamp: float          # 相对于录制开始的时间（秒）
    event_type: str          # "click", "scroll", "key_press", "key_release", "move"
    position: Optional[Tuple[int, int]] = None  # 鼠标位置
    button: Optional[str] = None                 # 鼠标按钮
    key: Optional[str] = None                    # 按键
    data: dict = field(default_factory=dict)     # 额外数据

@dataclass
class RecordingSession:
    """录制会话"""
    video_path: str
    events: List[InputEvent]
    start_time: float
    end_time: float
    screen_size: Tuple[int, int]
    fps: int = 10

class ScreenRecorder:
    """
    屏幕录制器

    同时捕获：
    1. 屏幕视频
    2. 鼠标点击、移动、滚动
    3. 键盘输入
    """

    def __init__(self, output_dir: str = "./recordings", fps: int = 10):
        self.output_dir = output_dir
        self.fps = fps
        self.is_recording = False
        self.events: List[InputEvent] = []
        self.start_time: float = 0
        self.screen_size = pyautogui.size()

        # 录制线程
        self._video_thread: Optional[threading.Thread] = None
        self._video_writer: Optional[cv2.VideoWriter] = None

        # 输入监听器
        self._mouse_listener: Optional[mouse.Listener] = None
        self._keyboard_listener: Optional[keyboard.Listener] = None

    def _on_mouse_click(self, x: int, y: int, button, pressed: bool):
        """鼠标点击回调"""
        if not self.is_recording:
            return

        event = InputEvent(
            timestamp=time.time() - self.start_time,
            event_type="click" if pressed else "release",
            position=(x, y),
            button=str(button),
            data={"pressed": pressed}
        )
        self.events.append(event)

    def _on_mouse_scroll(self, x: int, y: int, dx: int, dy: int):
        """鼠标滚动回调"""
        if not self.is_recording:
            return

        event = InputEvent(
            timestamp=time.time() - self.start_time,
            event_type="scroll",
            position=(x, y),
            data={"dx": dx, "dy": dy}
        )
        self.events.append(event)

    def _on_key_press(self, key):
        """键盘按下回调"""
        if not self.is_recording:
            return

        try:
            key_str = key.char if hasattr(key, 'char') else str(key)
        except AttributeError:
            key_str = str(key)

        event = InputEvent(
            timestamp=time.time() - self.start_time,
            event_type="key_press",
            key=key_str
        )
        self.events.append(event)

    def _record_video(self, output_path: str):
        """视频录制线程"""
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        self._video_writer = cv2.VideoWriter(
            output_path,
            fourcc,
            self.fps,
            self.screen_size
        )

        interval = 1.0 / self.fps

        while self.is_recording:
            frame_start = time.time()

            # 截取屏幕
            screenshot = pyautogui.screenshot()
            frame = np.array(screenshot)
            frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)

            # 在帧上绘制鼠标位置（用于调试）
            mouse_pos = pyautogui.position()
            cv2.circle(frame, mouse_pos, 10, (0, 0, 255), -1)

            self._video_writer.write(frame)

            # 控制帧率
            elapsed = time.time() - frame_start
            if elapsed < interval:
                time.sleep(interval - elapsed)

        self._video_writer.release()

    def start_recording(self, session_name: str = None) -> str:
        """
        开始录制

        Args:
            session_name: 会话名称，用于生成文件名

        Returns:
            录制文件路径
        """
        if self.is_recording:
            raise RuntimeError("Already recording")

        import os
        os.makedirs(self.output_dir, exist_ok=True)

        if session_name is None:
            session_name = datetime.now().strftime("%Y%m%d_%H%M%S")

        video_path = f"{self.output_dir}/{session_name}.mp4"

        self.events = []
        self.start_time = time.time()
        self.is_recording = True

        # 启动视频录制线程
        self._video_thread = threading.Thread(
            target=self._record_video,
            args=(video_path,)
        )
        self._video_thread.start()

        # 启动输入监听
        self._mouse_listener = mouse.Listener(
            on_click=self._on_mouse_click,
            on_scroll=self._on_mouse_scroll
        )
        self._keyboard_listener = keyboard.Listener(
            on_press=self._on_key_press
        )
        self._mouse_listener.start()
        self._keyboard_listener.start()

        print(f"Recording started: {video_path}")
        return video_path

    def stop_recording(self) -> RecordingSession:
        """
        停止录制

        Returns:
            RecordingSession 对象
        """
        if not self.is_recording:
            raise RuntimeError("Not recording")

        self.is_recording = False
        end_time = time.time()

        # 停止监听器
        if self._mouse_listener:
            self._mouse_listener.stop()
        if self._keyboard_listener:
            self._keyboard_listener.stop()

        # 等待视频线程结束
        if self._video_thread:
            self._video_thread.join()

        # 获取视频路径
        video_path = self._video_writer.getfilename() if self._video_writer else ""

        session = RecordingSession(
            video_path=video_path,
            events=self.events.copy(),
            start_time=self.start_time,
            end_time=end_time,
            screen_size=self.screen_size,
            fps=self.fps
        )

        # 保存事件数据
        events_path = video_path.replace('.mp4', '_events.json')
        self._save_events(events_path)

        print(f"Recording stopped. Duration: {end_time - self.start_time:.1f}s")
        print(f"Captured {len(self.events)} input events")

        return session

    def _save_events(self, path: str):
        """保存事件到 JSON 文件"""
        events_data = [
            {
                "timestamp": e.timestamp,
                "event_type": e.event_type,
                "position": e.position,
                "button": e.button,
                "key": e.key,
                "data": e.data
            }
            for e in self.events
        ]

        with open(path, 'w') as f:
            json.dump(events_data, f, indent=2)

# generated by hugo's coding agent

1.2 关键帧提取

录制完成后，我们需要从视频中提取关键帧，用于后续的 AI 分析。

import cv2
import numpy as np
from typing import List, Tuple
from dataclasses import dataclass

@dataclass
class KeyFrame:
    """关键帧"""
    frame_index: int
    timestamp: float
    image: np.ndarray
    associated_events: List[InputEvent]
    change_score: float  # 与前一帧的变化程度

class KeyFrameExtractor:
    """
    关键帧提取器

    策略：
    1. 基于画面变化检测
    2. 基于输入事件（点击、按键）
    3. 固定间隔采样
    """

    def __init__(self,
                 change_threshold: float = 0.1,
                 min_interval: float = 0.5,
                 event_window: float = 0.2):
        """
        Args:
            change_threshold: 画面变化阈值（0-1）
            min_interval: 关键帧最小间隔（秒）
            event_window: 事件关联时间窗口（秒）
        """
        self.change_threshold = change_threshold
        self.min_interval = min_interval
        self.event_window = event_window

    def extract(self,
                video_path: str,
                events: List[InputEvent]) -> List[KeyFrame]:
        """
        从视频中提取关键帧

        Args:
            video_path: 视频文件路径
            events: 输入事件列表

        Returns:
            关键帧列表
        """
        cap = cv2.VideoCapture(video_path)
        fps = cap.get(cv2.CAP_PROP_FPS)

        keyframes = []
        prev_frame = None
        prev_keyframe_time = -float('inf')
        frame_index = 0

        while True:
            ret, frame = cap.read()
            if not ret:
                break

            timestamp = frame_index / fps

            # 计算与前一帧的变化
            change_score = 0.0
            if prev_frame is not None:
                change_score = self._compute_change(prev_frame, frame)

            # 判断是否为关键帧
            is_keyframe = False

            # 条件1：画面变化超过阈值
            if change_score > self.change_threshold:
                is_keyframe = True

            # 条件2：有关联的点击事件
            associated = self._find_associated_events(timestamp, events)
            click_events = [e for e in associated if e.event_type == "click"]
            if click_events:
                is_keyframe = True

            # 条件3：满足最小间隔
            if is_keyframe and (timestamp - prev_keyframe_time) >= self.min_interval:
                keyframe = KeyFrame(
                    frame_index=frame_index,
                    timestamp=timestamp,
                    image=frame.copy(),
                    associated_events=associated,
                    change_score=change_score
                )
                keyframes.append(keyframe)
                prev_keyframe_time = timestamp

            prev_frame = frame
            frame_index += 1

        cap.release()

        print(f"Extracted {len(keyframes)} keyframes from {frame_index} frames")
        return keyframes

    def _compute_change(self, frame1: np.ndarray, frame2: np.ndarray) -> float:
        """计算两帧之间的变化程度"""
        # 转换为灰度图
        gray1 = cv2.cvtColor(frame1, cv2.COLOR_BGR2GRAY)
        gray2 = cv2.cvtColor(frame2, cv2.COLOR_BGR2GRAY)

        # 计算结构相似性
        diff = cv2.absdiff(gray1, gray2)
        change_ratio = np.mean(diff) / 255.0

        return change_ratio

    def _find_associated_events(self,
                                 timestamp: float,
                                 events: List[InputEvent]) -> List[InputEvent]:
        """找到时间窗口内的关联事件"""
        return [
            e for e in events
            if abs(e.timestamp - timestamp) <= self.event_window
        ]

# generated by hugo's coding agent

Step 2: 使用多模态 AI 理解操作

2.1 构建分析 Prompt

import base64
import io
from PIL import Image
from typing import List, Dict, Any

class OperationAnalyzer:
    """
    使用多模态 LLM 分析操作序列

    支持的模型：
    - OpenAI GPT-4 Vision
    - Anthropic Claude 3
    - Google Gemini Pro Vision
    """

    ANALYSIS_PROMPT = """你是一个 RPA 自动化专家。请分析以下屏幕截图序列，理解用户正在执行的操作。

对于每个截图：
1. 描述当前屏幕的状态
2. 识别用户点击/操作的元素
3. 推断用户的意图

最后，请将整个操作序列抽象为一个可重复执行的自动化流程。

输出格式（JSON）：
{
    "workflow_name": "流程名称",
    "description": "流程描述",
    "steps": [
        {
            "step_id": 1,
            "action": "click|type|scroll|wait|verify",
            "target": {
                "description": "目标元素的描述",
                "visual_features": ["视觉特征1", "特征2"],
                "text_content": "元素包含的文本（如有）",
                "relative_position": "相对位置描述"
            },
            "parameters": {
                "text": "要输入的文本（type 操作）",
                "direction": "滚动方向（scroll 操作）",
                "condition": "验证条件（verify 操作）"
            },
            "expected_result": "执行后的预期结果"
        }
    ],
    "preconditions": ["前置条件1", "前置条件2"],
    "error_handling": {
        "timeout": "超时处理策略",
        "element_not_found": "元素未找到处理策略"
    }
}
"""

    def __init__(self, model_provider: str = "openai", api_key: str = None):
        self.model_provider = model_provider
        self.api_key = api_key

    def _encode_image(self, image: np.ndarray) -> str:
        """将图像编码为 base64"""
        # 转换为 PIL Image
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        pil_image = Image.fromarray(image_rgb)

        # 压缩并编码
        buffer = io.BytesIO()
        pil_image.save(buffer, format="JPEG", quality=85)
        return base64.b64encode(buffer.getvalue()).decode()

    def _build_messages(self, keyframes: List[KeyFrame]) -> List[Dict]:
        """构建多模态消息"""
        messages = [
            {"role": "system", "content": self.ANALYSIS_PROMPT}
        ]

        # 构建用户消息，包含所有关键帧
        content = []

        for i, kf in enumerate(keyframes):
            # 添加图像
            content.append({
                "type": "image_url",
                "image
---
原文链接: [通过桌面录屏实现自动化 RPA 的最佳实践](https://hugozhu.site/post/2026/121-desktop-screen-recording-rpa-best-practices/)