多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

Workbuddy对接蓝印RPA:AI驱动的流程自动化架构

Workbuddy对接蓝印RPA:AI驱动的流程自动化架构 如果你已经在 RPA 领域写过几个流程大概率会有一种感觉真正卡住效率的往往不是机器人执行的那几秒钟而是“把业务需求翻译成动作序列”的这个过程。一个流程要能在生产环境里稳定跑起来至少需要梳理页面元素、确认触发条件、设计异常分支、处理超时和重试每一步都要花大量时间在调试上。哪怕你用的是影刀、蓝印这类已经做了大量组件的 RPA 工具最终交付的依然是一个需要人工反复维护的脚本。这中间就出现了一个非常现实的问题既然 AI 已经能理解自然语言也能生成代码为什么不能让它直接生成 RPA 流程为什么很多 RPA 厂商的“AI 生成流程”仍然停留在把一句话变成几个固定模板的阶段这篇文章要写的就是一条更务实的对接路径用 Workbuddy 这类具备深度任务规划能力的 AI 工作台对接蓝印 RPA 这样的流程执行平台把“理解需求、拆解步骤、生成流程定义、执行页面操作”拆成两个层次。Workbuddy 负责当“流程架构师”蓝印负责当“执行引擎”。读完这篇文章你可以建立一个可以落地的对接模型拿到一套从需求输入到流程执行的完整代码示例并且知道哪些环节容易出错、哪些做法更适合生产环境。1. 这篇文章真正要解决的问题先说一下为什么这个话题值得写。很多人以为 AI 时代的 RPA 就是“对着机器人说一句话它自动把流程建好”。实际去看现有方案会发现市面上的“AI 生成流程”大多还是模板式生成AI 从一段话里提取几个关键词然后拼装一个固定流程。一旦流程里出现条件分支、循环、数据校验AI 就无能为力了。更麻烦的问题在维护阶段。RPA 流程最大的痛点是脆弱页面改一个按钮 ID脚本就废了领导换一个表格模板解析逻辑全得重写。如果 AI 只是帮你生成了一堆一次性代码那它不但没有解决维护成本反而可能制造更多技术债。所以这里的核心矛盾不是“AI 能不能写 RPA 脚本”而是“AI 能不能像 RPA 工程师一样先做流程架构再做动作编排”。这正好是 Workbuddy 这类 Agent 工具擅长的方向它不只是写代码而是可以管理任务上下文把一个复杂需求拆成多步计划再调用外部能力去执行。再来看蓝印 RPA。蓝印这类平台的强项是执行打开浏览器、填表单、点按钮、抓数据、操作 Excel这些动作它已经封装得很成熟。但它的弱项是编排流程稍微复杂一点依然需要人用流程图去拖拽节点。把两者放在一起就能得到一种比较合理的分工Workbuddy 负责把自然语言需求转化为结构化流程描述。蓝印 RPA 负责按流程描述执行具体动作。中间的连接器负责把 Workbuddy 生成的文件转成蓝印可识别的任务。如果你正在做 RPA 项目或者你在企业里负责自动化推进这篇文章会帮你判断一件事AI 到底应该在自动化体系里扮演什么角色。它不该是接线员它应该是总设计师。2. Workbuddy 与蓝印 RPA 的核心概念和分工要理解这个对接方案先要把两个工具各自的特性和边界搞清楚。很多人把 AI 助手和 RPA 当成同一类产品其实它们的抽象层次完全不同。2.1 Workbuddy一个有 Skill 机制的 AI 工作台从现有材料看Workbuddy 是一个偏“AI 工作台”定位的工具。它能搭建工作台、管理多任务上下文、通过 Skill 机制扩展能力也支持把复杂的任务拆成可执行的步骤链。你可以把它理解成 AI 世界的“项目经理”它能理解你的目标比如“每周一早上从后台导出前一天的订单整理成 Excel 发给运营”。它能规划步骤比如先去后台再筛选数据再导出再清洗再发送。它能通过 Skill 调用外部工具比如执行 Python 脚本、读写文件、调用 API。很多人在用 Workbuddy 时只把它当成一个高级聊天框这其实是最大的浪费。它的价值不在于“对话”而在于“把一个目标稳定地拆成一连串可以被执行的指令”。这就是它适合做 RPA 流程设计层的原因。2.2 蓝印 RPA流程执行层蓝印 RPA 的定位是执行层。它提供组件化的自动操作能力比如浏览器操作、窗口操作、鼠标键盘模拟、Excel 处理、文件操作、数据库访问等。它和影刀这类工具的类似点在于都强调“拖拽式流程设计”降低 RPA 开发门槛。但只要是视觉拖拽方案就绕不开一个局限流程的描述成本最终还是落在人身上。一个流程要描述清楚至少需要每一步操作什么元素、操作顺序是什么、条件分支在什么情况下跳转、失败时如何重试、数据如何流转。这些信息如果让人类写是开发成本如果让 AI 写就只是“描述成本”。2.3 两者的职责边界对接方案里最忌讳的是职责混乱。比如有人会问既然 Workbuddy 能操作浏览器为什么还要蓝印答案很简单浏览器自动化的稳定性要求很高。Workbuddy 作为 AI 工作台适合做规划、生成代码、编排任务流蓝印这类 RPA 引擎在网页元素定位、异常恢复、日志追踪上有更成熟的运行时机制。让 AI 去做页面点击不是不行而是在生产环境里把执行稳定性和流程智能分开才是更可靠的架构。所以推荐的分工是层次角色负责内容Workbuddy流程设计层理解需求、拆解步骤、生成流程描述文件连接适配层翻译器把流程描述文件转成 RPA 执行任务蓝印 RPA流程执行层执行页面操作、异常重试、日志记录这个模型把“做什么”和“怎么做”彻底分离。Workbuddy 不关心按钮的 CSS 选择器长什么样蓝印也不关心业务目标是什么它只需要按步骤执行。3. 对接架构如何用流程定义文件串联两个系统架构设计的核心是确定“中间协议”。我们选择一个非常通用、可读性强、易调试的格式JSON 流程定义文件。整个流程可以描述成这样用户在 Workbuddy 工作台里用自然语言描述自动化需求。Workbuddy 通过自身规划能力生成一份结构化的流程描述保存为workflow.json。对接脚本读取workflow.json解析每一步的动作、选择器、参数。蓝印 RPA 执行器按步骤执行并把执行结果写回workflow_result.json。Workbuddy 读取执行结果向用户汇报完成情况。为什么用 JSON 而不是直接用蓝印原生的流程图格式因为 JSON 有这几个优势通用性好任何语言都能解析。易于调试人工可以直接打开查看。方便版本管理流程变更可以通过 Git 追踪。便于 AI 生成大语言模型输出结构化 JSON 的可靠性远高于直接生成 RPA 工程文件。下面是一份最小化的流程定义示例。{ task_id: daily_order_report, name: 每日订单导出, version: 1.0.0, owner: automation-team, steps: [ { id: step_1, action: open_url, target: https://admin.example.com/login, params: { wait_seconds: 3 } }, { id: step_2, action: fill_input, target: #username, params: { value: {{ENV.ADMIN_USER}} } }, { id: step_3, action: fill_input, target: #password, params: { value: {{ENV.ADMIN_PASS}} } }, { id: step_4, action: click, target: button[typesubmit], params: { wait_seconds: 5 } }, { id: step_5, action: extract_table, target: #order-table, params: { export_path: ./output/orders.csv } } ] }这份文件里有一个非常重要的设计敏感信息不直接写在流程文件里而是通过{{ENV.ADMIN_USER}}这样的占位符引用环境变量。因为 workflow.json 会被 Workbuddy 生成也可能被提交到 Git 仓库如果直接把账号密码写进去等于把生产环境密钥放在了明面上。这个后面会在最佳实践里再详细说。4. 环境准备与前置条件由于这篇文章介绍的是通用思路工具版本建议以实际官方发布为准这里只讲需要准备哪些能力以及它们各自解决什么问题。4.1 搭建 Workbuddy 运行环境Workbuddy 的安装方式以官方文档为准。从材料看它支持桌面端安装也涉及工作台搭建和 Skill 管理。安装完成后你至少需要确认三件事能正常创建工作台或项目空间。能调用 Skill 或自定义脚本能力。能读写本地文件尤其是保存和加载 JSON 文件。如果你想把流程分享给团队建议把 Workbuddy 的项目配置文件纳入 Git 管理。这样每次流程定义的变更都能追溯。4.2 准备 Python 执行环境对接脚本一般用 Python 写因为生态成熟尤其是 Playwright、Selenium 这类自动化库都很稳定。这里以 Playwright 为例。python -m venv .venv source .venv/bin/activate pip install playwright playwright install chromium如果你在 Windows 上使用激活命令改为.venv\Scripts\activate。安装完 Playwright 后建议写一个最小脚本验证浏览器能正常启动不要等整个流程跑完才发现环境问题。4.3 确认蓝印 RPA 的对接方式蓝印 RPA 具体如何接收外部任务以官方 API 或接口文档为准。这里提供一种通用思路蓝印提供本地 API 服务或支持命令行触发或可以通过任务队列接收外部提交的任务描述。如果蓝印支持 HTTP API对接脚本只需要把 workflow.json 通过 POST 请求提交给任务接口。如果不支持退一步的做法是对接脚本直接调用 Playwright 或 Selenium 执行动作把蓝印的部分能力用代码模拟出来。这个方案不是最优但至少能让你先把闭环跑通。5. 核心流程拆解从需求到执行的四步5.1 第一步需求输入与 Workbuddy 规划用户先向 Workbuddy 描述需求。关键点在于不要只给一句话而是给尽量完整的背景目标系统是什么、需要操作哪些页面、数据要导到哪里、频率是什么。让 Workbuddy 生成流程定义前建议在提示词里要求它输出结构化 JSON。你可以把上面那个 workflow.json 的格式作为示例放在上下文里让 AI 严格参照这个 schema 输出。用 Workbuddy 的 Skill 机制把这个 schema 固化下来每次生成流程时自动调用能显著减少格式错误。5.2 第二步流程定义校验Workbuddy 生成 workflow.json 之后不能直接拿去执行。要对 JSON 做两个层面的校验格式校验所有字段是否完整action 是否在支持列表里。逻辑校验比如登录动作之后才能点击后台页面导出动作放在筛选动作之后。为什么不能跳过这一步因为大模型生成的流程偶尔会“一本正经地胡说八道”比如生成了一个不存在的选择器。在进入执行层之前拦住错误成本是最低的。一个简单做法是写一个 validate.py把所有不支持的 action 名称直接报错。SUPPORTED_ACTIONS {open_url, fill_input, click, extract_table, wait} def validate_workflow(data): name data.get(name, unknown) steps data.get(steps, []) if not steps: raise ValueError(f流程 [{name}] 没有定义任何步骤) for step in steps: action step.get(action) if action not in SUPPORTED_ACTIONS: raise ValueError(f不支持的 action: {action}, 请检查流程定义) if target not in step: raise ValueError(f步骤 {step.get(id)} 缺少 target 参数) print(f流程 [{name}] 校验通过, 共 {len(steps)} 步)5.3 第三步执行器运行执行器读取 workflow.json逐条执行。每一步都记录开始时间、结束时间、状态、错误信息。这个日志非常重要因为 RPA 流程一旦在深夜自动运行失败你只能靠日志定位问题。5.4 第四步结果回写与闭环执行完成后把结果写回 result 文件。Workbuddy 读取结果后可以形成闭环成功了通知用户文件已导出失败了分析是哪一步失败尝试修正流程定义。这一步已经具备“自动化运维”的雏形。6. 完整示例与代码实现这一节我们用一个真实的小任务走通全流程自动登录一个假设的管理后台导出一张订单表。注意这里的关键是展示对接逻辑而不是某个特定网站的具体实现。如果网页选择器发生变化请以你实际页面为准。6.1 示例一Workbuddy 生成的流程定义先让 Workbuddy 输出一个稍复杂的流程里面加入了等待时间和数据导出步骤。为了演示通用性我们将上面的 workflow.json 扩展为包含“失败重试”和“执行结果引用”的版本。{ task_id: daily_order_report_v2, name: 每日订单导出, version: 2.0.0, owner: automation-team, retry: 2, steps: [ { id: step_1, action: open_url, target: https://admin.example.com/login, params: { wait_after: 2 } }, { id: step_2, action: fill_input, target: #username, params: { value: {{ENV.ADMIN_USER}}, description: 输入管理员账号 } }, { id: step_3, action: fill_input, target: #password, params: { value: {{ENV.ADMIN_PASS}}, description: 输入管理员密码 } }, { id: step_4, action: click, target: button[typesubmit], params: { wait_after: 5, description: 点击登录 } }, { id: step_5, action: wait_for_selector, target: #order-table, params: { timeout: 10000, description: 等待订单表格加载 } }, { id: step_6, action: extract_table, target: #order-table, params: { export_path: ./output/orders.csv, description: 导出订单表 } } ] }6.2 示例二通用的 Python 执行器下面的脚本就是对接模型里的“执行引擎”。它不做业务判断只负责按步骤执行。脚本采用 Playwright 的同步 API结构上尽可能保持简单方便你在此基础上扩展。 文件路径: executor/blueprint_executor.py 用途: 读取 workflow.json, 按步骤执行浏览器自动化操作 运行: python blueprint_executor.py --config workflow.json import argparse import json import os import time from datetime import datetime from playwright.sync_api import sync_playwright def resolve_value(raw: str): 把 {{ENV.XXX}} 占位符替换为真实环境变量值 if raw.startswith({{ENV.) and raw.endswith(}}): env_name raw[6:-2] value os.getenv(env_name) if value is None: raise ValueError(f缺少环境变量: {env_name}) return value return raw def run_step(step: dict, page, context): 执行单个步骤 action step[action] target step.get(target, ) params step.get(params, {}) log { step_id: step.get(id), action: action, start_time: datetime.now().isoformat(), status: pending } try: if action open_url: page.goto(target, wait_untildomcontentloaded) wait_after params.get(wait_after, 0) if wait_after: time.sleep(wait_after) elif action fill_input: value resolve_value(params.get(value, )) page.fill(target, value) wait_after params.get(wait_after, 0) if wait_after: time.sleep(wait_after) elif action click: page.click(target) wait_after params.get(wait_after, 0) if wait_after: time.sleep(wait_after) elif action wait_for_selector: timeout params.get(timeout, 5000) page.wait_for_selector(target, timeouttimeout) elif action extract_table: rows page.query_selector_all(f{target} tbody tr) export_path params.get(export_path) if not export_path: raise ValueError(extract_table 缺少 export_path 参数) os.makedirs(os.path.dirname(export_path) or ., exist_okTrue) with open(export_path, w, encodingutf-8) as f: for row in rows: cells row.query_selector_all(td) line ,.join(cell.inner_text().strip() for cell in cells) f.write(line \n) print(f已导出 {len(rows)} 行数据到 {export_path}) else: raise ValueError(f不支持的 action: {action}) log[status] success except Exception as exc: log[status] failed log[error] str(exc) raise RuntimeError(json.dumps(log, ensure_asciiFalse)) from exc finally: log[end_time] datetime.now().isoformat() context[logs].append(log) print(json.dumps(log, ensure_asciiFalse)) def execute_workflow(config_path: str): with open(config_path, r, encodingutf-8) as f: workflow json.load(f) task_id workflow.get(task_id, unknown) retry workflow.get(retry, 0) context {logs: []} print(f开始执行任务: {task_id}) with sync_playwright() as p: browser p.chromium.launch(headlessFalse) page browser.new_page() try: for step in workflow[steps]: for attempt in range(retry 1): try: run_step(step, page, context) break except RuntimeError: if attempt retry: raise print(f步骤 {step[id]} 失败, 准备第 {attempt 2} 次重试) time.sleep(2) print(全部步骤执行完成) finally: browser.close() with open(workflow_result.json, w, encodingutf-8) as f: json.dump({task_id: task_id, logs: context[logs]}, f, ensure_asciiFalse, indent2) if __name__ __main__: parser argparse.ArgumentParser(description执行 Workbuddy 生成的 RPA 流程定义) parser.add_argument(--config, defaultworkflow.json, help流程定义文件路径) args parser.parse_args() execute_workflow(args.config)这段代码里有几个设计值得记住resolve_value负责环境变量替换避免在流程定义里出现明文密码。run_step里每个操作都写结构化日志日志直接打印成 JSON方便后续采集。整个 workflow 设置了一个retry字段失败后自动重试但重试次数由流程定义决定而不是执行器硬编码。无论成功失败最终都会生成workflow_result.json把每一步的状态回传给 Workbuddy。6.3 示例三对接适配器如果你的蓝印 RPA 支持 HTTP 接口那么对接脚本可以更简单。这里展示一个适配器思路通过 HTTP 提交 workflow.json 给蓝印然后查询任务状态。 文件路径: executor/blueprint_adapter.py 用途: 把 workflow.json 提交给蓝印 RPA 接口, 并轮询任务状态 注意: 接口地址仅作示例, 以蓝印实际提供的接口为准 import json import time import requests # 这些配置从环境变量读取, 不要写死 BLUEPRINT_API os.getenv(BLUEPRINT_API, http://localhost:8080/api/tasks) API_TOKEN os.getenv(BLUEPRINT_TOKEN, ) def submit_workflow(config_path: str): with open(config_path, r, encodingutf-8) as f: payload json.load(f) headers {Authorization: fBearer {API_TOKEN}} resp requests.post(BLUEPRINT_API, jsonpayload, headersheaders, timeout30) resp.raise_for_status() task_id resp.json().get(task_id) print(f任务已提交, task_id{task_id}) # 轮询任务状态 for _ in range(30): status_resp requests.get( f{BLUEPRINT_API}/{task_id}, headersheaders, timeout30 ) status_resp.raise_for_status() state status_resp.json().get(state) print(f当前任务状态: {state}) if state in (success, failed): return status_resp.json() time.sleep(5) raise TimeoutError(任务执行超时) if __name__ __main__: submit_workflow(workflow.json)如果你的蓝印环境不支持 HTTP API也不要直接放弃这个思路。你完全可以在本地把 workflow.json 中的每个 action 映射成蓝印对应的组件调用本质是一样的蓝印只是执行器流程定义才是核心资产。6.4 示例四环境变量文件示例敏感信息统一放在.env文件里并在 Git 中忽略它。# 文件路径: .env ADMIN_USERadminexample.com ADMIN_PASSyour-secure-password BLUEPRINT_APIhttp://localhost:8080/api/tasks BLUEPRINT_TOKENyour-api-token运行前用set -a; source .env; set a加载Linux/macOS或者用dotenv库在 Python 里加载。总之不要让账号密码出现在 JSON 文件里。7. 运行结果与效果验证7.1 如何运行把前面几个文件准备好后目录结构大致是这样的project/ ├── workflow.json ├── .env ├── executor/ │ ├── blueprint_executor.py │ └── blueprint_adapter.py └── output/ └── orders.csv执行命令python executor/blueprint_executor.py --config workflow.json预期输出类似这样开始执行任务: daily_order_report_v2 {step_id: step_1, action: open_url, start_time: 2025-01-01T09:00:01, status: success, end_time: 2025-01-01T09:00:03} {step_id: step_2, action: fill_input, start_time: 2025-01-01T09:00:03, status: success, end_time: 2025-01-01T09:00:03} ... 已导出 37 行数据到 ./output/orders.csv 全部步骤执行完成7.2 如何判断成功不要只看“全部步骤执行完成”就认为成功了。建议做两层验证第一层是执行日志。检查每个 step 的状态是否都是 success有没有触发重试。触发重试本身不代表失败但如果某个步骤频繁重试才成功说明选择器或网络条件不稳定需要优化。第二层是数据验证。导出的orders.csv是否为空行数是否符合预期关键字段是否有值这一步可以在 Workbuddy 的下一轮任务里加上数据质量检查让 AI 自动判断结果是否合理。7.3 失败时先看哪里失败后的排查顺序建议这样安排先看workflow_result.json中最后一个 status 为 failed 的步骤。看该步骤的 error 信息。如果是 element not found去页面确认选择器是否失效。看浏览器是否被检测到自动化特征。有些网站会拦截 Playwright 的自动化访问这种情况下你可能需要增加 stealth 配置或改用真实的浏览器环境。确认网络环境稳定。RPA 流程在夜间执行时网络抖动、系统弹窗、资源加载慢都可能导致失败。8. 常见问题与排查思路下面这个表格整理了对接方案里比较容易踩的坑。问题现象可能原因排查方式解决方案Workbuddy 生成的 JSON 格式错误提示词没有给出严格的 schema 示例在 Workbuddy 的 Skill 里固化 JSON schema 并要求校验后输出加上校验流程, 格式错误直接让 AI 重新生成浏览器打开但页面空白Playwright 未安装对应浏览器内核检查 playwright install 是否执行成功执行playwright install chromium登录失败, 页面提示密码错误workflow.json 里的凭据来自环境变量但未加载检查 .env 是否被正确加载运行时打印环境变量是否存在, 但不要打印值选择器变化导致点击失败前端重构改了元素属性打开 DevTools 查看当前元素实际属性改用更稳定的>
返回列表