微信机器人处理简单指令够了——用户说"查物流",查一条回一条。处理复杂需求就不行——"帮我把昨天的订单都核对一遍,有问题的标出来,然后生成报告发给我",背后是查订单、逐单核对、标记问题、生成报告、发送用户多个子任务。复杂需求不是字数多的需求,是不能一步完成、需要拆成子任务按依赖关系执行的需求。智能任务拆解要解决三个问题:要不要拆、拆多细、按什么顺序执行。

一、复杂度识别——先判断要不要拆

不是所有请求都需要拆解。用户说"查物流"直接查,不需要拆。复杂度识别是任务拆解的第一道关卡——判断是简单指令还是复杂需求,简单的直接执行,复杂的才进入拆解流程。判断维度有三个:是否涉及多个对象(一个订单 vs 全部订单)、是否需要多步骤(查+核对+标记+报告)、是否有依赖关系(标记依赖核对结果)。

复杂度识别的价值是避免过度拆解。简单请求强行拆成子任务,拆解开销超过执行开销——查物流拆成"解析单号+查询+组装"三个子任务,除了增加调度没有意义。复杂度阈值要可配置,按业务场景调。消息解析和复杂度判断的接口在 Eyun 开发文档 中有对应支撑。

二、拆解粒度控制——拆多细是个工程选择

确认要拆后,拆多细是关键选择。拆太粗——"核对全部订单"拆成一个子任务,执行时还是什么都干,等于没拆。拆太细——"核对全部订单"拆成"取列表+取每个详情+核对每个+收集结果",10个订单拆出40个子任务,调度开销爆炸。合理的粒度是按业务动作拆——一个不可再分的业务动作是一个子任务,"核对单个订单"是一个子任务,10个订单是10个子任务实例。

粒度控制的原则是业务动作不可再分。"查昨日订单"是一个动作,"核对单个订单"是一个动作,"生成报告"是一个动作。按业务动作拆而非按技术步骤拆,能保证子任务数量与业务规模成正比,不会因输入量大而拆解层数暴增。拆解粒度的控制策略和任务调度的接口在 Eyun 开发文档 中有对应能力。

三、子任务依赖排序——执行顺序由依赖关系决定

子任务拆出来后,执行顺序不是按拆解顺序而是按依赖关系。核对订单依赖查询结果,生成报告依赖核对结果,发送报告依赖报告生成完成。依赖排序的价值是保证子任务在依赖项完成后才执行——没查到订单就开始核对,核对的是空数据。

依赖排序的实现是DAG(有向无环图)——子任务是节点,依赖关系是边,拓扑排序后按序执行。DAG中无依赖关系的子任务可以并行——10个订单的核对子任务互不依赖,可以并行执行。依赖排序要注意环检测——循环依赖会导致执行死锁,拆解时就要检测并报错。子任务执行和结果汇总的接口在 Eyun 开发文档 中有接口能力。

任务拆解三环节对照

环节

判断维度

控制原则

输出

复杂度识别

对象数+步骤数+依赖

阈值可配置

拆/不拆

拆解粒度

业务动作不可再分

按动作非按步骤

子任务列表

依赖排序

子任务间依赖关系

DAG拓扑排序

执行顺序

任务拆解与依赖排序实现

class TaskDecomposer:
    COMPLEXITY_THRESH = {"multi_obj": 3, "multi_step": 2}

    def decompose(self, request):
        if not self.is_complex(request):        # 复杂度识别
            return [request]                      # 简单指令不拆
        subtasks = self.split_by_action(request)  # 按业务动作拆
        return self.topo_sort(subtasks)           # DAG依赖排序

    def is_complex(self, request):
        # 判断:对象数+步骤数同时超阈值
        return (request.object_count() >= self.COMPLEXITY_THRESH["multi_obj"]
                and request.action_count() >= self.COMPLEXITY_THRESH["multi_step"])

    def split_by_action(self, request):
        actions = self.extract_actions(request)
        tasks = []
        for act in actions:
            if act.type == "check_batch":
                # 批量动作按对象拆成子实例
                tasks.extend(self.expand_batch(act))
            else:
                tasks.append(Task(act))
        return tasks

    def topo_sort(self, tasks):
        graph = self.build_graph(tasks)          # 建DAG
        if self.has_cycle(graph):                # 环检测
            raise DecomposeError("循环依赖")
        return graph.topo_order()

落地建议

复杂度识别阈值别拍脑袋定——先观察真实用户请求,统计简单和复杂的分布,再定阈值。拆解粒度按业务动作拆而非按技术步骤拆,保证子任务数量与业务规模成正比。依赖排序必须做环检测——循环依赖导致执行死锁,拆解时检测比运行时报错好排查。微信侧的消息解析、子任务执行和结果汇总由 Eyun 这类个人微信API平台 提供,复杂度识别和依赖排序在自建服务实现,接口字段以平台开发文档为准。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