""" GuiAgent集成模块 (使用guiagent_core重构版本) 用于在DroidBot中调用GuiAgent处理特定场景 重构说明： - 使用guiagent_core的GuiAgentDecisionMaker替代完整GuiAgent - 消除重复device连接，复用DroidBot的device - 保持所有接口不变，确保向后兼容 - 场景配置从统一配置文件加载，支持平台特定覆盖 """ import os import sys import logging import time from typing import Dict, List, Optional, Tuple, Any # 添加autool根路径到系统路径（用于导入setting） from pathlib import Path autool_root = str(Path(__file__).resolve().parent.parent) if autool_root not in sys.path: sys.path.insert(0, autool_root) # 尝试导入guiagent_core try: from .guiagent_core import GuiAgentDecisionMaker, ContextManager GUIAGENT_AVAILABLE = True except ImportError as e: logging.error(f"无法导入guiagent_core: {e}") GUIAGENT_AVAILABLE = False class GuiAgentBridge: """ DroidBot与GuiAgent之间的桥梁（使用guiagent_core重构版本） 主要改进： 1. 使用GuiAgentDecisionMaker替代完整GuiAgent 2. 复用DroidBot的device进行截图和执行 3. 添加详细debug日志 4. 保持所有现有接口不变 """ def __init__(self, device, app=None, app_name=None, utg=None, input_manager=None): """ 初始化GuiAgent桥接器 :param device: DroidBot的设备实例 :param app: DroidBot的App实例（可选） :param app_name: App名称（可选，直接指定） :param utg: UTG实例（用于记录每步state transition） :param input_manager: InputManager实例（用于设置执行flag） """ self.logger = logging.getLogger(self.__class__.__name__) self.device = device self.app = app self.app_name = app_name self.utg = utg # UTG引用，用于记录每步state transition self.input_manager = input_manager # input_manager引用，用于设置执行flag self.decision_maker = None self.context = None self.is_enabled = GUIAGENT_AVAILABLE # 加载场景配置 (从统一配置文件加载，支持平台覆盖) platform_name = device.get_platform_name() if device else "android" try: from .guiagent_core.scene_config_loader import load_scene_config scene_config = load_scene_config(platform_name) self.keywords = scene_config.get("keywords", {}) self.instructions = scene_config.get("instructions", {}) self.step_limits = scene_config.get("step_limits", {}) self.logger.info(f"[GuiAgent] 加载场景配置成功: platform={platform_name}, scenes={list(self.keywords.keys())}") except Exception as e: self.logger.error(f"[GuiAgent] 加载场景配置失败: {e}, 使用空配置") import traceback self.logger.debug(traceback.format_exc()) # 降级处理: 使用空配置 self.keywords = {} self.instructions = {} self.step_limits = {} # 上次动作坐标（用于在截图上绘制绿圈标记） self._last_action_coords = None # (x, y) 绝对像素坐标 # 状态监控相关 self.last_state_time = time.time() self.last_state_hash = None # 界面状态跟踪 - 记录每个界面已经处理的操作类型 self.processed_states = {} # {state_hash: set(categories)} if self.is_enabled: self._init_agent() def _init_agent(self): """初始化GuiAgent决策引擎（使用guiagent_core）""" try: # 获取设备分辨率 display_info = self.device.get_display_info() width = display_info.get('width', 1080) height = display_info.get('height', 1920) resolution = (width, height) self.logger.debug(f"初始化GuiAgent决策引擎: resolution={resolution}") # 根据设备平台动态配置 platform_name = self.device.get_platform_name() if self.device else "android" # 强制使用归一化坐标 absolute_mode = False # 创建决策引擎（不创建设备连接） self.decision_maker = GuiAgentDecisionMaker( platform=platform_name, resolution=resolution, absolute_mode=absolute_mode ) self.logger.info(f"GuiAgent决策引擎初始化成功 (guiagent_core版本)") self.logger.debug(f"决策引擎配置: platform={platform_name}, resolution={resolution}, absolute_mode={absolute_mode}") except Exception as e: self.logger.error(f"GuiAgent决策引擎初始化失败: {e}") import traceback self.logger.debug(traceback.format_exc()) self.is_enabled = False def is_state_processed(self, current_state, category: str) -> bool: """ 检查当前界面是否已经处理过特定类别的操作 :param current_state: 当前设备状态 :param category: 操作类别 :return: 如果已处理过返回True，否则返回False """ if not current_state or not hasattr(current_state, 'state_str'): return False state_hash = current_state.state_str if state_hash in self.processed_states: return category in self.processed_states[state_hash] return False def mark_state_processed(self, current_state, category: str): """ 标记当前界面已处理过特定类别的操作 :param current_state: 当前设备状态 :param category: 操作类别 """ if not current_state or not hasattr(current_state, 'state_str'): return state_hash = current_state.state_str if state_hash not in self.processed_states: self.processed_states[state_hash] = set() self.processed_states[state_hash].add(category) self.logger.info(f"[GuiAgent] 标记界面 {state_hash[:16]}... 已处理 '{category}' 操作") def check_keywords(self, text: str) -> Optional[str]: """ 检查文本中是否包含需要特殊处理的关键词 :param text: 要检查的文本 :return: 如果找到关键词，返回对应的类别，否则返回None """ if not text: return None text_lower = text.lower() for category, keywords in self.keywords.items(): for keyword in keywords: if keyword.lower() in text_lower: self.logger.debug(f"[GuiAgent] 检测到关键词: '{keyword}' (类别: {category})") return category return None def get_text_within_bounds(self, view: dict, all_views: list) -> str: """ 获取控件边界框内所有视图的文本（用于关键词检测） :param view: 目标视图字典 :param all_views: 当前状态的所有视图列表 :return: 边界框内所有文本的拼接字符串 """ bounds = view.get('bounds', [[0, 0], [0, 0]]) left, top = bounds[0] right, bottom = bounds[1] if right <= left or bottom <= top: return view.get('text', '') or view.get('content_description', '') or '' texts = [] for v in all_views: v_bounds = v.get('bounds', [[0, 0], [0, 0]]) v_left, v_top = v_bounds[0] v_right, v_bottom = v_bounds[1] if v_left >= left and v_top >= top and v_right <= right and v_bottom <= bottom: if v.get('text'): texts.append(v['text']) elif v.get('content_description'): texts.append(v['content_description']) return ' '.join(texts) def _get_scene_max_steps(self, category: str, explicit_max_steps: Optional[int] = None) -> int: """ 获取指定场景的最大步数 优先级: 显式传入参数 > scene_configs.json 场景配置 > default_steps > 硬编码兜底20 :param category: 场景类别 :param explicit_max_steps: 显式传入的步数（None 表示未指定） :return: 最大步数 """ if explicit_max_steps is not None: self.logger.debug(f"[GuiAgent] 场景 '{category}' 使用显式步数: {explicit_max_steps}") return explicit_max_steps # 过滤掉注释键，只查找场景配置 limits = {k: v for k, v in self.step_limits.items() if not k.startswith('_')} default_steps = limits.get("default_steps", 20) scene_steps = limits.get(category, default_steps) source = "场景配置" if category in limits else "default_steps" self.logger.debug(f"[GuiAgent] 场景 '{category}' 使用步数: {scene_steps} (来源: {source})") return scene_steps def handle_with_guiagent(self, category: str, context: Dict[str, Any] = None, max_steps: Optional[int] = None, max_error_steps: int = 20) -> Tuple[bool, Optional[str], Optional[int]]: """ 使用GuiAgent处理特定场景 注意：此方法会执行完整个任务（多步操作），然后返回成功/失败 这与input_policy的预期一致：调用后等待agent处理完，记录为一次事件 重要改进： 1. 设置执行flag，执行过程中不进行前台应用拉回检测，不进行总步数累计 2. 记录每次agent决策event前后的state到UTG 3. 记录token开销并在执行完成后打印 :param category: 场景类别（login, register, payment, game等） :param context: 上下文信息 :param max_steps: 最大执行步数（默认 20） :param max_error_steps: 最大连续错误步数（默认 20） :return: (处理是否成功, 失败原因, stuck_reason_code) stuck_reason_code: 仅当 report_stuck_reason 时返回具体code，否则为None """ if not self.is_enabled or not self.decision_maker: self.logger.warning("[GuiAgent] GuiAgent不可用，无法处理") return False, None, None current_state = context.get("state") if context else None # 检查当前界面是否已经处理过此类操作 if current_state and self.is_state_processed(current_state, category): self.logger.info(f"[GuiAgent] 当前界面已处理过 '{category}' 操作，跳过") return False, None, None # 设置执行flag - 执行过程中跳过前台检查和步数累计 if self.input_manager: self.input_manager.is_guiagent_executing = True self.logger.debug("[GuiAgent] 设置执行flag，跳过前台检查和步数累计") try: # 生成任务指令 # 生成任务指令 instruction = self._generate_instruction(category, context) self.logger.info(f"[GuiAgent] 开始处理任务 (类别: {category}): {instruction}") # [二级检测] 验证当前界面是否确实属于目标场景（跳过卡住状态的验证） if category not in ("game_initial", "explore_stuck", "stuck_escape"): # 先截图用于验证 screenshot_path = self.device.take_screenshot() if screenshot_path and not self.decision_maker.verify_screen(category, screenshot_path=screenshot_path): self.logger.warning(f"界面验证失败，当前可能并非 {category} 场景，跳过 GuiAgent 处理") return False, None, None # 创建新的上下文（重置上次动作坐标，从fastbot重新进入时不记忆） self._last_action_coords = None self.context = self.decision_maker.create_context(instruction) MAX_STEPS = self._get_scene_max_steps(category, max_steps) MAX_ERROR_STEPS = max_error_steps self.logger.info(f"[GuiAgent] 场景 '{category}' 步数限制: {MAX_STEPS} 步") executed_steps = 0 error_steps = 0 success_count = 0 task_finished = False # 标记agent是否明确报告任务完成 # 记录初始state before_state = self.device.get_current_state() guiagent_message = None # 统一的消息记录（成功或失败） stuck_reason_code = None # 记录卡住原因代码（仅 report_stuck_reason 时设置） while executed_steps < MAX_STEPS: if error_steps >= MAX_ERROR_STEPS: self.logger.error(f"[GuiAgent] 总计失败{MAX_ERROR_STEPS}次，终止处理") if category in ('login', 'register'): category_name = "登录" if category == "login" else "注册" guiagent_message = f"{category_name}失败: 连续错误次数达到上限({MAX_ERROR_STEPS}次)" self.logger.warning(f"[GuiAgent] {guiagent_message}") error_steps = 0 break self.logger.debug(f"[GuiAgent] 执行第 {executed_steps + 1}/{MAX_STEPS} 步") # 1. 截图（使用DroidBot的设备） screenshot_path = self.device.take_screenshot() if not screenshot_path: self.logger.warning("[GuiAgent] 截图失败，终止处理") error_steps += 1 # 尝试等待 WDA 就绪 if hasattr(self.device, '_wait_wda_ready'): wda_ready = self.device._wait_wda_ready(timeout=30) if not wda_ready and hasattr(self.device, '_on_wda_failure'): self.device._on_wda_failure("GuiAgent 截图失败") # 等待 WDA 恢复完成 time.sleep(5) else: time.sleep(2) continue self.logger.debug(f"[GuiAgent] 截图完成: {screenshot_path}") # 决定是否需要添加网格和动作标记（例如卡住检测和某些特定场景不需要） draw_grid_and_marker = category not in ("explore_stuck",) # 2. 调用决策引擎（传入step用于日志记录） decision = self.decision_maker.decide_next_action( screenshot_path=screenshot_path, context=self.context, step=executed_steps + 1, last_action_coords=self._last_action_coords, draw_grid_and_marker=draw_grid_and_marker ) if not decision.get('success'): error_msg = decision.get('error', 'Unknown error') self.logger.error(f"[GuiAgent] 决策失败: {error_msg}") error_steps += 1 time.sleep(2) continue executed_steps += 1 action_type = decision['action_type'] action = decision.get('action_data', {}) thought = decision.get('thought', '') self.logger.info(f"[GuiAgent] LLM决策: {action} | Thought: {thought}") # 3. 检查是否完成 if decision.get('is_finished'): self.logger.info(f"[GuiAgent] 任务完成 (共执行 {executed_steps} 步)") success_count += 1 task_finished = True # 明确标记任务完成 # 处理 report_stuck_reason 动作 if action_type == 'report_stuck_reason': reason_code = str(action.get('reason_code', '0')) raw_message = action.get('message', '') stuck_reason_code = int(reason_code) # 记录 reason_code 用于上报 STUCK_REASON_MAP = { 1: "[失败] 登录注册", 2: "[失败] 启动异常", 3: "[成功] 测试正常", 4: "[失败] 地区限制", 5: "[失败] 需付费", 6: "[失败] 虚拟手机号无效", 7: "[失败] 虚拟身份无效", 8: "[失败] 注册校验失败", 9: "[失败] 非开放注册", 10: "[失败] 应用停服", 11: "[失败] 需实体证件", 12: "[失败] Root模式下无法使用", 0: "[失败] 其他原因", } if reason_code == '3': self.logger.info("[GuiAgent] 卡住诊断: 测试正常，尝试脱离卡住状态") if current_state and self.is_state_processed(current_state, "stuck_escape"): self.logger.info("[GuiAgent] stuck_escape 已处理过，不再重复调用") task_finished = False else: escape_success, _, _ = self.handle_with_guiagent( category="stuck_escape", context=context ) if escape_success: # 成功脱困不属于错误上报，保持消息为空。 self.logger.info("[GuiAgent] 已脱离卡住状态，不生成汇报消息") else: stuck_reason_code = 0 if raw_message: guiagent_message = f"探索卡住: [失败] {raw_message}" else: guiagent_message = "探索卡住: [失败] 脱离卡住状态失败" self.logger.warning(f"[GuiAgent] 脱离卡住状态失败，保留原始卡住原因: {guiagent_message}") else: reason_desc = STUCK_REASON_MAP.get(int(reason_code), f"[失败] 未知原因({reason_code})") guiagent_message = f"探索卡住: {reason_desc}" if raw_message: guiagent_message += f" ({raw_message})" self.logger.info(f"[GuiAgent] 卡住诊断结果: {guiagent_message}") elif category in ('login', 'register'): category_name = "登录" if category == "login" else "注册" guiagent_message = f"{category_name}成功" self.logger.info(f"[GuiAgent] {guiagent_message}") break # 4. 处理receive_email特殊动作：获取邮件内容并反馈给上下文 if action_type == 'receive_email': from .guiagent_core.utils import receive_email email_results = receive_email() if email_results: feedbacks = [] for i, email in enumerate(email_results): feedbacks.append(f"邮件{i+1}主题: '{email.get('subject', '')}', 内容: {email.get('content', '')}\n") feedback = "\n".join(feedbacks) self.logger.info(f"[GuiAgent] 收到 {len(email_results)} 封邮件: {feedback[:100]}...") self.context.add_user_message(f"近期收到以下邮件，请你根据信息判断选取其中的哪一封:\n{feedback}") self.logger.info("[GuiAgent] 已将邮件内容发送给智能体") else: self.context.add_user_message("未收到新邮件") self.logger.warning("[GuiAgent] 未收到新邮件") success_count += 1 continue # 不执行设备动作，继续下一轮决策 # 5. 处理wait动作：等待一段时间 if action_type == 'wait': wait_duration = decision.get('action_data', {}).get('duration', 2) self.logger.info(f"[GuiAgent] 执行等待动作: {wait_duration}秒") time.sleep(wait_duration) success_count += 1 continue # 不生成设备事件，继续下一轮决策 # 5.1 处理滑块验证码技能 if action_type == 'solve_slider_captcha': from .guiagent_core.skills import solve_slider_captcha action_data = decision.get('action_data', {}) captcha_region = action_data.get('captcha_region', []) slider_position = action_data.get('slider_position', []) # 坐标转换: 归一化坐标 [0-1000] -> 绝对像素坐标 display_info = self.device.get_display_info() width = display_info.get('width', 1080) height = display_info.get('height', 1920) def to_abs(coords, is_region=False): if not coords: return [] if is_region and len(coords) == 4: return [ int(coords[0] * width / 1000), int(coords[1] * height / 1000), int(coords[2] * width / 1000), int(coords[3] * height / 1000) ] elif len(coords) == 2: return [int(coords[0] * width / 1000), int(coords[1] * height / 1000)] return coords region_abs = to_abs(captcha_region, is_region=True) slider_abs = to_abs(slider_position) self.logger.info(f"[GuiAgent] 执行滑块验证码技能: region={region_abs}, slider={slider_abs}") success, msg = solve_slider_captcha( device=self.device, captcha_region=region_abs, slider_position=slider_abs, debug=True ) # 将结果反馈给上下文 if success: self.context.add_user_message(f"滑块验证码操作已完成: {msg}") self.logger.info(f"[GuiAgent] 滑块验证码成功: {msg}") else: self.context.add_user_message(f"滑块验证码操作失败: {msg}，请尝试其他方式或手动处理") self.logger.warning(f"[GuiAgent] 滑块验证码失败: {msg}") success_count += 1 time.sleep(2) # 等待验证结果 continue # 技能已执行，继续下一轮决策 # 5.2 处理图片验证码技能 if action_type == 'solve_image_captcha': from .guiagent_core.skills import solve_image_captcha from .core import PlatformFactory action_data = decision.get('action_data', {}) captcha_region = action_data.get('captcha_region', []) input_field = action_data.get('input_field', []) # 坐标转换 display_info = self.device.get_display_info() width = display_info.get('width', 1080) height = display_info.get('height', 1920) def to_abs(coords, is_region=False): if not coords: return [] if is_region and len(coords) == 4: return [ int(coords[0] * width / 1000), int(coords[1] * height / 1000), int(coords[2] * width / 1000), int(coords[3] * height / 1000) ] elif len(coords) == 2: return [int(coords[0] * width / 1000), int(coords[1] * height / 1000)] return coords region_abs = to_abs(captcha_region, is_region=True) field_abs = to_abs(input_field) self.logger.info(f"[GuiAgent] 执行图片验证码技能: region={region_abs}, input_field={field_abs}") success, captcha_text = solve_image_captcha( device=self.device, captcha_region=region_abs, debug=True ) if success and captcha_text: # 先点击输入框 if field_abs: platform = self.device.get_platform_name() TouchEvent = PlatformFactory.get_event_class(platform, 'touch') tap_event = TouchEvent(x=field_abs[0], y=field_abs[1]) self.device.send_event(tap_event) time.sleep(0.5) # 输入验证码文本 platform = self.device.get_platform_name() SetTextEvent = PlatformFactory.get_event_class(platform, 'set_text') text_event = SetTextEvent(text=captcha_text) self.device.send_event(text_event) self.context.add_user_message(f"图片验证码已识别并输入: {captcha_text}") self.logger.info(f"[GuiAgent] 图片验证码成功: {captcha_text}") else: self.context.add_user_message(f"图片验证码识别失败: {captcha_text}，请尝试其他方式") self.logger.warning(f"[GuiAgent] 图片验证码失败: {captcha_text}") success_count += 1 time.sleep(1) continue # 技能已执行，继续下一轮决策 if action_type == 'login_ios': from .guiagent_core.utils import login_ios login_success, login_message = login_ios(self.device) if login_success: self.context.add_user_message(f"iOS登录脚本执行成功: {login_message}") self.logger.info(f"[GuiAgent] login_ios 成功: {login_message}") success_count += 1 else: self.context.add_user_message(f"iOS登录脚本失败: {login_message}，请根据截屏内容完成登录步骤") self.logger.warning(f"[GuiAgent] login_ios 失败: {login_message}") error_steps += 1 time.sleep(2) continue # 技能已执行，继续下一轮决策 # 6. 转换为平台事件并执行 event = self._convert_decision_to_event(decision, before_state) if event: self.logger.debug(f"[GuiAgent] 执行事件: {type(event).__name__}") # 执行前保存状态 before_exec_state = self.device.get_current_state() # 执行动作 self.device.send_event(event) time.sleep(2) # 等待UI更新 # 执行后获取状态 after_exec_state = self.device.get_current_state() # 验证执行效果 verify_success, verify_msg = self._verify_action_effect( before_exec_state, after_exec_state, action_type ) if verify_success: self.context.add_execution_feedback(True) success_count += 1 else: self.context.add_execution_feedback(False, verify_msg) self.logger.warning(f"[GuiAgent] 执行验证失败: {verify_msg}") error_steps += 1 # 记录state transition到UTG if self.utg: if before_state and after_exec_state: self.utg.add_transition(event, before_state, after_exec_state, is_guiagent_event=True) self.logger.debug(f"[GuiAgent] 记录state transition: {before_state.state_str[:16]}... -> {after_exec_state.state_str[:16]}...") before_state = after_exec_state # 更新before_state为下一步 else: self.logger.warning(f"[GuiAgent] 无法生成事件 (action_type: {action_type})") executed_steps -= 1 error_steps += 1 self.context.add_execution_feedback(False, f"无法生成 {action_type} 事件") time.sleep(2) continue # 判断是否成功（有执行过有效操作） success = success_count > 0 # 打印Agent步数 self.logger.info(f"[GuiAgent] 任务执行完成, Agent步数={success_count}") # 检查login/register是否因超出步数限制而失败 if category in ('login', 'register') and executed_steps >= MAX_STEPS and not task_finished: category_name = "登录" if category == "login" else "注册" guiagent_message = f"{category_name}失败: 超出最大步数限制({MAX_STEPS}步)" self.logger.warning(f"[GuiAgent] {guiagent_message}") # 只有当agent明确报告任务完成时，才标记界面已处理 # 这样处理失败的界面下次还会重试 if task_finished: self.logger.info(f"[GuiAgent] 处理成功 (类别: {category}, 执行了 {success_count} 个有效操作)") if current_state: self.mark_state_processed(current_state, category) else: if success: self.logger.warning(f"[GuiAgent] 任务未完成但执行了操作 (类别: {category}, 执行了 {success_count} 个操作, 下次将重试)") else: self.logger.warning(f"[GuiAgent] 处理失败 (类别: {category}, 未执行任何有效操作, 下次将重试)") return success, guiagent_message, stuck_reason_code except KeyboardInterrupt: raise except Exception as e: self.logger.error(f"[GuiAgent] 处理异常: {e}") import traceback self.logger.debug(traceback.format_exc()) return False, f"GuiAgent处理异常: {e}", None finally: # 确保执行flag被重置 if self.input_manager: self.input_manager.is_guiagent_executing = False self.logger.debug("[GuiAgent] 重置执行flag") def _convert_decision_to_event(self, decision: Dict[str, Any], current_state=None): """ 将GuiAgent决策转换为平台事件 注意：guiagent_core返回的是归一化坐标[0-1000]，需要转换为绝对像素坐标 使用PlatformFactory获取平台特定的事件类，支持Android和iOS """ from .core import PlatformFactory action_type = decision['action_type'] action_data = decision['action_data'] # 获取设备平台名称 platform = self.device.get_platform_name() # 获取分辨率用于坐标转换 display_info = self.device.get_display_info() width = display_info.get('width', 1080) height = display_info.get('height', 1920) resolution = (width, height) def normalize_to_absolute(coords): """归一化坐标 [0-1000] -> 绝对像素坐标""" if not coords or len(coords) != 2: return coords # 只有在非 absolute_mode 时才进行归一化转换 if self.decision_maker and self.decision_maker.absolute_mode: return [int(coords[0]), int(coords[1])] x, y = coords abs_x = int(x * resolution[0] / 1000) abs_y = int(y * resolution[1] / 1000) self.logger.debug(f"[GuiAgent] 坐标转换: ({x:.1f}, {y:.1f}) -> ({abs_x}, {abs_y})") return [abs_x, abs_y] try: if action_type == 'tap' or action_type == 'click': if 'target' in action_data and action_data['target']: x, y = normalize_to_absolute(action_data['target']) # 尝试找到最接近的 clickable view，使用精确的 view center if current_state: closest_view = self._find_closest_view(current_state, x, y) if closest_view: self.logger.info(f"[GuiAgent] 使用 view 对象替代坐标点击") self._last_action_coords = (x, y) # 记录原始坐标用于绿圈标记 TouchEvent = PlatformFactory.get_event_class(platform, 'touch') return TouchEvent(view=closest_view) self.logger.info(f"[GuiAgent] 生成TouchEvent: ({x}, {y})") self._last_action_coords = (x, y) TouchEvent = PlatformFactory.get_event_class(platform, 'touch') return TouchEvent(x=x, y=y) elif action_type == 'long_tap': if 'target' in action_data and action_data['target']: x, y = normalize_to_absolute(action_data['target']) self.logger.debug(f"[GuiAgent] 生成LongTouchEvent: ({x}, {y})") self._last_action_coords = (x, y) LongTouchEvent = PlatformFactory.get_event_class(platform, 'long_touch') return LongTouchEvent(x=x, y=y) elif action_type in ('drag', 'swipe'): if 'start' in action_data and 'end' in action_data: start_x, start_y = normalize_to_absolute(action_data['start']) end_x, end_y = normalize_to_absolute(action_data['end']) self.logger.debug(f"[GuiAgent] 生成SwipeEvent: ({start_x},{start_y}) -> ({end_x},{end_y})") self._last_action_coords = (start_x, start_y) SwipeEvent = PlatformFactory.get_event_class(platform, 'swipe') return SwipeEvent(start_x=start_x, start_y=start_y, end_x=end_x, end_y=end_y) elif action_type == 'type': if 'text' in action_data and action_data['text']: text = action_data['text'] self.logger.debug(f"[GuiAgent] 生成SetTextEvent: '{text}'") SetTextEvent = PlatformFactory.get_event_class(platform, 'set_text') return SetTextEvent(text=text) elif action_type == 'key_press': if 'key' in action_data and action_data['key']: key = action_data['key'] self.logger.debug(f"[GuiAgent] 生成KeyEvent: {key}") KeyEvent = PlatformFactory.get_event_class(platform, 'key') return KeyEvent(key_name=key) except Exception as e: self.logger.error(f"[GuiAgent] 事件转换失败: {e}") return None return None def _verify_action_effect(self, before_state, after_state, action_type: str) -> Tuple[bool, str]: """ 验证动作是否产生预期效果 Args: before_state: 执行前的设备状态 after_state: 执行后的设备状态 action_type: 动作类型 Returns: (是否成功, 失败原因) """ if not before_state or not after_state: return True, "" # 检查状态是否变化 state_changed = before_state.state_str != after_state.state_str if not state_changed: if action_type in ('tap', 'click', 'long_tap'): return False, "界面未发生变化，可能点击位置不准确或点击无效区域" elif action_type == 'type': return False, "输入未生效，可能未正确聚焦输入框" elif action_type in ('drag', 'swipe'): return False, "滑动未生效，可能滑动距离不足或方向错误" return True, "" def _find_closest_view(self, state, target_x: int, target_y: int, threshold: int = 50): """ 在 state.views 中找到最接近目标坐标的 clickable view Args: state: 当前设备状态 target_x: 目标 x 坐标 target_y: 目标 y 坐标 threshold: 最大距离阈值（像素） Returns: 最接近的 view 对象，如果没有找到则返回 None """ if not state or not hasattr(state, 'views'): return None closest_view = None min_distance = float('inf') for view in state.views: if not view.get('clickable') or not view.get('visible'): continue bounds = view.get('bounds', [[0, 0], [0, 0]]) center_x = (bounds[0][0] + bounds[1][0]) // 2 center_y = (bounds[0][1] + bounds[1][1]) // 2 distance = ((center_x - target_x)**2 + (center_y - target_y)**2)**0.5 if distance < min_distance and distance < threshold: min_distance = distance closest_view = view if closest_view: self.logger.debug(f"[GuiAgent] 找到最近的 view，距离: {min_distance:.1f}px") return closest_view def _generate_instruction(self, category: str, context: Dict[str, Any] = None) -> str: """ 根据类别生成任务指令 :param category: 场景类别 :param context: 上下文信息 :return: 任务指令 """ base_instruction = self.instructions.get(category, "继续操作") # 构建完整指令，在开头添加app名称 app_name = self._get_app_name() if app_name: full_instruction = f"你将要操作的app是{app_name}。{base_instruction}" else: full_instruction = base_instruction # 如果有额外上下文，添加到指令中 if context and "additional_info" in context: full_instruction += f"。附加信息: {context['additional_info']}" self.logger.debug(f"[GuiAgent] 生成指令: {full_instruction}") return full_instruction def _get_app_name(self) -> str: """ 获取当前app的名称 :return: app名称 """ # 1. 优先使用初始化时传入的app_name if self.app_name: return self.app_name # 2. 尝试从apk文件名获取 if self.app and hasattr(self.app, 'app_path'): import os apk_path = self.app.app_path filename = os.path.basename(apk_path) if filename.endswith('.apk'): return filename[:-4].replace('_', ' ') return ""