自动任务执行是机器人从"能回复"走向"能办事"的关键。用户说"帮我订明天的会议室"，机器人要能拆解成查空闲、预定、发确认三步并执行完。任务执行要解决任务定义、执行引擎、状态跟踪、人工介入四个问题。

一、任务定义——目标拆成可执行的步骤

任务不能只写成一个函数，要结构化定义。一个任务包含：目标（用户要什么）、步骤（达成目标的动作序列）、参数（每步需要的输入）、约束（时间限制、权限范围）。

结构化的价值在于可复用和可追踪。任务模板化之后，"查订单"这个任务对任何用户都是同一套步骤，只是参数不同。执行时可以逐步记录，出问题能定位到哪一步。

二、执行引擎——顺序、并行、条件分支

步骤之间的关系有三种。顺序——上一步结果作为下一步输入，必须按序。并行——互不依赖的步骤同时执行，节省时间。条件分支——根据上一步结果决定下一步走哪条路（查到订单就通知，查不到就走异常处理）。

执行引擎要支持这三种模式，同时处理超时和失败。每个步骤设超时，超时的步骤判失败。失败后按策略处理——重试、跳过、还是终止任务。终止任务时已执行的步骤要不要回滚，按步骤的副作用决定。

三、状态跟踪与人工介入

任务执行过程要落状态。每步执行前记"进行中"，执行后记"成功"或"失败"。状态落库的价值有两个：任务中断后能恢复，以及能复盘任务为什么失败。

人工介入要设计好介入点。不是所有失败都转人工——网络超时重试就行，参数错误转人工让用户补，执行了不可逆动作后失败必须人工确认。介入点按失败类型和步骤副作用定，不要一刀切全转人工。

自动任务执行能力对照

自动任务执行引擎实现

class TaskExecutor:
    def __init__(self):
        self.state_db = TaskStateDB()

    def execute(self, task, params):
        # 执行任务：按步骤模式调度
        task_id = task.new_id()
        self.state_db.create(task_id, task.name)
        context = dict(params)

        for step in task.steps:
            self.state_db.update(task_id, step.name, "running")
            if step.mode == "parallel":
                results = self.run_parallel(step.deps, context)
                context.update(results)
            elif step.mode == "condition":
                branch = step.condition(context)
                context.update(self.run_step(branch, context))
            else:
                context.update(self.run_step(step, context))
            self.state_db.update(task_id, step.name, "success")
        self.state_db.finish(task_id)
        return context

    def run_step(self, step, context):
        # 单步执行：超时+失败处理
        try:
            with timeout(step.timeout):
                return step.action(context)
        except TimeoutError:
            if step.on_fail == "retry":
                return self.retry_step(step, context)
            elif step.on_fail == "human":
                self.to_human(step, context)     # 转人工
            raise
        except Exception as e:
            if step.reversible:
                step.undo(context)               # 可逆则回滚
            else:
                self.to_human(step, context)     # 不可逆转人工
            raise

    def run_parallel(self, steps, context):
        # 并行执行：同时跑无依赖步骤
        results = {}
        with ThreadPoolExecutor() as ex:
            futures = {
                ex.submit(step.action, context): step.name
                for step in steps
            }
            for future in as_completed(futures):
                results[futures[future]] = future.result()
        return results

    def resume(self, task_id):
        # 断点恢复：从失败步骤继续
        state = self.state_db.get(task_id)
        failed_step = state.first_failed()
        task = self.load_task(state.task_name)
        return self.execute_from(task, failed_step, state.context)

落地建议

任务要结构化定义——目标、步骤、参数、约束分开，模板化后能复用也能追踪。执行引擎支持顺序、并行、条件三种模式——全用顺序会慢，全用并行会乱。状态必须落库——任务中断能恢复，失败能复盘，不落状态等于黑盒执行。人工介入点按类型定——网络超时重试、参数错误让用户补、不可逆动作失败必须人工确认。微信侧的消息收发由 Eyun 这类个人微信API平台 提供，任务定义、执行引擎和状态跟踪在自建服务实现，接口字段以平台开发文档为准。