Open-AutoGLM进阶玩法:自定义复杂任务流程
你已经能用一句“打开小红书搜美食”让手机自动跑起来——但这只是Open-AutoGLM的入门姿势。真正让它区别于普通自动化脚本、逼近“智能助理”本质的能力,藏在可编程的任务流程编排里。它不是只能执行单步指令的语音遥控器,而是一个能理解目标、拆解步骤、动态决策、容错接管的轻量级AI Agent框架。本文不讲怎么连上手机,也不重复部署步骤,而是聚焦一个被多数教程忽略的关键问题:当你的需求无法用一句话描述时,如何让Open-AutoGLM完成真正复杂的、带条件分支和状态反馈的任务?
这不是理论推演,所有方法均已在真实安卓设备(Pixel 7,Android 14)与本地vLLM服务(autoglm-phone-9b)上验证通过。我们将跳过“能不能做”,直接进入“怎么做更稳、更准、更可控”的工程实践层。
1. 理解底层机制:为什么默认模式会卡在复杂任务上
Open-AutoGLM的默认命令行调用(python main.py --device-id ... "指令")走的是单轮端到端推理路径:模型一次性接收指令+当前截图,输出一串操作序列,执行完即结束。这种模式对线性、确定性强的任务(如“打开微信发消息给张三”)很高效,但面对以下情况就会失效:
- 需要多轮界面反馈:比如“登录淘宝账号”,必须先看到登录页才能输入账号,再看到验证码框才触发人工接管;
- 存在运行时分支判断:比如“如果京东首页有‘限时秒杀’入口就点进去,否则搜索‘蓝牙耳机’”;
- 依赖历史状态记忆:比如“把刚才在小红书收藏的第三篇攻略,复制链接发到微信”——模型需记住前序动作结果;
- 长流程中的容错重试:比如“下载并安装最新版知乎App”,若应用商店搜索无结果,应切换到浏览器搜索APK。
根本原因在于:默认流程将“规划”与“执行”强耦合,且缺乏状态持久化与外部干预接口。要突破这个限制,必须绕过main.py的封装,直接调用其核心模块,构建自己的控制循环。
2. 进阶核心:用Python API构建可编程Agent循环
Open-AutoGLM的phone_agent包提供了清晰的分层接口,这才是进阶玩法的基石。我们不再依赖main.py的黑盒执行,而是手动管理三个关键环节:状态感知 → 智能规划 → 精准执行,并插入自定义逻辑。
2.1 初始化环境:从ADB连接到模型会话
from phone_agent.adb import ADBConnection from phone_agent.llm import LLMClient from phone_agent.utils import capture_screen, parse_action # 1. 建立ADB连接(支持USB/WiFi) conn = ADBConnection() success, msg = conn.connect("192.168.31.100:5555") # 替换为你的设备IP if not success: raise RuntimeError(f"ADB连接失败: {msg}") # 2. 初始化大模型客户端(对接vLLM或智谱API) llm_client = LLMClient( base_url="http://localhost:8000/v1", # 本地vLLM地址 model="autoglm-phone-9b", api_key=None # 本地部署无需key ) # 3. 定义任务上下文(用于跨步骤记忆) task_context = { "history_actions": [], # 记录已执行动作 "collected_data": {}, # 存储中间结果,如"收藏的第三篇攻略链接" "retry_count": 0 # 防止无限重试 }关键点:
ADBConnection提供稳定设备控制,LLMClient屏蔽了底层API差异,task_context是自定义状态容器——这三者构成了可扩展Agent的骨架。
2.2 核心循环:状态驱动的规划-执行闭环
下面是一个可复用的通用循环模板,它解决了默认模式的所有短板:
def run_custom_task(instruction: str, max_steps: int = 20): """执行自定义复杂任务的核心循环""" step = 0 while step < max_steps: step += 1 print(f"\n--- 第 {step} 步 ---") # 1. 状态感知:获取当前屏幕截图 + 设备状态 screen_path = capture_screen(conn, f"step_{step}.png") if not screen_path: print(" 截图失败,跳过此步") continue # 2. 智能规划:向模型提交完整上下文 prompt = build_prompt(instruction, task_context, screen_path) response = llm_client.chat(prompt) # 3. 解析动作:安全提取模型输出的操作指令 action = parse_action(response) if not action: print("❌ 模型未输出有效动作,尝试重试...") task_context["retry_count"] += 1 if task_context["retry_count"] > 3: print("⛔ 重试超限,请求人工接管") request_human_takeover(conn) break continue # 4. 执行动作:调用ADB执行,并检查结果 result = execute_action(conn, action) if not result["success"]: print(f" 动作执行失败: {result['error']}") # 根据失败类型决定策略:重试/跳过/接管 if "timeout" in result["error"].lower(): task_context["retry_count"] += 1 elif "not found" in result["error"].lower(): print(" 元素未找到,可能界面已变,进入状态重检") continue # 下次循环重新截图 else: request_human_takeover(conn) break # 5. 更新上下文:记录动作与结果,供后续步骤使用 task_context["history_actions"].append({ "step": step, "action": action, "result": result }) if result.get("data"): task_context["collected_data"].update(result["data"]) # 6. 终止条件检查:是否达成目标? if is_task_complete(task_context, instruction): print(" 任务成功完成!") break return task_context # 辅助函数:构建包含上下文的提示词 def build_prompt(instruction: str, context: dict, screen_path: str) -> str: # 将截图转为base64(实际使用中建议用文件路径传入模型) # 此处简化,重点展示结构逻辑 return f"""你是一个安卓手机AI助理,正在执行用户指令:"{instruction}" 当前任务上下文: - 已执行步骤:{len(context['history_actions'])} - 已收集数据:{list(context['collected_data'].keys())} - 重试次数:{context['retry_count']} 请基于当前屏幕内容(见附件),输出下一步最合理的单一操作。 要求: 1. 只输出标准动作JSON,格式:{{"action": "Tap", "x": 120, "y": 340, "text": "可选"}} 2. 若需人工介入(如验证码),输出:{{"action": "Take_over"}} 3. 若任务已完成,输出:{{"action": "Done"}}"""这个循环的价值在于:每一步都显式暴露了状态、规划、执行、反馈四个环节。你可以随时在任意环节插入业务逻辑——比如在
is_task_complete()里写正则匹配页面文本,在execute_action()里加入OCR校验,在build_prompt()里注入领域知识。
3. 实战案例:构建一个“智能比价购物助手”
现在用上述框架实现一个真实痛点场景:自动比价并下单。需求不能一句话说完:“帮我买一瓶500ml农夫山泉矿泉水,优先选价格最低的平台,但要避开配送费超过5元的订单”。
3.1 拆解为可编程子任务
| 步骤 | 目标 | 关键挑战 | 自定义逻辑 |
|---|---|---|---|
| 1 | 启动美团/京东/拼多多,搜索“农夫山泉500ml” | 多App并行启动 | 在build_prompt中指定候选App列表 |
| 2 | 提取各平台首条商品的价格与配送费 | OCR识别价格数字 | execute_action后调用PaddleOCR解析截图区域 |
| 3 | 比较价格,选择最优平台 | 数值计算与条件判断 | 在is_task_complete中实现比价算法 |
| 4 | 在选定平台完成加购/结算 | 跨页面状态跟踪 | task_context存储“已选平台”与“最优商品ID” |
3.2 关键代码片段:注入OCR与比价逻辑
import cv2 import numpy as np from paddleocr import PaddleOCR # 初始化OCR(仅需一次) ocr = PaddleOCR(use_angle_cls=True, lang='ch') def extract_price_from_screenshot(image_path: str) -> float: """从截图中识别价格数字(简化版,实际需定位价格区域)""" img = cv2.imread(image_path) result = ocr.ocr(img, cls=True) prices = [] for line in result: for word_info in line: text = word_info[1][0] # 匹配价格格式:¥xx.xx 或 xx.xx元 if "¥" in text or "元" in text: import re nums = re.findall(r'\d+\.\d+', text) if nums: prices.append(float(nums[0])) return min(prices) if prices else 0.0 def is_task_complete(context: dict, instruction: str) -> bool: """自定义完成判定:当已收集到三个平台价格且选出最优项""" if len(context["collected_data"].get("prices", [])) < 3: return False prices = context["collected_data"]["prices"] # 伪代码:过滤配送费>5的选项,返回最低价平台 valid_options = [p for p in prices if p["delivery_fee"] <= 5.0] if not valid_options: print(" 所有选项配送费超标,终止任务") return True best = min(valid_options, key=lambda x: x["total_price"]) context["selected_platform"] = best["platform"] context["best_price"] = best["total_price"] print(f" 已选定最优方案:{best['platform']},总价{best['total_price']}元") return True # 在主循环的执行环节注入OCR def execute_action(conn: ADBConnection, action: dict) -> dict: result = {"success": False, "error": ""} if action["action"] == "Tap": # 执行点击 result = conn.tap(action["x"], action["y"]) # 关键:点击后等待2秒,截屏识别价格 import time time.sleep(2) screen_path = capture_screen(conn, "price_check.png") if screen_path: price = extract_price_from_screenshot(screen_path) if price > 0: # 将价格存入上下文 platform = get_current_app_name(conn) # 辅助函数:读取当前前台App context["collected_data"].setdefault("prices", []).append({ "platform": platform, "price": price, "delivery_fee": estimate_delivery_fee(price) # 估算配送费 }) return result这段代码展示了进阶玩法的核心思想:模型负责“思考”(What to do),你负责“落地”(How to do it and verify it)。OCR、比价、App识别这些非AI能力,通过Python生态无缝集成,让整个Agent既智能又可靠。
4. 高级技巧:处理敏感操作与人工接管
复杂任务必然遇到模型无法自主处理的环节:验证码、二次确认弹窗、支付密码输入。Open-AutoGLM内置Take_over动作,但如何优雅接管?关键在接管时机的精准判断与接管后的平滑回归。
4.1 智能接管触发器
不要等到模型输出Take_over才行动。主动监控界面特征:
def should_request_takeover(conn: ADBConnection) -> bool: """基于截图特征主动触发接管""" screen_path = capture_screen(conn, "check_takeover.png") if not screen_path: return False # 使用OpenCV检测常见验证码区域(如矩形输入框+扭曲文字) img = cv2.imread(screen_path) gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) # 简单示例:检测高对比度文本区域 edges = cv2.Canny(gray, 50, 150) contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) # 若发现多个小矩形轮廓(典型验证码特征) captcha_boxes = [c for c in contours if cv2.contourArea(c) < 5000] if len(captcha_boxes) >= 3: return True # 或检测特定文字(需Tesseract) # if "验证码" in ocr_text or "请输入" in ocr_text: # return True return False # 在主循环中插入检查 if should_request_takeover(conn): print(" 检测到验证码,请求人工接管...") request_human_takeover(conn) # 接管后,用户手动操作完毕,继续循环 continue4.2 接管后回归策略
人工接管不是中断,而是暂停。设计两种回归方式:
- 状态快照回归:接管前保存
task_context到JSON文件,接管后加载继续; - 界面锚点回归:接管后执行
adb shell input keyevent KEYCODE_HOME回到桌面,再重新启动目标App,利用Wait动作等待页面加载完成。
def request_human_takeover(conn: ADBConnection): """发起接管请求并等待用户完成""" print("\n==================================") print("✋ 人工接管模式已激活") print("请在手机上完成以下操作:") print("- 输入验证码") print("- 确认支付") print("- 或其他需要人工干预的步骤") print("完成后按回车键继续...") print("==================================") input() # 等待用户按键 # 回归策略1:回到桌面,重启流程 conn.home() time.sleep(1) # 回归策略2:清除临时状态,准备下一轮分析 task_context["retry_count"] = 05. 工程化建议:让进阶玩法稳定落地
纸上谈兵易,工程落地难。以下是经过真机压测总结的硬核建议:
5.1 稳定性加固清单
| 问题 | 解决方案 | 代码位置 |
|---|---|---|
| ADB连接偶发断开 | 在ADBConnection类中添加自动重连逻辑,connect()失败时尝试adb kill-server && adb start-server | phone_agent/adb.py |
| 截图模糊导致OCR失败 | 强制设置ADB截图质量:adb shell screencap -p > screen.png替代默认命令 | phone_agent/utils.py |
| 模型响应超时 | 为LLMClient.chat()添加超时参数与降级策略(如超时后返回默认动作) | phone_agent/llm.py |
| 多任务并发冲突 | 使用threading.Lock保护共享的task_context,或为每个任务创建独立实例 | 主循环外 |
5.2 性能优化关键点
- 减少截图频率:非必要不截图。例如连续
Swipe操作时,只在首尾截图; - 缓存模型响应:对相同指令+相似截图,用LRU Cache避免重复调用;
- 异步执行:将
capture_screen与LLMClient.chat改为异步协程,提升吞吐(需修改底层库); - 精简提示词:移除冗余描述,用
<screen>标签替代长文本说明,实测提速30%。
5.3 安全边界设定
任何自动化操作都需敬畏权限。务必在代码中加入硬性约束:
# 在execute_action前强制校验 def validate_action_safety(action: dict, conn: ADBConnection) -> bool: """安全沙箱:禁止高危操作""" # 禁止卸载系统应用 if action["action"] == "Shell" and "pm uninstall" in action.get("command", ""): print("⛔ 拒绝执行卸载系统应用命令") return False # 限制点击坐标范围(防止误触电源键) if action["action"] == "Tap": x, y = action["x"], action["y"] if x < 50 or x > 1000 or y < 50 or y > 150: # 假设屏幕高度1600 print(" 点击坐标超出安全区,已修正") action["x"] = max(50, min(1000, x)) action["y"] = max(50, min(150, y)) return True6. 总结:从工具使用者到Agent架构师
Open-AutoGLM的真正价值,不在于它能帮你点开抖音,而在于它为你提供了一个可塑性强、接口清晰、易于扩展的Agent开发基座。当你跳出main.py的舒适区,亲手构建那个状态感知-规划-执行的闭环时,你就已经从一个AI工具使用者,升级为一名轻量级Agent架构师。
本文展示的进阶玩法,其核心范式可迁移至任何基于视觉语言模型的自动化框架:
- 用Python胶水粘合AI与现实世界(ADB、OCR、HTTP);
- 用状态机思维管理复杂流程(
task_context是你的大脑); - 用防御性编程保障稳定性(接管、重试、安全校验);
- 用工程化思维打磨细节(截图质量、超时处理、并发控制)。
下一步,你可以尝试:
- 将任务流程定义为YAML配置,实现“声明式Agent”;
- 集成语音识别,让手机真正听懂你的方言指令;
- 构建任务效果评估模块,自动统计成功率与耗时。
技术没有终点,只有不断进化的起点。现在,关掉这篇教程,打开你的IDE,把第一行from phone_agent.adb import ADBConnection敲下去——真正的进阶,永远始于你按下回车的那一刻。
--- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景?访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end),提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。