839 lines
40 KiB
Python
839 lines
40 KiB
Python
"""
|
||
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 ""
|
||
|