OpenAdapt：录制一次，确定性回放

源码级技术剖析 —— 基于 v1.25.1 代码库逆向分析。OpenAdapt 是一款开源 RPA（Robotic Process Automation）工具，核心理念是"录制一次，确定性回放"。本文基于 v1.25.1 版本约 28,000 行 Python 源码，从使用方法、后端体系到实现原理做系统拆解，并梳理当前版本的已知局限。

1. 使用方法

1.1 核心工作流

OpenAdapt 的核心流程分三步：录制 (record) → 编译 (compile) → 回放 (replay)。此外还有批量循环执行、故障修复、认证检查等辅助命令。

核心工作流

1.2 录制：record

启动一个带浏览器的录制会话，所有操作（点击、输入、按键、滚动、拖拽）被自动捕获。

bash

# 最简方式：打开指定 URL 的浏览器并开始录制
python -m openadapt_flow record --url https://example.com

# 指定输出目录和名称
python -m openadapt_flow record --url https://example.com \
    --out my_recordings/ --name "登录流程"

# 桌面端录制（Windows 原生应用）
python -m openadapt_flow record --backend windows --agent-url http://localhost:8080 \
    --identifier 0,0,1920,1080

# RDP 远程桌面录制
python -m openadapt_flow record --backend rdp --rdp-host 192.168.1.100 \
    --rdp-user admin --rdp-password pass123

# 参数化录制：将输入值声明为参数（回放时可替换）
# 在录制脚本中使用 recorder.type_text("张三", param="patient_name")

录制产物——一个包含原始证据的目录：

文件	内容
`meta.json`	viewport、URL、参数列表、时间戳
`events.jsonl`	每行一个事件：kind / x / y / text / url_before / url_after / structural_state …
`frames/0000_before.png`	第 0 步执行前截图
`frames/0000_after.png`	第 0 步执行后截图
`frames/0001_before.png`	第 1 步执行前截图（即上一步的 after）

关键设计：录制阶段不做 OCR、不做计算机视觉、不调任何模型。只是截图 + JSON 序列化。所有"理解画面内容"的工作留给编译阶段。这样保证录制实时性，不干扰用户操作节奏。

1.3 编译：compile

将录制目录转化为可回放的 Bundle。编译是纯本地计算，不联网。

bash

# 基本编译
python -m openadapt_flow compile recording_xxx/ --out bundle_xxx/

# 编译时确认参数（交互式）
python -m openadapt_flow compile recording_xxx/ --out bundle_xxx/ --accept-params patient_name

# 从文件读取参数确认
python -m openadapt_flow compile recording_xxx/ --out bundle_xxx/ --params-from decisions.json

编译产物：

文件	用途
`workflow.json`	完整的 Workflow IR：步骤定义、锚点、后置条件、参数
`manifest.json`	内容完整性校验清单（content_digest）
`templates/*.png`	每步的模板裁剪（160×64 px），回放时做模板匹配
`workflow.py`	人类可读的 Python 伪代码，仅用于审查，不参与执行

1.4 回放：replay

bash

# 无头回放（默认，不弹出浏览器窗口）
python -m openadapt_flow replay bundle_xxx/

# 有窗口回放（观察执行过程）
python -m openadapt_flow replay bundle_xxx/ --headed

# 指定部署配置（连接真实系统、启用效果验证）
python -m openadapt_flow replay bundle_xxx/ --config deployment.yaml

1.5 批量循环：for-each

对一个预定义工作列表（CSV / JSON）逐行执行同一个 Bundle 的循环体。

bash

# 准备 worklist.csv:
#   patient_name
#   张三
#   李四
#   王五

python -m openadapt_flow compile recording_xxx/ --accept-params patient_name
python -m openadapt_flow for-each bundle_xxx/ --worklist worklist.csv

注意：循环的迭代次数由 worklist 行数预先决定，不能运行时从页面动态提取目标。这是 OpenAdapt 与 Playwright 脚本的本质区别。

