影刀RPA 多流程协作编排:主流程调用子流程与参数传递实战
做自动化一段时间后,你会发现单文件流程根本扛不住复杂的业务。一个流程几千步,改一个小地方就可能整个崩掉。
这时候就需要多流程协作——把一个大流程拆成多个子流程,主流程负责调度,子流程各司其职。影刀原生支持子流程调用,而且参数传递做得很顺手。
这篇文章就讲清楚:什么时候拆流程、怎么调用、参数怎么传、出错怎么办。
什么情况用多流程
先判断要不要拆。不是所有流程都值得拆,以下四种情况是必须拆的:
1. 同一段逻辑在多个主流程里用到。
比如"登录系统"这个动作,A流程采集数据要登录,B流程发报表也要登录。不拆的话,每次改密码你得改两个地方,早晚漏一个。
2. 单文件流程超过500步。
不是说500步就崩,而是超过这个量级,你找bug的时候Ctrl+F都救不了你。拆成子流程后,每个子流程几百步在单独文件里,定位问题快得多。
3. 业务逻辑有明显边界。
比如一个流程做"从淘宝采集订单 → 写入ERP → 发邮件通知"。三个动作之间是松耦合的,天然适合拆成三个子流程。
4. 不同人负责不同环节。
团队协作时,A负责数据采集,B负责数据处理,C负责结果推送。每个人在自己的子流程文件里折腾,主流程只负责串起来。不会互相踩代码。
怎么调用子流程
影刀里调用子流程用的动作叫**【运行子流程】**,在"流程控制"分类下。操作步骤:
- 在主流程的合适位置,拖入**【运行子流程】**动作
- 点击动作配置,选择要调用的
.flow文件 - 设置输入参数和输出参数的映射
基本调用结构
主流程一般长这样:
主流程.flow: ↓ 【运行子流程】— 调用 采集订单.flow(传入:关键词、页码;传出:订单列表) ↓ 【运行子流程】— 调用 写入ERP.flow(传入:订单列表;传出:写入结果) ↓ 【运行子流程】— 调用 发送通知.flow(传入:写入结果;传出:无) ↓ 完成每个子流程是独立的.flow文件,放在同一个项目目录下就行。
拼多多店群自动化上架方案
子流程的入口设置
子流程文件里需要在开头声明输入参数和输出参数。
在子流程编辑器中,点击左侧的"流程参数"面板,添加入参和出参。比如一个"登录系统"的子流程:
- 输入参数:
用户名(字符串)、密码(字符串) - 输出参数:
登录结果(布尔值)、SessionToken(字符串)
主流程调用时,把主流程里的变量映射到子流程的输入参数;子流程执行完后,输出参数会自动写回主流程对应的变量。
参数传递的三种方式
方式一:直接变量映射(推荐)
最常用的方式。主流程里有一个变量account,子流程的输入参数叫用户名,在**【运行子流程】**的配置里直接把account映射给用户名。
主流程变量 account = "admin123" 子流程输入参数 用户名 ← account 子流程执行... 子流程输出参数 登录结果 → 主流程变量 is_login_success影刀的参数映射是值传递不是引用传递——子流程里改了参数值,不会影响主流程的原始变量(除非通过输出参数写回)。
方式二:全局变量中转
在主流程和子流程里都用全局变量。影刀的全局变量是在"变量管理"面板里定义的,所有流程文件都能读写。
主流程: 【设置全局变量】g_username = "admin123" 【运行子流程】调用 子流程.flow 【读取全局变量】result = g_login_result 子流程: 【读取全局变量】username = g_username 执行登录... 【设置全局变量】g_login_result = True全局变量方式灵活,但问题是:你不知道哪个流程在什么时候改了全局变量。调试时变量被莫名其妙覆盖是常事。建议只在传少量简单值的时候用,复杂流程老老实实用参数映射。
方式三:文件/数据库中转(大数据场景)
当需要传递的数据很大(比如几千条订单记录),参数映射和全局变量都扛不住。这时候用文件或数据库做中转:
主流程: 【写入Excel】把订单列表导出到 temp/orders.xlsx 【运行子流程】传入文件路径 temp/orders.xlsx 子流程: 【读取Excel】从传入的文件路径读取订单数据 处理订单... 【写入Excel】把处理结果导出到 temp/result.xlsx 主流程(子流程返回后): 【读取Excel】从 temp/result.xlsx 读取处理结果大数据的场景还是老老实实走文件,不要用变量硬传。
子流程的错误处理
子流程出错怎么办?这是多流程协作里最容易被忽略的问题。子流程崩了,主流程不能跟着崩。
错误传播机制
**【运行子流程】**动作有一个配置项叫"执行失败时继续"。勾上这个选项后,子流程抛出异常不会中断主流程。但代价是你得自己检查子流程是否执行成功。
做法是让子流程永远通过输出参数返回状态:
子流程的输出参数: success: True/False error_message: "" 主流程: 【运行子流程】(失败时继续 = True) 【IF】子流程输出 success == False 【日志输出】"子流程失败:" + error_message 【运行子流程】调用 告警通知.flow 【END IF】子流程超时控制
有些子流程可能挂死——比如网页一直加载不出来。影刀的**【运行子流程】**没有内置超时设置,需要自己在子流程里加。
在子流程的关键步骤前加一个超时检测:
子流程内部: 【记录时间】t_start 执行采集... 【记录时间】t_now 【IF】(t_now - t_start) > 600 秒 【设置输出】success = False, error_message = "子流程超时" 【终止流程】 【END IF】或者更简单的——在子流程结尾加一个总的判断,如果某个关键变量为空,说明中间出了问题没执行完,直接返回失败。
重试机制
子流程失败不一定是真失败,很多是网络波动之类的临时问题。在主流程里加个重试循环:
主流程: retry_count = 0 【WHILE】retry_count < 3 AND 子流程输出 success == False retry_count = retry_count + 1 【日志输出】"第" + retry_count + "次重试..." 【等待】5 秒 【运行子流程】(重新调用) 【END WHILE】 【IF】retry_count >= 3 AND 子流程输出 success == False 【日志输出】"重试3次仍失败,终止主流程" 【终止流程】 【END IF】实战案例:电商订单处理全流程
一个真实的多流程协作案例——从电商平台采集订单、写入ERP、发邮件通知,三个子流程加一个主流程调度。
主流程:订单处理调度
主流程 orchestrator.flow: ┌─────────────────────┐ │ 1. 读取配置 │ │ - 采集日期范围 │ │ - ERP登录信息 │ │ - 通知邮箱列表 │ └──────┬──────────────┘ ↓ ┌─────────────────────┐ │ 2. 运行子流程: │ │ 采集订单.flow │ │ 入参:日期范围 │ │ 出参:success, │ │ orders_data │ └──────┬──────────────┘ ↓ 【IF success == False】 → 记录日志,发告警,终止 ↓ ┌─────────────────────┐ │ 3. 运行子流程: │ │ 写入ERP.flow │ │ 入参:orders_data │ │ 出参:success, │ │ import_count│ └──────┬──────────────┘ ↓ 【IF success == False】 → 记录日志,发告警,终止 ↓ ┌─────────────────────┐ │ 4. 运行子流程: │ │ 发送通知.flow │ │ 入参:import_count│ │ 出参:success │ └──────┬──────────────┘ ↓ ┌─────────────────────┐ │ 5. 记录主流程日志 │ │ 本次处理订单数 │ │ 成功/失败状态 │ └─────────────────────┘子流程1:采集订单
采集订单.flow: 输入参数:start_date, end_date 输出参数:success, orders_data, error_message 步骤: 1. 【打开网页】进入商家后台 2. 【IF】检测登录状态 → 未登录:调用 登录模块(内置在子流程里的小逻辑) 3. 【填写】设置日期筛选条件 4. 【点击】搜索按钮 5. 【等待元素】等待结果表格加载 6. 【获取元素列表】提取订单数据 7. 对每条订单做字段补全(Python节点处理) 8. 设置输出:success = True, orders_data = 订单列表 异常处理(整个子流程包裹在try-catch里): → 任何异常:success = False, error_message = 异常信息子流程2:写入ERP
写入ERP.flow: 输入参数:orders_data 输出参数:success, import_count, error_message 步骤: 1. 【打开网页】ERP系统登录页 2. 【填写】用户名密码 3. 【点击】登录 4. 遍历 orders_data 的每条订单: a. 【点击】新增订单按钮 b. 【填写】客户信息、商品信息、金额等 c. 【点击】保存 d. import_count += 1 5. 设置输出:success = True, import_count = 导入成功数子流程3:发送通知
发送通知.flow: 输入参数:import_count 输出参数:success 步骤: 1. Python节点:拼接邮件内容 2. [video(video-OuLhzuaz-1784019385261)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营?)] subject = f"订单同步完成 - {今日日期}" body = f"本次成功导入 {import_count} 条订单" 3. 【执行SMTP发送】(影刀的邮件动作或Python smtplib) 4. 设置输出:success = True三个最容易踩的坑
坑1:子流程里用了主流程的变量名
子流程和主流程的变量是隔离的。你在子流程里定义了一个叫result的变量,跟主流程里的result完全不是同一个东西。
踩坑现场:主流程里把数据写进orders变量,然后调用子流程。子流程里也有个orders变量(用来存储本地结果)。主流程以为子流程执行完orders就更新了,其实没有——因为变量是隔离的。
正确做法:子流程的数据永远通过输出参数传回来。不要在思想上依赖"同名变量会互相影响"——不会的。
坑2:子流程结束后浏览器窗口被释放
如果你的子流程里打开了浏览器,子流程执行完成后浏览器默认会被关闭(取决于影刀的浏览器管理策略)。后续子流程想用同一个浏览器会话,发现窗口已经没了。
正确做法:
- 设计原则:每个子流程自己打开浏览器,自己关闭浏览器。不自作聪明地"共享"浏览器实例。
- 如果确实需要共享登录态,用**【保存Cookie】和【加载Cookie】**动作,而不是共享浏览器窗口。
坑3:子流程路径硬编码
子流程文件路径写死了绝对路径,换个电脑就跑不了。
错误:C:\Users\张三\Desktop\project\子流程.flow 正确:.\子流程.flow(相对于主流程文件所在目录)影刀的**【运行子流程】**支持相对路径。项目目录结构统一后,整个项目拷到别的电脑上也能直接跑。
另一个常见翻车:子流程文件名包含中文和特殊字符,在某些版本的影刀里路径解析会出问题。建议子流程文件名尽量用英文+数字,如fetch_orders.flow、import_erp.flow。
编排模式总结
最后总结三种常见的编排模式,选哪种看你的业务:
| 模式 | 结构 | 适用场景 | 特点 |
|---|---|---|---|
| 线性编排 | A→B→C | 步骤之间有严格先后顺序 | 简单直接,易调试 |
| 条件编排 | A→(条件)→B或C | 根据子流程返回结果决定下一步 | 加IF判断即可 |
| 并行编排 | A和B同时跑→汇合 | B不依赖A的结果,且互不冲突 | 影刀不支持真并行,但可用多开窗口模拟 |
线性编排占了80%的场景,够用了。条件编排在线性基础上加IF就行。并行编排比较进阶,影刀原生不太好做,真需要的话可以考虑开多个浏览器窗口分别操作。
核心原则就一句话:拆流程是为了让每个文件职责单一、好改好测。拆完以后,参数传递靠映射,错误处理留后路,子流程永远通过输出参数汇报结果。
多流程协作搞好了,复杂业务的自动化就不是噩梦了。
作者:林焱