AutoGLM-Phone任务中断怎么办?断点续传功能实现思路
1. 为什么任务会“突然停下”——AutoGLM-Phone的运行本质
你刚让AI代理打开小红书搜索美食,它正识别首页图标、点击搜索框、输入文字……突然卡住,命令行只留下一行未完成的日志。你重试,它又从头开始:重新截图、重新理解界面、重新规划路径。这不是模型“忘了”,而是整个系统缺乏对“进行中状态”的记忆与恢复能力。
AutoGLM-Phone 的核心工作流是感知—规划—执行闭环:每一步都依赖前序结果。它不是在手机里跑一个常驻App,而是在本地控制端发起一次“会话请求”,通过 ADB 截图获取屏幕图像,上传给云端视觉语言模型(VLM)分析,再接收动作指令(如“点击坐标(320, 680)”),最后调用 ADB 执行。整个过程像一次精密的远程手术——但手术中途断电了,医生不会从切口处继续缝合,而是得重新消毒、铺巾、定位。
这就是任务中断的真实代价:没有状态快照,就没有续传基础。而 Phone Agent 的设计初衷是“轻量、解耦、按需调用”,这恰恰让它天然缺少传统服务端应用的会话持久化机制。它不保存中间截图、不记录已执行动作、不缓存模型推理上下文。一旦网络抖动、ADB超时、或用户误关终端,所有进度清零。
所以,“断点续传”不是加个开关就能启用的功能,它是一次对系统执行范式的重构:从“无状态请求”转向“有状态任务”。
2. 断点续传不是魔法,是三步可落地的设计
要让 AutoGLM-Phone 在中断后能接上最后一句“我刚刚点到了搜索框”,关键不在模型多强,而在如何定义“断点”、如何保存“断点”、如何加载“断点”。我们不追求理论完美,只关注工程师能立刻写进代码的方案。
2.1 第一步:明确定义什么是“可续传的断点”
别一上来就想保存整个内存快照。AutoGLM-Phone 的任务天然具备清晰的阶段粒度。我们把一次完整任务拆解为原子动作序列,每个动作完成后,系统必须确认其可观测、可验证、不可逆的状态变更。例如:
有效断点:
“已成功点击‘搜索’图标(坐标已验证)”
“已输入文字‘美食’并触发软键盘回车”
“页面已跳转至搜索结果页(通过OCR识别标题‘相关笔记’确认)”
❌无效断点:
- “模型已输出动作指令”(指令未执行,不可靠)
- “截图已上传至服务器”(上传成功 ≠ 模型已处理)
- “ADB命令已发送”(未校验返回结果)
换句话说,断点 = 已完成且可被外部验证的动作终点。它不依赖模型内部状态,只依赖设备真实反馈——这才是稳定、可测试、不惧重启的基石。
2.2 第二步:用最朴素的方式持久化断点信息
不需要引入 Redis 或数据库。一个轻量级 JSON 文件,配合时间戳和设备标识,就足够支撑绝大多数场景。我们在Open-AutoGLM项目根目录下新增task_state/目录,每次任务启动时生成唯一任务ID(如task_20240521_142305_abc123),所有状态文件以此命名。
// task_state/task_20240521_142305_abc123.json { "task_id": "task_20240521_142305_abc123", "device_id": "emulator-5554", "start_time": "2024-05-21T14:23:05Z", "last_action": { "step": 3, "action": "tap", "target": "search_icon", "coordinates": [320, 680], "screenshot_hash": "a1b2c3d4e5f67890", "verified_at": "2024-05-21T14:23:22Z" }, "completed_steps": [ {"step": 1, "action": "screenshot", "hash": "x9y8z7w6v5u43210"}, {"step": 2, "action": "tap", "target": "home_tab", "coordinates": [180, 1200]} ], "current_context": { "page_title": "小红书首页", "visible_elements": ["搜索框", "推荐", "关注"] } }这个文件的关键设计在于:
last_action字段:明确记录最后成功执行并验证的动作,是续传的起点;screenshot_hash:用图片哈希值代替存储原始截图,节省空间且防篡改;completed_steps数组:提供完整动作链路,便于调试与审计;current_context:轻量级界面摘要(非全量OCR),供模型快速理解当前状态。
所有写入操作使用atomic write(先写临时文件,再os.replace),避免中断导致文件损坏。
2.3 第三步:让 main.py 主动识别并加载断点
修改main.py的入口逻辑,增加断点检查环节。当用户再次运行命令时,程序自动扫描task_state/目录,查找匹配device_id和相似指令关键词(如包含“小红书”“搜索”)的未完成任务。
# 在 main.py 中新增函数 def find_resumable_task(device_id: str, instruction: str) -> Optional[str]: """查找可续传的任务ID""" state_dir = Path("task_state") if not state_dir.exists(): return None # 粗筛:设备ID匹配 + 指令关键词命中 candidates = [] for f in state_dir.glob("task_*.json"): try: data = json.loads(f.read_text()) if data.get("device_id") == device_id and "小红书" in instruction and "搜索" in instruction: # 检查是否为未完成状态(无 final_result 字段) if "final_result" not in data: candidates.append((f, data.get("last_action", {}).get("verified_at"))) except Exception: continue if candidates: # 返回最新的一次 candidates.sort(key=lambda x: x[1], reverse=True) return str(candidates[0][0]) return None # 在主流程中插入 if __name__ == "__main__": args = parse_args() # 尝试续传 resume_file = find_resumable_task(args.device_id, args.instruction) if resume_file: print(f" 检测到未完成任务,将从 {resume_file} 续传...") state = json.loads(Path(resume_file).read_text()) # 跳过已执行步骤,从 last_action.step + 1 开始 start_step = state["last_action"]["step"] + 1 run_from_step(start_step, state, args) else: print("🆕 开启全新任务...") run_from_step(1, {}, args)注意:run_from_step()是新封装的函数,它会跳过初始化、首屏截图等前置步骤,直接基于state["current_context"]构造提示词,向模型发送:“你已点击首页搜索图标,当前页面标题为‘小红书首页’,可见元素包括‘搜索框’。请继续完成:输入‘美食’并搜索。”
3. 实战:三行代码让“打开抖音关注博主”支持续传
现在,我们把上述思路落地到具体指令:“打开抖音搜索抖音号为:dycwo11nt61d 的博主并关注他!”。整个流程原本需 7 步(启动App→找搜索→输ID→点搜索→进主页→找关注按钮→点击),中断常发生在第4步(等待搜索结果加载)。
3.1 修改 ADB 封装层:为关键动作添加验证钩子
在phone_agent/adb.py中,增强tap()和input_text()方法,加入屏幕变化校验:
# phone_agent/adb.py def tap(self, x: int, y: int, verify_change: bool = True) -> bool: """点击并可选验证屏幕是否变化""" self._run_adb_command(f"shell input tap {x} {y}") if verify_change: # 截图并计算哈希 new_screenshot = self.screenshot() new_hash = hashlib.md5(new_screenshot).hexdigest() # 与上次截图哈希对比(需在类中维护 last_screenshot_hash) if hasattr(self, 'last_screenshot_hash') and self.last_screenshot_hash != new_hash: self.last_screenshot_hash = new_hash return True else: # 无变化,可能点击无效,重试一次 time.sleep(1) new_screenshot = self.screenshot() new_hash = hashlib.md5(new_screenshot).hexdigest() if self.last_screenshot_hash != new_hash: self.last_screenshot_hash = new_hash return True return False这个改动极小,却让“点击是否生效”有了客观依据——不再是盲信 ADB 返回码,而是看屏幕是否真的变了。
3.2 扩展提示词模板:让模型理解“我在哪,要干啥”
在prompt_templates/phone_agent_vlm.jinja中,为续传场景新增分支:
{%- if resume_state %} 【续传上下文】 你已在上一轮完成以下动作: - 步骤{{ resume_state.last_action.step }}:{{ resume_state.last_action.action }} {{ resume_state.last_action.target }} - 当前页面标题:{{ resume_state.current_context.page_title }} - 可见关键元素:{{ resume_state.current_context.visible_elements | join(', ') }} 请严格基于此状态,继续执行用户原始指令的剩余部分。 {%- else %} 【全新任务】 用户指令:{{ instruction }} 请从零开始理解屏幕、规划并执行全部步骤。 {%- endif %}当检测到续传时,模型收到的不再是孤立指令,而是一份带进度条的工单。它知道“搜索框已打开”,就不会再浪费一次截图去确认首页状态。
3.3 验证效果:一次中断,两次体验
| 场景 | 全新执行耗时 | 续传执行耗时 | 关键差异 |
|---|---|---|---|
| USB 连接稳定 | 42秒 | — | 无中断 |
| WiFi 中断(第4步) | 重试需42秒 | 11秒 | 跳过前4步,直接从“输入ID”开始,仅需截图验证当前页面+发送文本+点击 |
11秒的完成时间,证明续传不是概念,而是实打实的效率提升。它省掉的不是几行代码,而是用户等待时的焦躁感。
4. 进阶思考:从“能续”到“更稳”的三个延伸方向
断点续传解决了“中断后怎么办”,但真正的工程健壮性,还藏在更底层的设计选择里。以下是三个值得你在实际部署中优先尝试的优化点:
4.1 动作幂等性:让“重复点击”不再引发错误
当前tap()操作不具备幂等性——如果网络延迟导致指令重复发送,可能连续点击两次“关注”按钮,触发异常弹窗。解决方案很简单:在点击前,先 OCR 识别目标区域文字。若已显示“已关注”,则跳过点击。
# 在 tap() 前插入 if target == "follow_button": text = self.ocr_region(x-100, y-50, x+100, y+50) # 截取按钮区域OCR if "已关注" in text or "Following" in text: print(" 目标已达成,跳过重复操作") return True这比加锁或队列更轻量,且完全兼容现有架构。
4.2 分布式任务队列:当一台电脑不够用时
Open-AutoGLM默认单机运行。若你需同时控制10台手机执行不同任务,手动管理10个终端窗口和10个断点文件会失控。此时,引入Celery + Redis是平滑演进路径:
- 控制端只发
task.apply_async(args=[device_id, instruction]) - Worker 进程负责执行、状态上报、断点写入
- Redis 存储所有任务状态,Web UI 可实时查看进度
所有业务逻辑(截图、OCR、模型调用)无需重写,只是执行载体从main.py变成了 Celery Worker。
4.3 用户接管无缝切换:验证码场景的优雅降级
Phone Agent 提到“支持人工接管”,但原生实现是粗暴暂停。更好的方式是:当模型识别到验证码图片时,自动生成一个带时效的二维码,用户扫码后,在手机端 H5 页面手动输入,结果自动回传至任务上下文,并触发续传。
这需要新增一个轻量 Web 服务(Flask 即可),但它带来的体验跃迁是质的:AI 不再是“卡死”,而是“邀请协作”。
5. 总结:断点续传的本质,是把AI当作一个可信赖的同事
我们花了大量篇幅讲技术实现,但最核心的启示其实很朴素:AutoGLM-Phone 不是一个黑箱工具,而是一个需要被赋予“工作记忆”的数字同事。它不该因为一次网络波动就忘记自己正在写的报告;也不该因为终端关闭就抹掉已经画好的半张设计图。
断点续传功能的价值,从来不止于节省那几十秒重试时间。它代表着一种工程思维的成熟——从“能跑通”到“敢交付”,从“Demo惊艳”到“生产可用”。当你在task_state/目录下看到一个个带时间戳的 JSON 文件,它们不只是状态快照,更是系统可靠性的无声证明。
下一步,你可以:
- 把本文的 JSON 断点方案直接集成进你的
Open-AutoGLM仓库; - 为
tap()和input_text()方法加上屏幕变化验证; - 尝试用
find_resumable_task()函数替换main.py的启动逻辑。
真正的 AI 助理,不在于它多快能完成第一个任务,而在于它多稳能陪你完成第一百个任务。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。