1.6 其他命令速览

命令	功能
`demo-record`	启动 MockMed 模拟医疗系统并录制标准演示流程
`tutorial`	运行交互式教学
`induce`	从多个录制中归纳参数化程序（多轨迹归纳）
`run`	完整录制 + 编译 + 回放，一站式执行
`resume`	从上次中断处继续回放
`repair`	对回放失败的 Bundle 进行故障修复
`lint`	静态检查 Bundle 的覆盖率和风险等级
`certify`	按策略（demo/standard/regulated）认证 Bundle 是否可安全无人值守运行
`bench`	对 Bundle 重复回放 N 次并汇总成功率
`visualize`	生成 Bundle 的可视化程序图
`seal`	将 Bundle 密封打包，禁止修改
`sanitize`	脱敏：擦除录制中的敏感数据（姓名、身份证号等）
`emit-skill`	将 Bundle 导出为 AI Agent 可调用的 Skill
`emit-mcp`	将 Bundle 导出为 MCP 工具

2. 支持的后端类型

OpenAdapt 通过 backend.py 中的 Protocol 体系支持多种驱动方式，由 --backend 参数选择。后端只负责 截图 + 执行操作，不参与编译逻辑。

2.1 后端类型一览

后端	标记	底层驱动	画面来源	操作方式	适用场景
Web	默认	Playwright / Chromium	浏览器视口截图	Playwright 鼠标/键盘 API	Web 应用自动化（最常见）
Windows	`--backend windows`	WAA HTTP Agent	桌面截图 API	通过 agent HTTP 接口发送坐标操作	Windows 原生桌面应用
macOS	`--backend macos`	macOS Accessibility API	单应用窗口截图	Accessibility 动作 + 坐标点击	macOS 原生应用
Linux	`--backend linux`	AT-SPI	单应用窗口截图	AT-SPI 动作，可选全局指针/键盘兜底	Linux 桌面应用
RDP	`--backend rdp`	FreeRDP / aardwolf	远程桌面画面截图	坐标点击/输入（像素级）	远程桌面、VDI、堡垒机
Citrix	`--backend citrix`	Citrix Workspace 窗口驱动	Citrix 客户端窗口截图	本地窗口内坐标操作	Citrix 虚拟桌面

2.2 Backend Protocol 分层体系

OpenAdapt 的后端不是单一接口，而是一组 可选 Protocol。不同后端按能力实现不同 Protocol，编译器据此决定能生成哪些信息：

Protocol	提供的能力	缺失的后果
`StructuralBackend`	URL、page_title、page_count	无法生成 structural postcondition
`IdentityBackend`	DOM/a11y 结构化文本 `structured_text_at(x,y)`	只能靠 OCR 做身份验证（O/0、l/1 字形混淆风险）
`StructuralActionBackend`	DOM 选择器 / UIA 标识符	只能用视觉模板匹配定位，无法用 CSS selector
`FieldLabelBackend`	当前聚焦字段的标签文本	编译时只能用 OCR 推测字段标签
`EffectBackend`	系统记录的写入痕迹（如数据库变更）	无法验证操作是否真的写入了系统
`TextIntegrityBackend`	输入框的实际字符值（非 OCR 猜测）	输入文本只能通过 OCR 读回验证

Web 后端最完整：web 后端基于 Playwright 实现了全部 Protocol，因此 Browser 录制的 Bundle 具有最丰富的锚点和验证手段（DOM 选择器 + 结构化身份 + visual template）。RDP/Citrix 后端则退回纯像素模式，只能依赖视觉模板匹配和 OCR。

3. 实现原理（源码级分析）

3.1 架构总览

OpenAdapt 的核心架构分为三个独立阶段。所有分析均基于 v1.25.1 源码，路径位于 openadapt_flow 包内。

架构总览

3.2 录制阶段：Recorder（约 484 行）

3.2.1 事件捕获循环

