企业官网建设流程全解析

UFO 项目 MobileAgent 三状态机深度解析：CONTINUE / FINISH / FAIL 状态流转与 4 阶段处理流水线

【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO

本篇技术指南聚焦于 UFO 开源仓库中MobileAgent（Android 移动端任务执行代理）的有限状态机（FSM）设计：它以极简的 3 状态（CONTINUE、FINISH、FAIL）驱动移动端 UI 自动化任务的完整生命周期。读完本文，你将掌握 MobileAgent 状态机的枚举定义、状态注册与查找机制、基于 LLM 决策的状态转移规则、CONTINUE 状态下 4 阶段处理流水线的底层实现，以及状态机与 MCP Server、ADB 设备交互之间的协作方式，可直接用于理解或二次开发移动端智能代理的调度逻辑。

状态机架构概览

MobileAgent 使用一个3 状态有限状态机（FSM）来管理 Android 设备上的任务执行流程。极简的状态集合在保证执行进度可追踪的同时，维持了行为的简单性与可预测性。状态转移由LLM 的决策与动作执行结果共同驱动：LLM 在每一轮交互中返回下一个状态标识，Agent 据此推进或终止任务。

关联阅读：

• MobileAgent 架构总览：架构与核心职责
• 处理策略：CONTINUE 状态下的 4 阶段处理流水线
• MCP 命令：可用的移动端交互命令
• 快速开始指南：部署你的第一个移动端 Agent

状态枚举定义

状态机的状态集合定义在MobileAgentStatus枚举中，实际实现在 ufo/agents/states/mobile_agent_state.py：

class MobileAgentStatus(Enum): """ Store the status of the mobile agent. """ FINISH = "FINISH" CONTINUE = "CONTINUE" FAIL = "FAIL"

状态注册与查找机制

MobileAgent 的状态由MobileAgentStateManager管理，它实现了Agent 状态注册表模式（Agent State Registry Pattern）：

class MobileAgentStateManager(AgentStateManager): _state_mapping: Dict[str, Type[MobileAgentState]] = {} @property def none_state(self) -> AgentState: """ The none state of the state manager. """ return NoneMobileAgentState()

从源码实现（ufo/agents/states/basic.py）可以看出该模式的关键设计：

• 注册装饰器：所有 MobileAgent 状态类通过@MobileAgentStateManager.register装饰器注册，装饰器内部执行cls._state_mapping[state_class.name()] = state_class，将状态名映射到状态类；
• 单例管理器：AgentStateManager继承SingletonABCMeta（由SingletonMeta与ABCMeta组合而成），保证整个进程中只有唯一的状态管理器实例；
• 惰性实例化：get_state(status)在首次查询某状态时才创建状态对象并缓存到_state_instance_mapping，未注册的未知状态名会回退到none_state，保证了状态查找的健壮性。

状态转移图

整个状态机只有一条“执行主干”：任务从[*]启动后进入CONTINUE，在CONTINUE内部自循环迭代执行，直到 LLM 判定完成（→FINISH）或发生不可恢复错误（→FAIL）。两个终止态中，FAIL还会自动转移到FINISH以完成清理收尾流程。

状态定义详解

1. CONTINUE 状态（活跃执行态）

用途：MobileAgent 处理用户请求并执行移动端动作的核心活跃状态。

@MobileAgentStateManager.register class ContinueMobileAgentState(MobileAgentState): """The class for the continue mobile agent state""" async def handle(self, agent: "MobileAgent", context: Optional["Context"] = None): """Execute the 4-phase processing pipeline""" await agent.process(context) def is_round_end(self) -> bool: return False # Round continues def is_subtask_end(self) -> bool: return False # Subtask continues @classmethod def name(cls) -> str: return MobileAgentStatus.CONTINUE.value

在 MobileAgent 类定义 中可以看到，MobileAgent.__init__通过self.set_state(self.default_state)将初始状态设置为ContinueMobileAgentState，即 Agent 一经创建就进入活跃执行态；default_state属性直接返回ContinueMobileAgentState()实例。

CONTINUE 状态下的行为流程：

1. 数据收集阶段：

   • 截取设备屏幕截图
   • 获取已安装应用列表
   • 收集当前屏幕 UI 控件
   • 生成带控件 ID 标注的注解截图

2. LLM 交互阶段：

   • 构造包含截图与控件信息的提示词
   • 获取 LLM 的下一步动作决策
   • 解析并校验响应

3. 动作执行阶段：

   • 执行移动端动作（点击、滑动、输入文本、启动应用等）
   • 捕获执行结果

4. 记忆更新阶段：

   • 用截图和动作结果更新记忆
   • 存储控件信息供下一轮使用

5. 状态判定：

   • 分析 LLM 响应以确定下一个状态

状态转移逻辑：

