一、为什么RPA需要设计模式在回答这个问题前我们先看一个典型的复杂RPA场景企业财务自动化需要从多个系统获取数据ERP、CRM、银行经过清洗、验证、转换然后生成报表并上传至OA系统期间涉及多次人工审批、异常处理、重试和日志记录。如果采用传统脚本方式所有逻辑堆砌在单个函数中上千行代码难以阅读理解增加一个新数据源需要修改核心流程极易引入bug状态变更散落在各处无法保证一致性无法复用相同的数据处理逻辑。而设计模式正是为了解决这类问题而生。它们将变化的部分封装将稳定的部分抽象让代码结构清晰变更局部化。下面我们将逐一深入三种模式。二、流程控制模式让流程灵活可控2.1 模式定义流程控制模式并不是单一的设计模式而是一组用于组织业务流程执行逻辑的模式组合常见的有模板方法模式Template Method定义流程的骨架将可变步骤延迟到子类实现。策略模式Strategy封装不同的业务规则使它们可以互相替换。责任链模式Chain of Responsibility为请求创建处理链每个处理者决定是否处理或传递给下一个。在RPA中一个完整业务流程往往由多个步骤组成步骤可能因客户、环境、数据类型而异。流程控制模式可以帮助我们将步骤的“顺序结构”与“具体实现”分离。2.2 模板方法模式实现流程骨架假设我们有一个通用的数据导入流程读取数据 → 清洗 → 验证 → 转换 → 写入目标系统。不同场景下清洗、验证、转换的具体规则可能不同但流程顺序固定。fromabcimportABC,abstractmethodclassDataImportPipeline(ABC):模板类定义流程骨架defrun(self,source):模板方法定义了算法骨架dataself.extract(source)cleanedself.clean(data)validatedself.validate(cleaned)transformedself.transform(validated)self.load(transformed)defextract(self,source):默认实现子类可覆盖print(f从{source}提取原始数据)return{raw:source}abstractmethoddefclean(self,data):子类必须实现清洗逻辑passabstractmethoddefvalidate(self,data):子类必须实现验证逻辑passabstractmethoddeftransform(self,data):子类必须实现转换逻辑passdefload(self,data):默认实现子类可覆盖print(f加载数据到目标系统:{data})classErpInvoicePipeline(DataImportPipeline):ERP发票导入流程defclean(self,data):print(清洗ERP发票数据去空格、格式统一)# 具体清洗代码returndatadefvalidate(self,data):print(验证发票必需字段)# 验证逻辑returndatadeftransform(self,data):print(转换为财务系统格式)return{finance:data}优点流程结构稳定复用性强新增一种数据源只需实现新的子类不影响已有流程符合开闭原则。2.3 策略模式实现可插拔的步骤有时流程的某一步有多种算法可选如不同的数据加密方式、不同的文件解析器使用策略模式可以避免大量的if-else。fromtypingimportProtocolclassValidationStrategy(Protocol):defvalidate(self,data):...classStrictValidation:defvalidate(self,data):print(严格校验所有字段必填)returnall(data.values())classLooseValidation:defvalidate(self,data):print(宽松校验只检查关键字段)returndata.get(id)isnotNoneclassDataProcessor:def__init__(self,validation_strategy:ValidationStrategy):self._strategyvalidation_strategydefprocess(self,data):ifnotself._strategy.validate(data):raiseValueError(数据校验失败)# 继续处理print(数据处理成功)# 使用processorDataProcessor(StrictValidation())processor.process({id:1,name:test})# 通过2.4 责任链模式实现多级审批流在RPA中常遇到需要多级审批的业务如采购订单部门主管 → 财务 → 总经理。责任链模式让每个审批者独立易于添加或调整顺序。fromabcimportABC,abstractmethodclassApprover(ABC):def__init__(self):self._nextNonedefset_next(self,approver):self._nextapproverreturnapproverabstractmethoddefprocess(self,request):passclassManagerApprover(Approver):defprocess(self,request):ifrequest.amount10000:print(f部门经理审批通过金额{request.amount})returnTrueelifself._next:returnself._next.process(request)returnFalseclassFinanceApprover(Approver):defprocess(self,request):ifrequest.amount50000:print(f财务审批通过金额{request.amount})returnTrueelifself._next:returnself._next.process(request)returnFalseclassCeoApprover(Approver):defprocess(self,request):print(fCEO最终审批金额{request.amount})returnTrue# 构建责任链managerManagerApprover()financeFinanceApprover()ceoCeoApprover()manager.set_next(finance).set_next(ceo)# 发起请求classRequest:def__init__(self,amount):self.amountamount manager.process(Request(8000))# 经理审批manager.process(Request(30000))# 财务审批manager.process(Request(80000))# CEO审批流程控制模式的RPA应用场景多系统数据同步流程模板方法根据不同客户配置不同的业务规则策略采购订单审批、工单流转责任链三、数据管道模式打造可复用的数据处理链3.1 模式概述数据管道模式Pipeline Pattern源自Unix管道哲学它将数据处理过程分解为一系列独立的处理单元过滤器每个单元接收输入、处理后输出并传递给下一个单元。这种模式非常适合RPA中常见的数据抽取-转换-加载ETL场景。3.2 核心要素数据源提供原始数据文件、数据库、API过滤器对数据进行操作清洗、转换、验证、增强管道连接过滤器的通道控制数据流动数据汇最终存储或输出3.3 Python实现一个轻量级管道我们可以用装饰器或链式调用来实现管道。fromtypingimportCallable,AnyclassPipeline:通用管道类支持多个处理器def__init__(self):self._handlers[]defadd(self,handler:Callable[[Any],Any]):self._handlers.append(handler)returnself# 支持链式调用defexecute(self,initial_data):datainitial_dataforhandlerinself._handlers:datahandler(data)returndata# 定义各个处理函数defextract_from_csv(file_path):print(f从{file_path}读取CSV)return[{id:1,name:A},{id:2,name:B}]defclean_data(records):print(清洗数据去除空格)forrinrecords:r[name]r[name].strip()returnrecordsdefenrich_with_timestamp(records):print(添加时间戳)fromdatetimeimportdatetime tsdatetime.now().isoformat()forrinrecords:r[processed_at]tsreturnrecordsdefload_to_db(records):print(f加载{len(records)}条记录到数据库)returnTrue# 构建管道pipelinePipeline()pipeline.add(extract_from_csv).add(clean_data).add(enrich_with_timestamp).add(load_to_db)# 执行pipeline.execute(data.csv)输出从data.csv读取CSV 清洗数据去除空格 添加时间戳 加载2条记录到数据库3.4 管道模式的进阶支持异步、条件分支复杂场景下可能需要根据数据内容走不同分支或并行处理。我们可以对管道进行扩展引入条件过滤、异常处理等。classConditionalHandler:def__init__(self,condition,true_handler,false_handlerNone):self.conditioncondition self.true_handlertrue_handler self.false_handlerfalse_handlerdef__call__(self,data):ifself.condition(data):returnself.true_handler(data)elifself.false_handler:returnself.false_handler(data)returndata应用场景从多个数据源抽取并统一格式数据质量校验与修复实时数据流处理如监控日志并触发RPA动作3.5 数据管道模式的优势解耦每个处理器独立开发、测试、维护复用相同处理器可在不同管道中使用可测性每个单元可单独测试可观测性可在管道中插入日志、监控节点四、状态机模式让业务流程状态清晰可控4.1 为什么需要状态机在RPA中很多业务流程具有明确的状态例如订单处理待支付 → 已支付 → 已发货 → 已完成工单审批待提交 → 主管审批 → 经理审批 → 归档数据同步待同步 → 同步中 → 同步成功 / 同步失败 → 重试如果使用简单的变量和if-else管理状态当状态增多、转换条件复杂时代码极易出错。有限状态机FSM将状态、事件、转换规则显式定义确保系统始终处于合法状态。4.2 状态机的核心元素状态State系统可能处于的稳定阶段事件Event触发状态转换的输入转换Transition定义在某个状态下发生某个事件后转移到哪个新状态并执行相应动作动作Action状态转换时执行的业务逻辑4.3 使用Python实现状态机我们可以使用transitions库快速构建也可以手动实现轻量级版本。示例使用transitions库推荐pip install transitionsfromtransitionsimportMachineclassInvoiceOrder:states[pending,paid,shipped,completed,cancelled]def__init__(self,name):self.namename self.machineMachine(modelself,statesInvoiceOrder.states,initialpending)# 定义转换self.machine.add_transition(pay,pending,paid,afterafter_pay)self.machine.add_transition(ship,paid,shipped,afterafter_ship)self.machine.add_transition(complete,shipped,completed,afterafter_complete)self.machine.add_transition(cancel,[pending,paid],cancelled,afterafter_cancel)defafter_pay(self):print(f订单{self.name}已支付记录财务信息)defafter_ship(self):print(f订单{self.name}已发货更新物流)defafter_complete(self):print(f订单{self.name}已完成)defafter_cancel(self):print(f订单{self.name}已取消释放库存)# 使用orderInvoiceOrder(ORD001)order.pay()# 支付order.ship()# 发货order.complete()# 完成order.cancel()# 此时状态已是completed不允许取消会触发异常手动实现轻量级状态机用于理解原理classStateMachine:def__init__(self):self.stateinitialself.transitions{(initial,start):processing,(processing,success):completed,(processing,fail):error,(error,retry):processing,}deftrigger(self,event):key(self.state,event)ifkeyinself.transitions:old_stateself.state self.stateself.transitions[key]self.on_transition(old_state,event)else:raiseException(f非法转换{self.state}-{event})defon_transition(self,old,event):print(f{old}通过{event}转为{self.state})4.4 状态机在RPA中的高级应用场景一个需要与用户交互的RPA任务例如自动填报系统可能被用户暂停、恢复、手动干预。状态机可以清晰定义这些交互状态。classRpaTask:states[idle,running,paused,error,completed]def__init__(self,task_id):self.task_idtask_id self.machineMachine(modelself,statesRpaTask.states,initialidle)self.machine.add_transition(start,idle,running)self.machine.add_transition(pause,running,paused)self.machine.add_transition(resume,paused,running)self.machine.add_transition(fail,running,error)self.machine.add_transition(retry,error,running)self.machine.add_transition(complete,running,completed)defon_enter_running(self):# 启动实际业务流程pass优点状态流转显式化易于理解和测试避免非法状态转换可以轻松添加持久化将状态保存到数据库实现断点续传五、三种模式的协同应用在实际RPA项目中这三种模式往往同时使用互相配合。下面通过一个综合案例展示它们的协同场景一个企业RPA需要定期从多个ERP系统抽取销售数据经过清洗、验证、转换然后通过审批流程最后同步到数据仓库。同时整个过程需要记录状态支持断点续传和重试。流程控制模式使用模板方法定义整体流程骨架抽取→清洗→转换→审批→加载。其中审批环节使用责任链模式实现多级审批。数据管道模式在数据处理阶段构建数据管道将清洗、验证、转换作为独立的过滤器方便复用和测试。状态机模式为每个批次任务维护状态待抽取、抽取中、清洗中、审批中、同步中、完成、失败使用状态机管理确保任务状态转换正确并支持重试。通过这种组合代码结构清晰每个模块职责单一系统整体健壮性大幅提升。