录制器的核心是 Recorder._record() 方法（recorder.py:260）。每次用户操作触发以下序列：

python

# recorder.py:260 _record() 方法的执行流程（简化）

1. 截取 before_png = backend.screenshot()
2. 采集 structural_state（URL / title / page_count）
3. 执行操作：backend.click(x, y) / backend.type_text(text) / ...
4. 等待画面静止：_wait_settled()
   └─ 轮询截图，直到连续 N 帧 phash 一致（默认超时 2s）
5. 截取 after_png = backend.screenshot()
6. 写入 events.jsonl 一行，保存 before.png / after.png

3.2.2 画面静止检测

_wait_settled()（recorder.py:460）使用感知哈希（phash）检测页面是否渲染完毕：

python

# recorder.py:460 _wait_settled() 核心逻辑

deadline = time.monotonic() + self._settle_timeout_s  # 默认 2 秒
png = self._backend.screenshot()
prev = _phash(png)
consecutive = 1

while consecutive < SETTLE_CONSECUTIVE and time.monotonic() < deadline:
    time.sleep(SETTLE_INTERVAL_S)   # 默认 0.3 秒
    png = self._backend.screenshot()
    curr = _phash(png)
    if phash_distance(prev, curr) == 0:
        consecutive += 1
    else:
        consecutive = 1
    prev = curr

return png  # 最后一张稳定帧

3.2.3 结构化信息被动采集

录制器还记录一份 structural_state（recorder.py:433），包括每步前后的 URL、页面标题、标签页数量：

python

# recorder.py:433 _structural_state()
state = {}
for attr, key in (("url", "url"), ("page_title", "title"), ("page_count", "pages")):
    value = getattr(self._backend, attr, None)
    if value is not None:
        state[key] = value
return state

3.3 编译阶段：Compiler（约 2,203 行）

编译器是整个系统最复杂的模块，输入录制目录，输出自包含的 Bundle。

3.3.1 锚点生成（Anchor Generation）

这是回放时"如何找到目标"的核心。每步提取以下锚点组件（compile.py:1581）：

python

# compile.py:1581 锚点构建逻辑（简化）

# 1. 模板裁剪
crop_region = _discriminative_crop_region(frame, click_point)
#    默认 160×64，如像素方差不足则阶梯扩至 640×256
#    CROP_GROWTH_LADDER = ((160,64), (240,96), (320,128), (480,192), (640,256))

# 2. OCR 文字
ocr_text = _best_crop_text(ocr(before_png, region=crop_region))

# 3. 上下文文字（用于身份验证）
context_text = identifier_text_from_lines(frame_lines, ...)

# 4. 地标（用于几何校准）
landmarks = _landmarks_for(frame_lines, crop_region, click_point)
#    取帧内 10~12 个最显著的 OCR 行作为锚点参照系

# 5. 结构化定位器（DOM / UIA）
structural = StructuralLocator.model_validate(event.get("structural"))

# 最终 Anchor 对象
anchor = Anchor(
    template=template_rel,      # 模板 PNG 路径
    region=crop_region,         # 裁剪区域坐标
    click_point=click,          # 点击坐标
    ocr_text=ocr_text,          # 锚点文字
    context_text=context_text,  # 身份上下文
    structural=structural,      # DOM 选择器
    landmarks=landmarks,        # 地标列表
)

3.3.2 后置条件挖掘（Postcondition Mining）

_postconditions() 函数（compile.py:656）从 before/after 帧差异中自动推断验证条件。

REGION_STABLE（区域稳定性）：

python

# compile.py:722 REGION_STABLE 生成逻辑

# 1. 像素差分找最大变化区域
changed = _largest_changed_region(before_png, after_png)
#    流程：cv2.absdiff → 阈值化(25) → cv2.findContours → 最大连通区域

# 2. 添加 24px 内边距（包裹结构边框）
padded = _pad_region(changed, frame_w, frame_h)

