数据截至 (上游 commit 96983c73ed09)
第 4 章 · 它怎么「动手」
这章讲什么: 模型回了一句
{"function": "click_input", "arguments": {"id": "7"}},这句话怎么变成屏幕上真实的一次点击。答案链路比想象中长,但每一节都有它存在的理由。
4.1 要解决的小问题
直觉上,拿到函数名和参数,直接 getattr(controls[7], "click")() 就完了。UFO 不这么干,因为它同时要满足四个约束:
| 约束 | 直接调用为什么不行 |
|---|---|
| 同一套上层代码,既能操作本机也能操作远程机器 | 直接调用绑死在本进程 |
| 不同应用要有不同的工具集(Word 能调 COM,记事本不能) | 直接调用没法按应用切换 |
| 工具描述要能自动喂进 prompt | 手写函数没有机器可读的 schema |
| 换平台(Linux/Android)不改上层 | 直接调用绑死在 pywinauto |
解法是把所有动作统一成 MCP 工具调用。
4.2 全景:一次点击的完整路径
怎么读这张图:从上到下,每一层只做一件事,层间只传数据结构。
模型 JSON: {"function":"click_input","arguments":{"id":"7","name":"保存"}}
│ _action_to_command()
▼
① Command 信封 { tool_name, parameters, tool_type, call_id }
│
▼
② CommandDispatcher 本地版直接调 / WebSocket 版发出去等回来
│
▼
③ CommandRouter → ComputerManager → Computer
按「agent 名 + 应用 root 名」挑出一台虚拟"电脑"
│ command2tool():查工具注册表
▼
④ MCP 工具调用(fastmcp Client,跑在独立线程)
│
▼
⑤ 具体实现:AppPuppeteer → Receiver → pywinauto / Word COM
│
▼
⑥ Result 信封 { status, result, error, call_id } 原路返回
4.3 ① Command 信封
统一信封定义在 aip/messages.py:149 Command:
| 字段 | 含义 |
|---|---|
tool_name | 工具名,如 click_input |
parameters | 参数字典 |
tool_type | 只能是 "data_collection" 或 "action" |
call_id | 调用 id,用来把 Result 配回来 |
转换只有一行(_action_to_command,ufo/agents/processors/strategies/app_agent_processing_strategy.py:1444-1454)。
返回的 Result(aip/messages.py:183)带一个三值状态:SUCCESS / FAILURE / SKIPPED。第三个值是给「前一条失败后跳过后续」用的,下面会讲。
关键点: 从这一层往下,已经完全看不出这是 GUI 操作还是别的什么。信封是中立的。
4.4 ② 两种 dispatcher,一个接口
BasicCommandDispatcher 只有一个抽象方法 execute_commands(ufo/module/dispatcher.py:25-41)。两个实现:
LocalCommandDispatcher | WebSocketCommandDispatcher | |
|---|---|---|
| 位置 | ufo/module/dispatcher.py:67 | ufo/module/dispatcher.py:134 |
| 怎么执行 | 直接 await command_router.execute(...) | 打包成 ServerMessage 经 AIP 发出,等 future |
| 用在哪 | 单机模式 | 服务端驱动远程设备 |
| 超时 | asyncio.wait_for,默认 6000 秒 | 同上,外加协议层重连 |
远程版的配对逻辑:发出去时把 response_id 记进 self.pending 字典对应一个 future,WebSocket 收到回包时由 set_result 把结果塞进去(:206-252)。
两者出错时都调同一个 generate_error_results,把异常转成一条给模型看的失败结果(:43-64):
error_msg = f"Error occurred while executing command {command}: {error}, please retry or execute a different command."
注意那句尾巴。 错误信息是写给模型读的,直接建议它「重试或换个命令」。这是把异常处理当成 prompt 工程的一部分。
4.5 ③ Computer:一台虚拟电脑 = 一套工具集
缓存键是三元组
ComputerManager.get_or_create 用 agent_name::process_name::root_name 做 key(ufo/client/computer.py:612-670)。也就是说:
同一个 agent 在不同应用里,拿到的是不同的 Computer 实例、不同的工具集。
工具从哪来:那张路由表
配置在 config/ufo/mcp.yaml。结构是三层:Agent 名 → 应用 root 名 → { data_collection: [...], action: [...] }。
摘几行看形状:
| Agent | 应用 root | data_collection | action |
|---|---|---|---|
HostAgent | default | UICollector | HostUIExecutor、CommandLineExecutor |
AppAgent | default | UICollector | AppUIExecutor、CommandLineExecutor |
AppAgent | WINWORD.EXE | UICollector | AppUIExecutor、WordCOMExecutor |
AppAgent | EXCEL.EXE | UICollector | AppUIExecutor、ExcelCOMExecutor |
AppAgent | explorer.exe | UICollector | AppUIExecutor、PDFReaderExecutor |
ConstellationAgent | default | —(无) | ConstellationEditor |
LinuxAgent | default | — | BashExecutor(HTTP,8010 端口) |
MobileAgent | default | MobileDataCollector(HTTP) | MobileActionExecutor(HTTP) |
找不到对应 root 名就退回 default(ufo/client/computer.py:640-647)。
这张表是整个框架的可扩展性总开关。 加一个应用专属能力,不改一行 Python:写个 MCP 服务器,在表里加两行。
注意最后三行:Galaxy 的「改 DAG」和 Linux/Android 的能力,跟「点 Windows 按钮」在这张表里是平级的。
工具键带命名空间
工具注册表的键是 tool_type::tool_name(make_tool_key,ufo/client/computer.py:562-570)。所以「采集类的 get_ui_tree」和「动作类的 get_ui_tree」可以共存不打架。
command2tool 在 tool_type 缺失时会依次去两个命名空间里找,都找不到才报错(:512-560)。
工具描述怎么进 prompt
agent 初始化时会发一条特殊命令 list_tools,把当前这台「电脑」的工具清单要回来,转成 MCPToolInfo 交给 prompter 拼成 API 段(AppAgent._load_mcp_context,ufo/agents/agent/app_agent.py:496-530;HostAgent 同理,host_agent.py:283-311)。
list_tools 本身是个「元工具」——不走 MCP 服务器,由 Computer 直接处理(_run_action 里的 meta tool 分支,ufo/client/computer.py:170-190)。
4.6 ④ MCP 调用:线程隔离
Computer._run_action 没有直接 await MCP 客户端,而是把整个调用丢进线程池,并在线程里新开一个事件循环(ufo/client/computer.py:198-230):
# 示意,非源码 —— 演示线程隔离的形状
def _call_tool_in_thread():
loop = asyncio.new_event_loop() # 这个线程自己的循环
asyncio.set_event_loop(loop)
async def _do_call():
async with Client(server) as client:
return await client.call_tool(name=tool_name, arguments=params,
raise_on_error=False)
return loop.run_until_complete(_do_call())
result = await asyncio.wait_for(
loop.run_in_executor(self._executor, _call_tool_in_thread),
timeout=self._tool_timeout) # 6000 秒
源码注释说明了理由:MCP 工具里有阻塞操作(比如 time.sleep),直接跑会卡住主事件循环,进而导致 WebSocket 断连(:206-208)。
这是一个很现实的工程妥协:GUI 自动化天生就有大量同步阻塞代码,与其把它们全改成异步,不如整体隔离到线程池(10 个 worker,:64-66)。
4.7 早退与失败传播
CommandRouter.execute 默认 early_exit=True(ufo/client/computer.py:693-780)。一旦某条命令失败,后续命令不执行,而是各自返回一个 SKIPPED 结果。
为什么不直接抛异常? 因为模型下一步要看到「第 1 条成功、第 2 条失败、第 3 条被跳过」这个完整图景才能正确纠错。全部丢掉或者只报一个异常,模型就不知道做到哪儿了。
对多动作序列(模型一次回好几个动作),这条规则尤其重要——中间一步点空了,后面几步大概率也是错的,继续做只会让屏幕状态更乱。
4.8 ⑤ 落地层:命令模式三件套
MCP 工具函数内部,最终调的是 AppPuppeteer(ufo/automator/puppeteer.py:22)。这里是一个标准的命令模式:
| 角色 | 类 | 职责 |
|---|---|---|
| Invoker(调用者) | AppPuppeteer | 按命令名找到接收者,造命令对象,执行 |
| Receiver(接收者) | ControlReceiver / WordWinCOMReceiver / … | 真正干活的对象 |
| Command(命令) | CommandBasic 子类 | 一个动作一个类,带 execute / undo |
注册也是装饰器式的(ufo/automator/basic.py:52-60):
# 真实写法,见 ufo/automator/app_apis/word/wordclient.py:204
@WordWinCOMReceiver.register
class InsertTableCommand(WinCOMCommand):
def execute(self): ...
@classmethod
def name(cls) -> str: return "insert_table"
ReceiverManager 维护「命令名 → 接收者」的映射(ufo/automator/puppeteer.py:184-247),接收者又分两类:
- UI 接收者:每次动作前用当前控件重新创建(
create_ui_control_receiver)。 - API 接收者:按应用 root 名一次性创建(
create_api_receiver),内部持有 COM 对象。
4.9 混合动作:点鼠标 vs 调 API
这是 UFO 相对同类项目最实在的差异点。
同一件事,Word 里有两条路:
| 任务 | GUI 路径 | COM API 路径 |
|---|---|---|
| 插入 3×4 表格 | 点「插入」标签 → 点「表格」→ 拖选格子 → 确认 | 一次 insert_table(rows=3, columns=4) |
| 选中某段文字 | 找到位置 → 按住拖选 | 一次 select_text(text="...") |
| 另存为 PDF | 点文件 → 另存为 → 选格式 → 输文件名 → 确认 | 一次 save_as(...) |
COM 路径的实现在 ufo/automator/app_apis/word/wordclient.py(WordWinCOMReceiver,:12),对应 的 MCP 工具在 ufo/client/mcp/local_servers/word_wincom_mcp_server.py(create_word_mcp_server,:59)。
模型不需要知道两者的区别。 它看到的只是工具清单里多了几个函数,而这几个函数只在打开 Word 时才出现——因为 mcp.yaml 里 WINWORD.EXE 那一节多挂了 WordCOMExecutor。
配置里还有个细节:COM 类的服务器都标了 reset: true(config/ufo/mcp.yaml:64、:80、:96),意思是切换到新应用时重建这个服务器——因为它持有的 COM 对象绑定着具体文档。UI 类的标 reset: false,可以复用。
4.10 id + name 双保险
要解决的小问题
模型可能报错编号:说要点「保存」,给的却是 7 号,而 7 号其实是「另存为」。
做法:让模型同时报编号和名字,程序对照
每个控件类工具都要求两个必填参数:id 和 name(ufo/client/mcp/local_servers/ui_mcp_server.py:285-320 的 click_input)。
_verify_id 检查这个 id 对应控件的真实名字跟模型给的 name 是否一致(:170-188),但不一致时不拒绝执行:
# 真实逻辑,见 ufo/client/mcp/local_servers/ui_mcp_server.py:312-327
control_verified = _verify_id(id, name, ui_state.control_dict)
result = _execute_action(action)
if control_verified:
return result
else:
true_name = ui_state.control_dict.get(id).element_info.name
return (f"Warning: The name of your chosen control id {id} is {true_name}, "
f"but the name argument is {name}. The action is performed on "
f"control {id}:{true_name}.")
这个设计取舍很值得琢磨: 以 id 为准执行(id 是程序发的、可信),但把不一致明确告诉模型。模型下一步看到「你以为点的是保存,其实点的是另存为」,就能自己纠偏。
拒绝执行反而更糟——屏幕没变化,模型不知道发生了什么,大概率原样重试。
4.11 执行前的控件有效性检查
ActionExecutor.execute 在真正动手前会验一次控件是否 enabled 且 visible(ufo/automator/action_execution.py:85-133)。不通过就抛一个带指引的异常:
... is not available or not interactable for the action ..., please refresh the application state to get the latest interactable control information.
跟 4.4 节一样,异常文本是写给模型看的。这句话会顺着 Result.error 一路回到下一轮 prompt 里。
4.12 HostAgent 的动作:选应用也是一次工具调用
HostAgent 的动作只有一个:选应用。但它有两种目标(HostActionExecutionStrategy._execute_application_selection,ufo/agents/processors/strategies/host_agent_processing_strategy.py:826-870):
| 目标 kind | 走哪条分支 | 效果 |
|---|---|---|
third_party_agent | _select_third_party_agent | 记录「这个子任务派给第三方 agent」,不碰窗口 |
| 其他(window) | _select_regular_application | 发 select_application_window 命令,聚焦窗口 |
真正的窗口聚焦发生在 MCP 侧(ufo/client/mcp/local_servers/ui_mcp_server.py:192-247):set_focus(),按配置可选 maximize() 和 draw_outline()(红框提示,由 SHOW_VISUAL_OUTLINE_ON_SCREEN 控制)。选中后返回应用的 root 名,这个名字会决定下一步 AppAgent 拿到哪套工具——闭环就在这里合上了。
4.13 代码地图
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 统一信封定义 | aip/messages.py | Command、Result、ResultStatus、MCPToolInfo、MCPToolCall |
| 动作 → 信封 | ufo/agents/processors/strategies/app_agent_processing_strategy.py | AppActionExecutionStrategy._action_to_command |
| 两种 dispatcher | ufo/module/dispatcher.py | BasicCommandDispatcher、LocalCommandDispatcher、WebSocketCommandDispatcher |
| 错误结果生成(写给模型看) | ufo/module/dispatcher.py | BasicCommandDispatcher.generate_error_results |
| 虚拟电脑与工具注册表 | ufo/client/computer.py | Computer、Computer.command2tool、Computer.make_tool_key |
| 线程隔离的 MCP 调用 | ufo/client/computer.py | Computer._run_action |
| 按 agent+应用选电脑 | ufo/client/computer.py | ComputerManager.get_or_create |
| 命令路由与早退 | ufo/client/computer.py | CommandRouter.execute |
| MCP 服务器管理 | ufo/client/mcp/mcp_server_manager.py | MCPServerManager、LocalMCPServer、HTTPMCPServer |
| MCP 工厂注册表 | ufo/client/mcp/mcp_registry.py | MCPRegistry.register_factory_decorator |
| GUI 工具服务器 | ufo/client/mcp/local_servers/ui_mcp_server.py | create_app_action_mcp_server、create_host_action_mcp_server、_verify_id、UIServerState |
| Word COM 工具服务器 | ufo/client/mcp/local_servers/word_wincom_mcp_server.py | create_word_mcp_server |
| 命令模式三件套 | ufo/automator/basic.py | ReceiverBasic、CommandBasic、ReceiverFactory |
| 调用者与接收者管理 | ufo/automator/puppeteer.py | AppPuppeteer、ReceiverManager |
| Word COM 接收者与命令 | ufo/automator/app_apis/word/wordclient.py | WordWinCOMReceiver、InsertTableCommand、SaveAsCommand |
| 执行前控件校验 | ufo/automator/action_execution.py | ActionExecutor.execute、_control_validation |
| HostAgent 选应用 | ufo/agents/processors/strategies/host_agent_processing_strategy.py | HostActionExecutionStrategy._execute_application_selection |
| 路由配置表 | config/ufo/mcp.yaml | (配置文件) |