影刀RPA店群自动化教程Python驱动规则引擎与智能运营决策实战自动化不只是代替点击而是代替判断。拼多多店群自动化上架方案当系统能够根据实时数据自主决定“要不要调价”“要不要补货”“要不要报名活动”运营才真正被解放。店群运营做久了会发现一个规律很多操作是“条件反射”式的。如果竞品降价了我就跟着降如果库存低于警戒线我就补货如果差评率超过阈值我就暂停投放。这些决策逻辑简单清晰但过去都需要运营每天盯着数据看板手动判断和执行。我们后来把这类“条件→动作”的决策抽象成了一个规则引擎让系统自动完成从感知到执行的闭环。这篇文章就展开这套规则引擎的架构设计、DSL定义、模式匹配以及与影刀RPA执行层的协同方式。TEMU店群如何管理运营一、规则引擎的核心模型条件 动作 规则规则引擎的本质是当某些条件成立时自动触发一系列动作。一个规则由三部分组成触发条件基于实时数据的逻辑判断支持比较、范围、多条件组合动作列表满足条件后要执行的操作可以是启动一个影刀流程、调整一个参数、发送一条通知元数据规则名称、优先级、冷却时间、生效范围等fromdataclassesimportdataclass,fieldfromenumimportEnumfromtypingimportAny,OptionalclassOperator(Enum):EQUALeqNOT_EQUALneqGREATER_THANgtLESS_THANltIN_RANGEin_rangeCONTAINScontainsdataclassclassCondition:data_source:str# 数据来源如 product.price order.refund_rateoperator:Operator value:Any cooldown_minutes:int0# 条件触发后冷却时间避免反复触发dataclassclassAction:action_type:str# start_flow / update_config / send_alertparams:dictfield(default_factorydict)dataclassclassRule:rule_id:strname:strdescription:strpriority:int0shop_filter:str*# 适用店铺范围表达式conditions:listfield(default_factorylist)# 支持 AND/OR 组合actions:listfield(default_factorylist)enabled:boolTruecooldown_seconds:int0# 规则级冷却last_triggered_at:float0.0 运营可以通过管理后台配置规则例如-规则名竞品降价跟进--条件同款竞品最低价我的售价*0.95--动作启动调价流程降低售价至竞品最低价-1元--冷却1小时内不重复触发---## 二、规则DSL设计让运营也能配置面向运营的规则配置不能写代码。我们设计了一套简单的JSON DSL支持嵌套逻辑。 json{rule_id:price_follow_001,name:竞品降价跟进,shop_filter:platformpdd AND categoryfashion,conditions:{logic:AND,children:[{source:competitor.lowest_price,op:lt,value:{{ my_price * 0.95 }}},{source:product.sales_rank,op:gt,value:50},{source:product.stock,op:gt,value:10}]},actions:[{type:start_flow,params:{flow_name:pdd_update_price,target_price:{{ competitor.lowest_price - 1 }}}},{type:send_alert,params:{channel:wechat,message:已自动调价}}],cooldown_seconds:3600} Python解析器将JSON转换为规则对象并处理{{}}变量表达式。 pythonimportreimportastclassRuleParser:EXPR_PATTERNre.compile(r\{\{\s*(.?)\s*\}\})defparse(self,rule_json:dict)-Rule:conditionsself._parse_conditions(rule_json[conditions])actionsself._parse_actions(rule_json[actions])returnRule(rule_idrule_json[rule_id],namerule_json[name],shop_filterrule_json.get(shop_filter,*),conditionsconditions,actionsactions,cooldown_secondsrule_json.get(cooldown_seconds,0),)def_parse_conditions(self,cond_json):iflogicincond_json:return{logic:cond_json[logic],children:[self._parse_conditions(c)forcincond_json[children]]}else:returnCondition(data_sourcecond_json[source],operatorOperator(cond_json[op]),valuecond_json[value],cooldown_minutescond_json.get(cooldown_minutes,0)) 变量表达式会在评估时被替换为实时数据。比如{{my_price*0.95}}会从数据上下文获取my_price计算后作为比较阈值。---## 三、规则评估引擎如何高效匹配规则店铺数量多、规则多时逐条评估效率很低。我们采用基于数据源的索引来加速。 当任何数据源更新时如新采集的竞品价格写入Redis触发只评估依赖该数据源的规则而不是全量扫描。 pythonclassRuleEngine:def__init__(self,rule_store,data_context_factory,action_dispatcher):self.storerule_store# 规则仓库self.data_factorydata_context_factory# 根据shop_id生成数据上下文self.dispatcheraction_dispatcher# 动作分发器self.source_index:Dict[str,set]self._build_index()def_build_index(self):indexdefaultdict(set)forruleinself.store.get_all():sourcesself._extract_sources(rule.conditions)forsrcinsources:index[src].add(rule.rule_id)returnindexdef_extract_sources(self,conditions):sourcesset()ifisinstance(conditions,dict)andlogicinconditions:forchildinconditions[children]:sources|self._extract_sources(child)elifisinstance(conditions,Condition):sources.add(conditions.data_source)returnsourcesasyncdefevaluate_for_shop(self,shop_id:str,changed_source:strNone):data_contextawaitself.data_factory.build(shop_id)candidate_rulesself._get_candidate_rules(shop_id,changed_source)forruleincandidate_rules:ifnotself._match_shop_filter(rule,shop_id,data_context):continueifself._is_in_cooldown(rule):continueifself._evaluate_conditions(rule.conditions,data_context):awaitself._trigger_actions(rule,shop_id,data_context)def_evaluate_conditions(self,conditions,data_context)-bool:ifisinstance(conditions,dict)andlogicinconditions:results[self._evaluate_conditions(c,data_context)forcinconditions[children]]returnall(results)ifconditions[logic]ANDelseany(results)elifisinstance(conditions,Condition):actual_valueself._resolve_source(conditions.data_source,data_context)expectedself._resolve_expression(conditions.value,data_context)returnself._compare(actual_value,conditions.operator,expected)def_resolve_source(self,source:str,context:dict):# 支持点号访问嵌套数据partssource.split(.)valuecontextforpinparts:valuevalue.get(p,{})ifisinstance(value,dict)elsegetattr(value,p,None)returnvaluedef_compare(self,actual,op:Operator,expected)-bool:ifopOperator.EQUAL:returnactualexpectedelifopOperator.GREATER_THAN:returnfloat(actual)float(expected)elifopOperator.LESS_THAN:returnfloat(actual)float(expected)elifopOperator.IN_RANGE:returnexpected[0]actualexpected[1]elifopOperator.CONTAINS:returnexpectedinstr(actual)returnFalseasyncdef_trigger_actions(self,rule:Rule,shop_id:str,context:dict):rule.last_triggered_attime.time()foractioninrule.actions:resolved_paramsself._resolve_expression(action.params,context)awaitself.dispatcher.dispatch(action.action_type,shop_id,resolved_params)awaitself.store.record_trigger(rule.rule_id,shop_id) 这样当数据管道写入新的竞品价格时只需调用 evaluate_for_shop(shop_id,changed_sourcecompetitor.lowest_price)引擎只评估依赖此数据源的规则。---## 四、规则冷却机制防止决策震荡有些场景下规则触发后短时间内条件仍然满足如果不加限制会反复执行。 比如调价后竞品可能会再次降价导致循环调价。 冷却机制分三级-**规则级冷却**整条规则触发后N秒内不再触发--**条件级冷却**某个具体条件项如特定商品的价格比较有独立的冷却时间--**动作级去重**相同动作参数在短时间内不重复执行 pythondef_is_in_cooldown(self,rule:Rule)-bool:ifrule.cooldown_seconds0:returnFalsereturn(time.time()-rule.last_triggered_at)rule.cooldown_seconds 冷却时间在配置中灵活设定运营可根据业务敏感度调整。---## 五、规则动作与影刀流程的无缝衔接规则触发的动作最终需要影刀RPA来执行。我们定义了标准的动作接口 pythonclassActionDispatcher:asyncdefdispatch(self,action_type:str,shop_id:str,params:dict):ifaction_typestart_flow:awaitself.task_orchestrator.submit_event_task({shop_id:shop_id,flow_name:params[flow_name],params:params.get(flow_params,{}),priority:7,# 规则触发优先级较高trigger_source:rule_engine})elifaction_typeupdate_config:awaitself.config_service.update_shop_config(shop_id,params[config_key],params[value])elifaction_typesend_alert:awaitself.alert_service.send(params[channel],params[message])elifaction_typepause_shop:awaitself.scheduler.pause_shop_tasks(shop_id,durationparams.get(duration_seconds)) 所有动作通过统一分发确保审计和权限一致。---## 六、规则应用场景举例**自动调价**采集到竞品最低价低于我的售价5%且我的销量排名在前50自动调价并记录。**自动补货**库存低于安全线且近7天日均销量高于阈值自动生成采购计划或触发进货流程。**差评预警**店铺差评率在1小时内上升超过2%自动暂停自动回复流程通知人工介入。**活动报名**检测到平台大促报名入口开放且商品满足活动要求自动报名。 这些规则都配置在规则库中引擎每30秒遍历一次实时响应。---## 七、规则效果追踪与复盘每条规则触发后会记录完整的触发快照包括当时的数据上下文和触发结果。 运营可以按周查看规则执行报表触发了多少次、带来了多少效果如调价后销量变化、是否存在误触发。 基于复盘数据运营可以持续优化规则条件提高决策精准度。 pythonclassRuleAnalytics:asyncdefcalculate_impact(self,rule_id:str,shop_id:str,window_hours:int24):triggersawaitdb.fetch(SELECT * FROM rule_triggers WHERE rule_id$1 AND shop_id$2 AND triggered_at NOW() - $3::interval,rule_id,shop_id,f{window_hours}hours)fortriggerintriggers:# 对比触发前后的指标变化pre_metricsawaitself.get_metrics(shop_id,beforetrigger[triggered_at])post_metricsawaitself.get_metrics(shop_id,aftertrigger[triggered_at])trigger[impact]self._diff_metrics(pre_metrics,post_metrics)returntriggers ---## 八、与现有系统的关系规则引擎不替代调度引擎和编排引擎而是作为它们上游的“决策大脑”。 调度引擎负责任务何时执行编排引擎负责如何执行规则引擎负责决定做什么。 三者松耦合通过标准化的消息格式协作。---## 九、踩坑记录**规则爆炸**早期没有冷却机制一条“竞品降价跟进”规则在一小时内触发了40次调价被平台判定异常。加入冷却后平稳。**数据延迟导致误判**竞品价格采集有延迟规则触发时使用的可能是5分钟前的旧数据。我们为数据源添加了时效性标注过期数据不参与规则评估。**规则配置错误传播**运营配置错误将gt大于写成了lt小于导致所有商品都被调成最高价。我们在规则保存时加入了模拟测试功能用历史数据跑一遍规则展示会触发多少次让运营确认后再发布。---## 十、写在最后让机器学会“判断”是自动化进化的下一站。 规则引擎把运营的经验和策略固化为可配置、可追踪、可优化的决策逻辑让系统从“自动化操作工”升级为“自动化运营助手”。一个好的决策引擎不会取代运营而是让他们更少地充当“传感器”和“执行器”更多地扮演“策略制定者”的角色。---*作者林焱*