# 3. 自变异检测（过滤动画/时钟/toast）
if next_before_png is not None:
    phash_now = phash_png(after_png, region=padded)
    phash_next = phash_png(next_before_png, region=padded)
    if phash_distance(phash_now, phash_next) > REGION_STABLE_TOLERANCE:
        changed = None  # 丢弃！该区域是自变异的

# 4. 参数隔离（如果区域含有参数文字则切掉参数行）
if exclude_texts and _param_text_in_region(padded, after_lines, exclude_texts):
    band = _param_free_band(padded, carriers, after_lines)
    if band is not None:
        padded = band

# 5. 生成 postcondition
expect.append(Postcondition(
    kind=PostconditionKind.REGION_STABLE,
    region=padded,
    phash=phash_png(after_png, region=padded),
    phash_tolerance=REGION_STABLE_TOLERANCE,  # 16
))

TEXT_PRESENT（新文本出现）：

python

# compile.py:536 _new_text_postcondition() 核心逻辑

# 1. OCR after 帧，找出 before 帧中不存在的新文本行
new_lines = [line for line in after_lines if not seen_before(line)]

# 2. 过四道过滤器
for line in new_lines:
    text = normalize_text(line.text)

    # 过滤器 A：长度 ≥ MIN_TEXT_PRESENT_LEN (3)
    if len(text) < 3: continue

    # 过滤器 B：挥发性分类器
    if volatility.is_volatile(text, reference_date): continue
    #    拦截：时钟(18:38)、相对时间(3 min ago)、计数器(1-5 of 12)

    # 过滤器 C：点击区域去重
    if _matches_click_text(text, click_text): continue

    # 过滤器 D：已在前一帧可见（模糊匹配）
    if _was_visible_in_before(text, before_lines): continue

    candidates.append((score, text))

# 3. 取得分最高的一个
return Postcondition(kind=PostconditionKind.TEXT_PRESENT, text=chosen)

结构性兜底（Structural Fallback）：

python

# compile.py:880 _structural_postconditions()
# 仅当前两步挖出零视觉后置条件时触发

pcs = []
if pages_after > pages_before:
    pcs.append(Postcondition(kind=PostconditionKind.NEW_TAB_OPENED))
if url_before != url_after:
    pcs.append(Postcondition(kind=PostconditionKind.URL_CHANGED))
elif title_before != title_after:
    pcs.append(Postcondition(kind=PostconditionKind.TITLE_CHANGED))
return pcs

3.3.3 挥发性分类器（Volatility Classifier）

volatility.py（366 行）是编译器中最关键的防御模块。它用一组正则表达式标记不可靠的文本证据：

分类器	拦截模式	示例	原因
`CLOCK_RE`	`\d{1,2}:\d{2}` 及 OCR 片段 `:01`	18:38、6:05、12:45:59	时钟分钟会走
`DOT_CLOCK_RE`	`\d+\.\d{2}` 欧式时钟	18.38、updated 8.30	同上
`RELATIVE_TIME_RE`	`\d+ (min	hour)s? ago`	3 min ago、just now、moments ago
`COUNT_RE`	`\d+ to \d+ of \d+`	1 to 5 of 12 entries	分页计数是瞬时状态
相对日期词	独立的 today / yesterday / now	Today、Yesterday	作为消息列表分组头时是瞬时的
近日期	距离录制日期 < 7 天的日期	2026-08-06（录制日期附近）	内容时间线而非身份数据

设计亮点：挥发性分类器对日期使用"近/远"而非简单的黑名单：距离录制日期 ≤7 天的日期视为内容时序（挥发性），而远的日期（如出生日期 DOB）视为身份数据（稳定性）。这比粗暴的"所有日期都过滤"方案更精细。

3.3.4 内容完整性自校验

编译完成后，manifest.json 中写入 content_digest，由 compute_content_digest() 函数计算。它使用 model_dump(exclude_none=True) 规范化序列化后做 SHA256，而非简单 json.dumps。这意味着任何对 workflow.json 的修改都需要重算 digest。