• CONTINUE → CONTINUE：任务需要更多动作才能完成（例如需要跨多个屏幕导航）
• CONTINUE → FINISH：LLM 判定任务已完成（例如成功填写表单并提交）
• CONTINUE → FAIL：遇到不可恢复错误（例如必需应用未安装、多次尝试后仍找不到控件）

2. FINISH 状态（成功终止态）

用途：表示任务成功完成的状态。

@MobileAgentStateManager.register class FinishMobileAgentState(MobileAgentState): """The class for the finish mobile agent state""" def next_agent(self, agent: "MobileAgent") -> "MobileAgent": return agent def next_state(self, agent: "MobileAgent") -> MobileAgentState: return FinishMobileAgentState() # Remains in FINISH def is_subtask_end(self) -> bool: return True # Subtask completed def is_round_end(self) -> bool: return True # Round ends @classmethod def name(cls) -> str: return MobileAgentStatus.FINISH.value

行为特征：

• 向会话管理器（Session Manager）发出任务完成信号
• 不再进行任何后续处理
• Agent 实例可以被终止
• 截图与动作历史保留在记忆中可供查询

进入 FINISH 状态的条件：

• 所有必需移动端动作已成功执行
• LLM 判定用户请求已得到满足
• 目标 UI 状态已达成（如表单已提交、信息已展示）
• 执行过程中未出现错误或异常

3. FAIL 状态（错误终止态）

用途：表示因不可恢复错误导致任务失败的状态。

@MobileAgentStateManager.register class FailMobileAgentState(MobileAgentState): """The class for the fail mobile agent state""" def next_agent(self, agent: "MobileAgent") -> "MobileAgent": return agent def next_state(self, agent: "MobileAgent") -> MobileAgentState: return FinishMobileAgentState() # Transitions to FINISH for cleanup def is_round_end(self) -> bool: return True # Round ends def is_subtask_end(self) -> bool: return True # Subtask failed @classmethod def name(cls) -> str: return MobileAgentStatus.FAIL.value

行为特征：

• 记录失败原因与上下文
• 捕获最终截图用于调试
• 转移到 FINISH 状态完成清理
• 会话管理器收到失败状态

进入 FAIL 状态的典型条件：

• 应用不可用：必需应用未安装或无法启动
• 控件未找到：多次尝试后仍无法定位目标 UI 控件
• 设备断开：执行过程中 ADB 连接丢失
• 权限被拒：设备未授予所需权限
• 超时：动作执行耗时过长
• LLM 明确失败：LLM 明确表示任务无法完成
• 连续动作失败：多个连续动作执行失败

错误信息记录：

尽管 FAIL 是终止状态，错误信息仍会被记录用于事后调试：

# Example error logging in FAIL state agent.logger.error(f"Mobile task failed: {error_message}") agent.logger.debug(f"Last action: {last_action}") agent.logger.debug(f"Current screenshot saved to: {screenshot_path}") agent.logger.debug(f"UI controls at failure: {current_controls}")

状态转移规则

转移决策逻辑

状态转移由CONTINUE 状态中 LLM 的响应决定。LLM 返回的结构化响应中包含status字段，Agent 据此更新自身状态并通过状态管理器解析出下一个状态对象：

# LLM returns status in response parsed_response = { "action": { "function": "click_control", "arguments": {"control_id": "5", "control_name": "Search"}, "status": "CONTINUE" # or "FINISH" or "FAIL" }, "thought": "Need to click the search button to proceed" } # Agent updates its status based on LLM decision agent.status = parsed_response["action"]["status"] next_state = MobileAgentStateManager().get_state(agent.status)

在 mobile_agent_state.py 的MobileAgentState.next_state()基类方法中可以看到通用实现：读取agent.status，通过MobileAgentStateManager().get_state(status)按名称解析出对应的状态对象。这一设计将“LLM 返回的状态字符串”与“具体状态类实例”解耦，新增状态只需注册新类即可。

状态转移矩阵

CONTINUE 状态的处理流水线

4 阶段处理管线

处于 CONTINUE 状态时，MobileAgent 执行完整的 4 阶段流水线：

源码级流水线实现

这 4 个阶段在 ufo/agents/processors/strategies/mobile_agent_strategy.py 中由多个策略类协作完成，并通过@depends_on/@provides装饰器显式声明阶段间的数据依赖：

• MobileScreenshotCaptureStrategy：通过命令分发器向 MCP Server 发送capture_screenshot数据收集命令，将返回的 base64 截图保存为action_step{session_step}.png文件（路径为{log_path}action_step{session_step}.png），并记录截图耗时；
• MobileAppsCollectionStrategy：调用get_mobile_app_target_info命令（默认include_system_apps=False，即默认不包含系统应用）获取已安装应用列表，将返回的TargetInfo/字典统一转换为{id, name, package}结构供提示词使用，获取失败时降级为空列表并记录警告；
• MobileControlsCollectionStrategy：调用get_app_window_controls_target_info命令获取当前屏幕控件，对每个控件的rect（bbox 格式[left, top, right, bottom]）进行合法性校验（要求right > left且bottom > top），过滤无效控件后通过PhotographerFacade生成带控件标注框的注解截图action_step{session_step}_annotated.png；
• MobileLLMInteractionStrategy：将安装应用、当前控件、干净截图、注解截图、黑板（blackboard）上下文与历史成功动作一并组装进提示词，调用 LLM 并解析为结构化响应；
• MobileActionExecutionStrategy：解析parsed_response.action并执行移动端动作，将执行结果绑定回动作对象，生成ActionCommandInfo列表用于记忆追踪。

