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

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

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

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

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

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

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

三、状态跟踪与人工介入

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

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

自动任务执行能力对照

自动任务执行引擎实现

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平台 提供，任务定义、执行引擎和状态跟踪在自建服务实现，接口字段以平台开发文档为准。