3.4 回放阶段：Replayer

3.4.1 分辨率阶梯（Resolution Ladder）

回放引擎对每步锚定采用从快到慢的多级策略（ir.py 中定义的 Resolution 枚举）：

1. structural        # DOM 选择器 / UIA 标识符 → 直接定位（毫秒级）
2. template_match    # 模板 PNG 在当前帧上做 cv2.matchTemplate
3. ocr               # OCR 当前帧，用锚点文本匹配
4. landmark          # 用地标的 OCR 位置做几何变换推算目标位置
5. geometry          # 纯坐标偏移（最后兜底）

当某一级失败时，引擎自动"自愈"（heal）到下一级。历史上的 replay 日志中 heal_events 字段记录了每次降级的详细信息。

3.4.2 身份验证

在点击/输入等操作前，回放引擎会验证"当前画面中的目标是对的"：

python

# 身份验证流程

1. band_region(click_point, crop_height, frame_size)
   → 在锚点位置周围取一个水平的"身份带"

2. OCR 这个 band 区域

3. 与录制时的 context_text 比对
   - 如果是参数化字段 → 用当前运行的参数值重新匹配
   - 如果模糊匹配通过 → 确认身份，执行操作
   - 如果匹配失败 → 降级到下一级分辨率阶梯

3.5 核心数据流总结

核心数据流

4. 当前不足之处

4.1 后置条件无法在录制时控制

postcondition 完全由编译器自动生成，用户没有 CLI 参数或 API 来干预。对于动态页面（搜索引擎、实时数据面板），编译器生成的 region_stable 可能监控会不断变化的区域，导致 replay 必然 HALT。

当前唯一解法：编译后手动编辑 workflow.json，清空有问题的 expect 数组，然后重算 content_digest。这是纯手工流程，没有工具支持。

4.2 不能运行时动态决策

OpenAdapt 的 loop authoring 要求预先声明的 worklist（CSV/JSON），不能从页面动态提取目标。无法实现"搜索关键词 → 爬取前 N 个结果"这类需要运行时发现的场景。

根因：架构设计上，Workflow 是静态 IR（Intermediate Representation），所有步骤在编译时确定。运行时只是"按图索骥地执行 + 自愈"，不做任何目标发现。

4.3 帧差分区域可能过大

_largest_changed_region() 在页面整体变化时（URL 跳转、整页刷新）可能返回全屏区域，导致 region_stable 监控整个视口。在存在动态元素（时钟、广告轮播）的页面，全屏 phash 几乎不可能稳定。

源码位置：compile.py 中 _largest_changed_region() 只取最大连通区域，不做语义分割。

4.4 自变异检测依赖帧间隔

编译器用 next_before_png（下一步执行前的帧）与 after_png 比较来检测自变异区域。如果录制时操作节奏太快，时钟还没跳到下一分钟、动画还没播完，自变异检测就会漏过。

源码位置：compile.py:722 的自变异检查仅比较两帧 phash 距离，不做时序建模。

4.5 不支持跨页爬取或数据提取

OpenAdapt 的目标是"自动化操作"，不是"数据抓取"。没有内置的 DOM 提取、列表遍历、分页处理能力。如果需要从页面上提取数据（如搜索结果链接、表格行），它做不到。

对比：Playwright 脚本可以直接 page.locator('.result-item') 提取元素列表，OpenAdapt 只能识别录制时见过的那个特定坐标。

4.6 锚点身份验证的局限性

在纯像素后端（RDP/Citrix）中，没有 StructuralActionBackend 和 IdentityBackend，只能靠 OCR 做身份验证。OCR 对字形混淆（O/0、l/1、I/l）无法区分，源码 backend.py:126 的文档明确承认：

"An adversarial review proved the OCR-only identity path cannot close the same-name / same-DOB glyph-collapse case: two DIFFERENT patients whose MRN differs only by an O/0 or l/1 glyph render to a byte-identical OCR band — so no function downstream of OCR can distinguish them"