这一阶段化的策略设计（详见 处理策略文档）正是 CONTINUE 状态await agent.process(context)调用的底层支撑。

终止状态（FINISH / FAIL）

终止状态不执行任何处理逻辑：

• FINISH：干净终止，结果与截图可从记忆获取
• FAIL：错误终止，错误详情与最终截图被记录

确定性控制流设计

3 状态设计保证了执行过程的确定性与可追踪性：

• 行为可预测：每条执行路径都有明确定义
• 可调试：状态转移伴随截图记录，支持可视化调试
• 可测试：有限状态空间简化了测试覆盖
• 可维护：简洁的状态集合降低复杂度
• 可视化可追踪：每个状态的截图构成完整视觉执行历史

与其他 Agent 的状态机对比

MobileAgent 的极简 3 状态设计体现了其聚焦的职责范围：通过执行移动端 UI 动作来满足用户请求。与 LinuxAgent 类似，它在保持极简状态集合的同时提供了视觉上下文支持，简化状态机既避免了不必要的复杂性，又保留了健壮的错误处理与完成检测能力（对比可参考 overview.md 中的详细对照表）。

移动端特有设计考量

基于截图的状态追踪

与 LinuxAgent（纯文本）或 AppAgent（Windows UI API）不同，MobileAgent 高度依赖截图来理解 UI 状态：

• 每个 CONTINUE 轮次都以一张全新的截图开始
• 注解截图标注控件 ID，支持精确交互
• 截图存入记忆，用于调试与分析
• 视觉上下文帮助 LLM 理解当前 UI 状态

控件缓存机制

MobileAgent 缓存控件信息以降低 ADB 开销。该机制在 ufo/client/mcp/http_servers/mobile_mcp_server.py 的 MCP Server 中实现：

# Cache expiration times (seconds) self.apps_cache_duration = 300 # 5 minutes for apps list self.controls_cache_duration = 5 # 5 seconds for screen controls self.ui_tree_cache_duration = 5 # 5 seconds for UI tree self.device_info_cache_duration = 60 # 1 minute for device info

核心要点：

• 控件缓存时长为5 秒，超过即视为过期并重新抓取
• 每次动作执行后控件缓存失效（UI 很可能已变化）
• 控件字典支持按 ID 快速查找
• 显著减少重复的 UI 树解析开销

触控式交互

MobileAgent 的状态推进由触控动作触发，而非键盘命令：

• Tap（点击）：主要交互方式
• Swipe（滑动）：滚动与手势操作
• Type（输入）：文本输入（需要控件处于聚焦状态）
• Long-press（长按）：呼出上下文菜单（规划中）

实现细节与源码索引

状态机核心实现位于：

ufo/agents/states/mobile_agent_state.py

关键类（均在 ufo/agents/states/mobile_agent_state.py 中）：

• MobileAgentStatus：状态枚举（CONTINUE、FINISH、FAIL）
• MobileAgentStateManager：状态注册表与查找
• MobileAgentState：抽象基类，定义handle、next_agent、next_state、is_round_end、is_subtask_end等接口
• ContinueMobileAgentState：活跃执行状态，承载 4 阶段流水线
• FinishMobileAgentState：成功完成状态
• FailMobileAgentState：错误终止状态
• NoneMobileAgentState：初始/未定义状态（其name()返回空字符串，next_state直接指向 FINISH）

状态机在整个 MobileAgent 生态中的协作链路可概括为：MobileAgent（定义于 ufo/agents/agent/customized_agent.py，通过AgentRegistry.register(agent_name="MobileAgent", third_party=True, processor_cls=MobileAgentProcessor)注册）→ContinueMobileAgentState.handle()→agent.process(context)→ 各策略类 → MCP Server（ADB 交互）。仓库中的集成测试 tests/integration/test_mobile_mcp_server.py 演示了如何配置 MobileAgent 的数据收集与动作 MCP Server 并验证命令分发流程。

下一步学习

• 处理策略：深入理解 CONTINUE 状态中执行的 4 阶段处理流水线
• MCP 命令：探索移动端 UI 交互与应用管理命令
• 架构总览：返回 MobileAgent 架构全景
• 快速开始：部署你的第一个移动端 Agent
• 作为 Galaxy 设备使用：在多设备协同工作流中配置 MobileAgent

【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO