单轮问答型机器人的天花板很低——用户问一句答一句,能力止步于"信息查询"。真正有业务价值的是让AI完成一个完整任务:"帮我把昨天咨询价格的客户整理成名单、打上高意向标签、每人发一条跟进消息"。这类任务跨越多次工具调用、耗时几分钟到几十分钟、中途可能需要人确认。从机器人到AI自动执行,跨越的是工作流编排这道工程门槛。
一、任务拆解——一句话需求变成可执行计划
用户给的是模糊目标,AI要先把它拆成有序的步骤清单再执行。这个"先规划后执行"(Plan-and-Execute)的模式和单轮决策的区别是:单轮只选一个工具,工作流要生成一个有依赖关系的任务DAG——先查昨天的咨询消息,从中筛出问价格的客户,给这批客户打标签,最后逐个发消息,四步串行且后一步依赖前一步的数据。
计划生成后不是立刻执行,先做"计划审查":校验步骤是否完整(发消息前必须先确定名单)、步骤是否越权(计划里出现了删除联系人这类高危操作直接驳回)、步骤是否可并行(打标签和查客户详情没有依赖关系可以并发提速)。审查通过的计划持久化到数据库,每一步的状态(待执行/成功/失败/跳过)独立记录。
二、人机协同断点——流程在关键处交给人
全自动执行只适用于低风险任务。涉及客户触达、数据修改、费用产生的步骤必须设置人工断点:AI执行到"发跟进消息"这一步时暂停,把生成好的消息文案和目标名单推给运营,运营审核批准后AI才批量发出;运营也可以修改文案、剔除部分客户再放行。
人机协同不是简单的"暂停-继续",要支持三种交接模式:批准式(人只点同意/拒绝)、修改式(人调整AI的方案后执行)、接管式(人发现情况复杂直接自己接手,AI退出)。接管时AI要把已完成的步骤、收集到的数据、下一步建议打包成交接摘要,人接手后不用从头了解情况。等待人工超过24小时未处理的任务自动标记过期,避免僵尸任务堆积。
三、执行回放与审计——每个动作可追溯可重跑
AI自动执行的动作影响真实客户,必须全程可审计。任务执行引擎记录完整轨迹:计划内容、每一步的输入输出、调用了哪个接口、耗时多少、成功还是失败。轨迹不只是日志,要支持"回放"——事后可以把整个任务从第一步逐步重演,看清AI当时为什么这么决策。
回放能力在两个场景下价值最大:任务出问题时定责(是AI规划错了、工具返回的数据错了、还是人审批时没看出来),以及任务优化时复盘(同一个目标换一种拆解方式会不会更好)。失败的步骤要支持"从断点重跑"——修复问题后不用整个任务重来,从失败的那一步继续,已成功的步骤不重复执行,靠幂等键保证重跑安全。
三个核心机制对照
机制 | 解决的问题 | 设计要点 |
|---|---|---|
任务拆解 | 模糊目标无法直接执行 | DAG计划+完整性越权审查 |
人机断点 | 高危动作不能全自动 | 批准/修改/接管三模式+交接摘要 |
回放审计 | 出问题无法定责 | 全轨迹记录+断点重跑+幂等 |
智能工作流引擎实现
class WorkflowEngine: def submit(self, wxid, goal): # 1. 规划:目标→步骤DAG plan = llm_plan( goal=goal, tools=WX_TOOLS, # 微信能力工具集 system="把目标拆成最小步骤,标注依赖关系," "涉及发消息/改数据的步骤标记need_approval") if not self.review(plan): # 计划审查 return "计划未通过安全审查" task = db.save_task({"goal": goal, "plan": plan, "status": "planned", "steps": []}) self.run(task) def run(self, task): for step in task.plan.ready_steps(): # 按依赖顺序 if step.need_approval: self.handoff_to_human(task, step) # 人工断点 return # 挂起,等审批回调 result = self.execute_step(task, step) if not result.ok: task.pause_at(step) # 失败挂起等修复 alert(f"任务{task.id}在步骤{step.name}失败") return def handoff_to_human(self, task, step): handoff = {"task_id": task.id, "step": step.name, "draft": step.draft, # AI生成的文案/方案 "context": task.summary(), # 已完成步骤摘要 "modes": ["approve", "edit", "takeover"]} push_to_ops(handoff) task.status = "awaiting_human" def on_human_decision(self, task_id, mode, payload): task = db.get_task(task_id) if mode == "approve": self.execute_step(task, task.current_step, payload) elif mode == "edit": self.execute_step(task, task.current_step, payload.edited) elif mode == "takeover": task.status = "human_taken_over" # AI退出 def replay(self, task_id): """逐步重演任务轨迹供审计""" return [{"step": s.name, "input": s.input, "output": s.output, "decision": s.why} for s in db.get_task(task_id).trace]落地建议
智能工作流从"只读型任务"起步——"整理名单、生成报告"这类不触发客户触达的任务可以全自动,先验证规划准确率。人工断点先做批准式一种模式,修改式和接管式等运营真的有需求再补。执行轨迹从第一天就要完整记录,它是团队敢不敢把更多动作交给AI的信任基础。微信侧的消息收发、标签、群管理等动作能力由 Eyun 平台 这类个人微信API平台提供,工作流引擎在自建服务中负责任务编排和状态管理,接口参数和回调字段说明参考 Eyun 开发文档。