4.7 编译产物的可维护性

编译后的 workflow.json 内部嵌入了 manifest，而 manifest.json 又是独立文件。修改 workflow 后需要同步更新两处的 digest。实操中常见改了 manifest.json 但没改 workflow.json 内嵌的 content_digest，导致 BundleIntegrityError。

更根本的问题：没有提供官方的 post-hoc 编辑工具或 CLI。所有修改都是手工 JSON 编辑 + 调用库函数重算 digest。

4.8 执行模型不可见

replay 默认 headless 模式（--headed 是 store_true，默认 False，headless = not headed = True）。用户看不到浏览器窗口，只能等 REPORT 输出。调试体验差。

源码位置：__main__.py:3338 定义 --headed action="store_true"，与 record 的默认可见行为相反。

免责声明：本文档基于 OpenAdapt v1.25.1 源码逆向分析生成，所有代码片段均来自实际源码文件。分析日期：2026-08-07，源码总行数约 28,000 行 Python。

OpenAdapt：录制一次，确定性回放 ​

目录 ​

1. 使用方法 ​

1.1 核心工作流 ​

1.2 录制：record ​

1.3 编译：compile ​

1.4 回放：replay ​

1.5 批量循环：for-each ​

1.6 其他命令速览 ​

2. 支持的后端类型 ​

2.1 后端类型一览 ​

2.2 Backend Protocol 分层体系 ​

3. 实现原理（源码级分析） ​

3.1 架构总览 ​

3.2 录制阶段：Recorder（约 484 行） ​

3.2.1 事件捕获循环 ​

3.2.2 画面静止检测 ​

3.2.3 结构化信息被动采集 ​

3.3 编译阶段：Compiler（约 2,203 行） ​

3.3.1 锚点生成（Anchor Generation） ​

3.3.2 后置条件挖掘（Postcondition Mining） ​

3.3.3 挥发性分类器（Volatility Classifier） ​

3.3.4 内容完整性自校验 ​

3.4 回放阶段：Replayer ​

3.4.1 分辨率阶梯（Resolution Ladder） ​

3.4.2 身份验证 ​

3.5 核心数据流总结 ​

4. 当前不足之处 ​

4.1 后置条件无法在录制时控制 ​

4.2 不能运行时动态决策 ​

4.3 帧差分区域可能过大 ​

4.4 自变异检测依赖帧间隔 ​

4.5 不支持跨页爬取或数据提取 ​

4.6 锚点身份验证的局限性 ​

4.7 编译产物的可维护性 ​

4.8 执行模型不可见 ​

OpenAdapt：录制一次，确定性回放

目录

1. 使用方法

1.1 核心工作流

1.2 录制：record

1.3 编译：compile

1.4 回放：replay

1.5 批量循环：for-each

1.6 其他命令速览

2. 支持的后端类型

2.1 后端类型一览

2.2 Backend Protocol 分层体系

3. 实现原理（源码级分析）

3.1 架构总览

3.2 录制阶段：Recorder（约 484 行）

3.2.1 事件捕获循环

3.2.2 画面静止检测

3.2.3 结构化信息被动采集

3.3 编译阶段：Compiler（约 2,203 行）

3.3.1 锚点生成（Anchor Generation）

3.3.2 后置条件挖掘（Postcondition Mining）

3.3.3 挥发性分类器（Volatility Classifier）

3.3.4 内容完整性自校验

3.4 回放阶段：Replayer

3.4.1 分辨率阶梯（Resolution Ladder）

3.4.2 身份验证

3.5 核心数据流总结

4. 当前不足之处

4.1 后置条件无法在录制时控制

4.2 不能运行时动态决策

4.3 帧差分区域可能过大

4.4 自变异检测依赖帧间隔

4.5 不支持跨页爬取或数据提取

4.6 锚点身份验证的局限性

4.7 编译产物的可维护性

4.8 执行模型不可见