UFO Command Dispatcher 深度解析：Agent 决策与本地/远程执行的桥梁

UFO Command Dispatcher 深度解析：Agent 决策与本地/远程执行的桥梁

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

UFO 的 Command Dispatcher（命令调度器）位于 Agent 决策引擎与真实执行环境之间，负责把 Agent 生成的Command列表路由到本地 MCP 工具服务器或远端 WebSocket 客户端，并统一管理结果回收、超时与异常处理。本文基于仓库中的 dispatcher 技术文档 与 dispatcher.py 源码 展开，覆盖抽象基类设计、两条执行路径（本地/远程）的完整调用链、错误处理策略、常用执行模式、超时配置与排障手段，帮助你理解并复用这一核心模块。

设计概览：基于命令模式的双路径调度架构

调度器系统实现了经典命令模式（Command Pattern），配合asyncio异步执行与全面的异常兜底。核心思路是：Agent 只负责"决策"（决定做什么），调度器负责"执行"（知道怎么做、去哪做）。

从 ufo/module/dispatcher.py 的源码结构可以清晰地看到三个层次：

层次	类	职责
抽象基类	BasicCommandDispatcher	定义调度器统一接口：execute_commands()与generate_error_results()
本地执行	LocalCommandDispatcher	通过CommandRouter→ComputerManager→ MCP Server 在本机直接执行工具调用
远程执行	WebSocketCommandDispatcher	通过 AIP 协议的TaskExecutionProtocol将命令发给远端客户端执行

两条路径最终都把执行结果收敛为统一的Result对象列表返回给 Agent，无论命令是在本机跑还是在千里之外的设备上跑，Agent 看到的都是同样的结果模型。

快速参考

• 本地执行（交互式会话、独立会话）：使用LocalCommandDispatcher
• 远程控制（服务会话、设备 Agent）：使用WebSocketCommandDispatcher
• 统一异常处理：generate_error_results()
• 自定义调度逻辑：继承BasicCommandDispatcher

BasicCommandDispatcher：所有调度器的统一契约

BasicCommandDispatcher是一个ABC抽象基类（见 dispatcher.py），它把"命令进来、结果出去"的接口固化为两个方法。

抽象方法 execute_commands()

async def execute_commands( self, commands: List[Command], timeout: float = 6000 ) -> Optional[List[Result]]

参数与返回值：

参数	类型	默认值	说明
commands	List[Command]	必填	待执行的命令列表
timeout	float	6000	等待结果的超时秒数

返回值：

• List[Result]：命令执行结果列表
• None：执行超时

必须由具体调度器实现，不同平台/传输方式需给出各自的执行逻辑。

错误兜底方法 generate_error_results()

def generate_error_results( self, commands: List[Command], error: Exception ) -> Optional[List[Result]]

当执行过程中抛出任何异常时，该方法会把异常翻译成结构化的失败结果。核心逻辑（对应 dispatcher.py）是：遍历每一个命令，为每个命令生成一个ResultStatus.FAILURE的结果，保证返回列表与命令列表一一对应，Agent 可以按索引zip对齐处理：

result_list = [] for command in commands: error_msg = f"Error occurred while executing command {command}: {error}, please retry or execute a different command." result = Result( status=ResultStatus.FAILURE, error=error_msg, result=error_msg, call_id=command.call_id, ) result_list.append(result) return result_list

最终生成的错误结果形如：

from aip.messages import Result, ResultStatus error_result = Result( status=ResultStatus.FAILURE, error="ConnectionRefusedError: [WinError 10061]", result="Error occurred while executing command click_element: " "ConnectionRefusedError, please retry or execute a different command.", call_id="cmd_12345" ) # Agent 侧检查 if result.status == ResultStatus.FAILURE: print(f"Action failed: {result.error}") # Agent 可以重试或换一种方式

LocalCommandDispatcher：本机 MCP 工具直连执行

LocalCommandDispatcher面向交互式会话（interactive）与独立会话（standalone），它把命令直接路由到本机的 MCP 工具服务器上执行，无需网络传输。

初始化与内部组件

from ufo.module.dispatcher import LocalCommandDispatcher from ufo.client.mcp.mcp_server_manager import MCPServerManager def _init_context(self) -> None: """Initialize context with local dispatcher.""" super()._init_context() # Create MCP server manager mcp_server_manager = MCPServerManager() # Create local dispatcher command_dispatcher = LocalCommandDispatcher( session=self, mcp_server_manager=mcp_server_manager ) # Attach to context self.context.attach_command_dispatcher(command_dispatcher)

参数	类型	用途
session	BaseSession	当前会话实例
mcp_server_manager	MCPServerManager	MCP 服务器生命周期管理器

构造时（dispatcher.py）内部会自动创建两个关键组件（源码中采用懒导入避免循环依赖）：

• ComputerManager：管理计算机级操作，按agent_name::process_name::root_name三元组缓存并复用Computer实例；
• CommandRouter：负责把命令路由到合适的 MCP 工具。

本地执行调用链

execute_commands()首先为每个命令分配call_id = str(uuid.uuid4())（与远程路径一致），然后通过asyncio.wait_for以超时保护方式调用CommandRouter.execute(...)：

action_results = await asyncio.wait_for( self.command_router.execute( agent_name=self.session.current_agent_class, root_name=self.session.context.get(ContextNames.APPLICATION_ROOT_NAME), process_name=self.session.context.get(ContextNames.APPLICATION_PROCESS_NAME), commands=commands, ), timeout=timeout, )

路由上下文来源（定义于 context.py 的 ContextNames）：

上下文	来源	用途
agent_name	session.current_agent_class	追踪是哪个 Agent 发出的命令
root_name	context.APPLICATION_ROOT_NAME	用于 UI 操作的应用根名称
process_name	context.APPLICATION_PROCESS_NAME	目标进程名
commands	命令列表	待执行动作

CommandRouter.execute()（见 ufo/client/computer.py）进一步做了几件重要的事：

1. 通过computer_manager.get_or_create(...)获取（或惰性创建）对应的Computer实例；
2. 对没有tool_name的命令直接返回"未采取任何动作"的成功结果；
3. 支持early_exit=True短路：一旦前面有命令失败，后续命令被标记为ResultStatus.SKIPPED并跳过执行；
4. 调用computer.command2tool(command)把Command转成MCPToolCall，再交给Computer.run_actions()执行；
5. 每个命令之间await asyncio.sleep(0.1)限速，避免瞬间压垮服务器。

在Computer内部，MCP 工具调用通过线程池（ThreadPoolExecutor(max_workers=10)）隔离执行，防止阻塞型工具（如time.sleep）卡住主事件循环导致 WebSocket 断连；工具注册遵循tool_type::tool_name的键格式，tool_type分为data_collection与action两个命名空间。

本地执行示例

from aip.messages import Command, ResultStatus # 本地执行命令 commands = [ Command( tool_name="click_element", parameters={"control_label": "1", "button": "left"}, tool_type="windows", # 路由到 Windows MCP server call_id="" # 将自动分配 ), Command( tool_name="type_text", parameters={"text": "Hello World"}, tool_type="windows", call_id="" ) ] # 本地执行 results = await context.command_dispatcher.execute_commands( commands=commands, timeout=30.0 ) # 处理结果 for i, result in enumerate(results): if result.status == ResultStatus.SUCCESS: print(f"Command {i+1} succeeded: {result.result}") else: print(f"Command {i+1} failed: {result.error}")

注意：Command.tool_type在当前仓库 aip/messages.py 中被类型约束为Literal["data_collection", "action"]，文档示例中的"windows"仅为示意，实际使用需传入这两个合法值之一。

本地错误场景

错误类型	触发条件	处理方式	结果
TimeoutError	执行超过timeout	generate_error_results()	带超时信息的错误结果
ConnectionError	MCP 服务器不可达	generate_error_results()	带连接错误信息的结果
ValidationError	命令参数非法	generate_error_results()	带校验错误信息的结果
RuntimeError	工具执行失败	generate_error_results()	带执行错误信息的结果

WebSocketCommandDispatcher：基于 AIP 协议的远程执行

WebSocketCommandDispatcher面向服务会话（service session）与远程控制场景，它借助AIP（Agent Interaction Protocol）协议把命令封装为ServerMessage，通过 WebSocket 发给远端客户端执行，再用asyncio.Future挂起等待回包。

初始化与协议依赖

from ufo.module.dispatcher import WebSocketCommandDispatcher from aip.protocol.task_execution import TaskExecutionProtocol def _init_context(self) -> None: """Initialize context with WebSocket dispatcher.""" super()._init_context() # Create WebSocket dispatcher with AIP protocol command_dispatcher = WebSocketCommandDispatcher( session=self, protocol=self.task_protocol # TaskExecutionProtocol instance ) # Attach to context self.context.attach_command_dispatcher(command_dispatcher)

参数	类型	用途
session	BaseSession	当前服务会话
protocol	TaskExecutionProtocol	AIP 协议处理器

WebSocketCommandDispatcher强制要求TaskExecutionProtocol实例：若传入None会直接抛出ValueError（见 dispatcher.py）。

该调度器还维护pending: Dict[str, asyncio.Future]（response_id → Future 的映射）与容量为 100 的send_queue；从源码注释可以确认，发送工作已交由 AIP 传输层处理，不再需要独立的_send_loop观察者任务。

消息构造：make_server_response()

def make_server_response(self, commands: List[Command]) -> ServerMessage: """ Create a server response message for the given commands. """ # Assign unique IDs for command in commands: command.call_id = str(uuid.uuid4()) # Extract context agent_name = self.session.current_agent_class process_name = self.session.context.get(ContextNames.APPLICATION_PROCESS_NAME) root_name = self.session.context.get(ContextNames.APPLICATION_ROOT_NAME) session_id = self.session.id response_id = str(uuid.uuid4()) # Build AIP message return ServerMessage( type=ServerMessageType.COMMAND, status=TaskStatus.CONTINUE, agent_name=agent_name, process_name=process_name, root_name=root_name, actions=commands, session_id=session_id, task_name=self.session.task, timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), response_id=response_id )

ServerMessage 字段说明（消息模型定义于 aip/messages.py）：

字段	来源	用途
type	ServerMessageType.COMMAND	标记为命令消息
status	TaskStatus.CONTINUE	任务进行中
agent_name	当前 Agent 类名	追踪命令发出者
process_name	上下文	目标进程
root_name	上下文	应用根名称
actions	命令列表	待执行命令
session_id	会话 ID	会话跟踪
task_name	会话任务	任务标识
timestamp	当前 UTC 时间	消息时序
response_id	UUID	请求/响应关联

远程执行调用链

execute_commands()的流程（dispatcher.py）：

1. 调用make_server_response()构造消息并生成response_id；
2. 用事件循环创建Future，以response_id为键存入pending字典；
3. 调用protocol.send_command(server_message)发送——TaskExecutionProtocol.send_command()（见 aip/protocol/task_execution.py）内部委托给传输层send_message()，并记录发送的命令数量日志；
4. 发送失败：从pending弹出该 response_id，返回错误结果；
5. 发送成功：await asyncio.wait_for(fut, timeout)等待远端回包；
6. 超时：asyncio.TimeoutError被捕获，调用generate_error_results()返回错误结果；
7. finally中无论成败都清理pending条目，避免内存泄漏。

结果回填：set_result()

远端客户端执行完毕后通过 WebSocket 返回ClientMessage，由 WebSocket handler 调用set_result()回填 Future：

async def set_result(self, response_id: str, result: ClientMessage) -> None: """ Called by WebSocket handler when client returns a message. :param response_id: The ID of the response. :param result: The result from the client. """ fut = self.pending.get(response_id) if fut and not fut.done(): fut.set_result(result.action_results)

pending Future 管理机制：

• 请求侧：execute_commands创建 Future → 存入pending字典 →await等待；
• 响应侧：WebSocket 收到结果 → 按response_id查 Future →set_result()解析等待中的协程。

远程执行示例

from aip.messages import Command # Session 是 ServiceSession，内置 WebSocketCommandDispatcher commands = [ Command( tool_name="capture_window_screenshot", parameters={}, tool_type="data_collection" ) ] # 通过 WebSocket 远程执行 results = await context.command_dispatcher.execute_commands( commands=commands, timeout=60.0 # 截图可能耗时较长 ) # 结果来自远端客户端 if results: screenshot_base64 = results[0].result # 处理截图...

远程错误场景

错误类型	触发条件	处理方式	结果
TimeoutError	客户端未及时响应	generate_error_results()	错误结果
ProtocolError	AIP 协议违规	generate_error_results()	错误结果
ConnectionError	WebSocket 断连	generate_error_results()	错误结果
ClientError	客户端报告执行失败	原样返回客户端的错误 Result	透传客户端错误

WebSocket 路径的注意事项：网络延迟要在超时上加缓冲；客户端可能正忙于其他任务；连接丢失需要实现重连逻辑；AIP 协议保证消息有序投递。

统一错误处理机制

两条执行路径共享同一套错误哲学：把一切异常都翻译成结构化的Result对象，让 Agent 的失败处理路径保持统一。

错误流程图核心逻辑：

1. 命令执行开始，进入 try 块；
2. 成功 → 直接返回结果列表；
3. 失败 → 判断是否为超时（asyncio.TimeoutError），否则视为其他异常；
4. 两种情况都进入generate_error_results()；
5. 为每个命令创建Result：status = FAILURE，填充错误信息与call_id；
6. 返回错误结果列表。

最终序列化后的错误结果形如：

{ "status": "failure", "error": "asyncio.TimeoutError: Command execution timed out", "result": "Error occurred while executing command <Command>: TimeoutError, " "please retry or execute a different command.", "call_id": "cmd_abc123" }

Agent 侧的失败处理范式

async def execute_action(self, context: Context) -> None: """Execute action with error handling.""" commands = self.generate_commands() results = await context.command_dispatcher.execute_commands( commands=commands, timeout=30.0 ) for command, result in zip(commands, results): if result.status == ResultStatus.FAILURE: # Log error self.logger.error(f"Command {command.tool_name} failed: {result.error}") # Decision logic if "timeout" in result.error.lower(): # Retry with longer timeout self.retry_count += 1 if self.retry_count < 3: return await self.execute_action(context) elif "connection" in result.error.lower(): # Switch to alternative approach return self.fallback_strategy() else: # Escalate to error state self.transition_to_error_state(result.error) else: # Process successful result self.process_result(result.result)

错误处理最佳实践：

• ✅ 使用result.result之前先检查result.status
• ✅ 记录带上下文的错误日志（命令、参数、错误信息）
• ✅ 对瞬时错误实现重试逻辑
• ✅ 对永久性失败提供备用策略
• ✅ 为用户提供有帮助的错误提示
• ❌ 不要忽略错误结果
• ❌ 不要假设所有命令都会成功
• ❌ 不要无退避地无限重试

调度器与 Session/Context 的装配关系

调度器通过Context.attach_command_dispatcher()（ufo/module/context.py）挂载到会话上下文中，Agent 统一通过context.command_dispatcher.execute_commands(...)调用（如 basic.py 中的截图捕获即通过该入口执行）。各会话类型在各自的_init_context()中选择调度器（见 ufo/module/sessions/ 目录）：

会话文件	使用的调度器	场景
session.py	LocalCommandDispatcher	Windows 交互式/独立会话
linux_session.py	LocalCommandDispatcher	Linux 本地会话
mobile_session.py	LocalCommandDispatcher	移动端本地会话
service_session.py	WebSocketCommandDispatcher	服务模式（服务器-客户端通信）

从源码可见 Linux / 移动端会话还各自提供了基于WebSocketCommandDispatcher的 service 模式分支，实现了"本地直连 + 远程服务"的双模切换。

使用模式

模式 1：顺序执行

一条一条执行命令，每条命令的结果决定下一条命令：

for command in command_list: results = await context.command_dispatcher.execute_commands( commands=[command], timeout=30.0 ) if results[0].status == ResultStatus.SUCCESS: # Process result and decide next command next_command = self.decide_next_action(results[0]) else: # Handle error and possibly abort break

模式 2：批量执行

将相互关联、彼此无依赖的命令打包提交：

# 一个子任务下的全部命令 commands = [ Command(tool_name="click_element", ...), Command(tool_name="type_text", ...), Command(tool_name="press_key", ...) ] results = await context.command_dispatcher.execute_commands( commands=commands, timeout=60.0 ) # Process all results for command, result in zip(commands, results): if result.status == ResultStatus.FAILURE: # One failure might invalidate the whole subtask self.handle_subtask_failure(command, result)

模式 3：条件执行

根据前序结果决定后续动作（如先读 UI 树再决策）：

# Check state first check_cmd = Command(tool_name="get_ui_tree", ...) check_results = await dispatcher.execute_commands([check_cmd]) if check_results[0].status == ResultStatus.SUCCESS: ui_tree = check_results[0].result # Decide action based on UI state if "Login" in ui_tree: action_cmd = Command(tool_name="click_element", parameters={"label": "Login"}) else: action_cmd = Command(tool_name="type_text", parameters={"text": "username"}) # Execute decided action await dispatcher.execute_commands([action_cmd])

模式 4：指数退避重试

import asyncio async def execute_with_retry( dispatcher, commands, max_retries=3, base_delay=1.0 ): """Execute commands with exponential backoff retry.""" for attempt in range(max_retries): results = await dispatcher.execute_commands(commands, timeout=30.0) # Check if all succeeded all_success = all(r.status == ResultStatus.SUCCESS for r in results) if all_success: return results # Not last attempt - retry with backoff if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) logger.warning(f"Retry attempt {attempt + 1} after {delay}s") await asyncio.sleep(delay) # All retries exhausted return results # Return last attempt results

超时配置与性能考量

按操作类型选择超时

操作类型	推荐超时	理由
UI 点击	10-30s	快速，但可能等待动画
文本输入	5-15s	通常很快
截图	30-60s	可能需要渲染时间
文件操作	60-120s	依赖 I/O
网络调用	120-300s	网络延迟 + 处理
批量操作	单项之和 + 20%	计入额外开销

何时批量、何时不批量

适合批量：

• ✅ 同一上下文中的关联动作（如填写表单字段）
• ✅ 彼此无依赖的命令
• ✅ 全部命令指向同一应用

不适合批量：

• ❌ 存在依赖的命令（需要顺序执行）
• ❌ 快慢操作混杂（一个超时拖累全部）
• ❌ 需要中间结果来决策下一步

资源管理

# Good: Reuse dispatcher attached to context results1 = await context.command_dispatcher.execute_commands(commands1) results2 = await context.command_dispatcher.execute_commands(commands2) # Bad: Creating new dispatchers dispatcher1 = LocalCommandDispatcher(session, mcp_manager) dispatcher2 = LocalCommandDispatcher(session, mcp_manager)

调度器应作为会话级单例复用（挂在Context上），不要频繁新建——ComputerManager内部按三元组缓存Computer实例，重复创建会浪费 MCP 服务器注册成本。

高级主题：自定义调度器

继承BasicCommandDispatcher可实现自定义执行逻辑，例如记录所有命令与结果：

from ufo.module.dispatcher import BasicCommandDispatcher from aip.messages import Command, Result, ResultStatus from typing import List, Optional class CustomCommandDispatcher(BasicCommandDispatcher): """ Custom dispatcher that logs all commands and results. """ def __init__(self, session, log_file: str): self.session = session self.log_file = log_file async def execute_commands( self, commands: List[Command], timeout: float = 6000 ) -> Optional[List[Result]]: """Execute with logging.""" # Log commands with open(self.log_file, 'a') as f: f.write(f"Executing {len(commands)} commands\n") for cmd in commands: f.write(f" {cmd.tool_name}: {cmd.parameters}\n") try: # Your custom execution logic here results = await self.custom_execute(commands, timeout) # Log results with open(self.log_file, 'a') as f: for result in results: f.write(f" Result: {result.status}\n") return results except Exception as e: # Log error with open(self.log_file, 'a') as f: f.write(f" ERROR: {e}\n") return self.generate_error_results(commands, e) async def custom_execute( self, commands: List[Command], timeout: float ) -> List[Result]: """Implement custom execution logic.""" # Your implementation here pass

按会话类型选择调度器

from ufo.module.dispatcher import LocalCommandDispatcher, WebSocketCommandDispatcher def attach_appropriate_dispatcher(session, context): """Attach correct dispatcher based on session type.""" if isinstance(session, ServiceSession): # Service session uses WebSocket dispatcher = WebSocketCommandDispatcher( session=session, protocol=session.task_protocol ) else: # Interactive session uses local execution mcp_manager = MCPServerManager() dispatcher = LocalCommandDispatcher( session=session, mcp_server_manager=mcp_manager ) context.attach_command_dispatcher(dispatcher)

排障指南

问题：命令超时

症状：

• 命令持续超时
• 日志中出现asyncio.TimeoutError
• 返回带超时信息的错误结果

排查：

# Check timeout value results = await dispatcher.execute_commands(commands, timeout=30.0) # Enable debug logging logging.getLogger('ufo.module.dispatcher').setLevel(logging.DEBUG)

解决方案：

1. 为慢操作调大超时
2. 检查 MCP 服务器健康状态（本地调度器）
3. 验证 WebSocket 连接（WebSocket 调度器）
4. 把大批量拆成小批次

问题：连接错误

症状：

• 连接被拒绝
• WebSocket 断连
• MCP 服务器无响应

排查：

# For LocalCommandDispatcher # Check MCP server status mcp_manager.check_server_health() # For WebSocketCommandDispatcher # Check WebSocket connection if protocol.is_connected(): print("WebSocket connected") else: print("WebSocket disconnected")

解决方案：

1. 重启 MCP 服务器
2. 重连 WebSocket
3. 检查防火墙/网络设置
4. 确认客户端正在运行

问题：使用了错误的调度器

症状：

• 命令路由错误
• 服务会话中调用了 MCP 工具
• 本地会话中出现 WebSocket 消息

排查：

# Check dispatcher type print(type(context.command_dispatcher)) # Should be LocalCommandDispatcher or WebSocketCommandDispatcher # Check session type print(type(session))

解决方案：确保在会话_init_context()中正确初始化调度器（参考上文"调度器与 Session/Context 的装配关系"一节）。

与底层依赖的关系

Command Dispatcher 位于 UFO 架构的中枢位置，与以下模块协作：

• Context / Session：调度器实例通过 Context.attach_command_dispatcher() 挂载，随会话生命周期复用，详见 infrastructure/modules/context.md 与 infrastructure/modules/session.md；
• AIP 协议：远程路径依赖 aip/protocol/task_execution.py 与 aip/messages.py 中的消息模型，协议全景见 aip/overview.md；
• MCP 集成：本地路径依赖 ufo/client/computer.py 与 ufo/client/mcp/ 目录，工具服务器配置见 config/ufo/mcp.yaml（其中每个 Agent 按data_collection/action两组命名空间声明工具服务器，支持 local/stdio 与 http 两种类型），集成方式见 mcp/overview.md；
• 测试印证：execute_commands的超时与发送失败路径在 tests/integration/test_device_communication.py 中有直接覆盖（分别验证asyncio.TimeoutError与send_command抛错时返回错误结果）。

总结

Command Dispatcher 通过"抽象基类 + 双实现"的结构，把 UFO 中 Agent 决策与命令执行解耦：LocalCommandDispatcher依托CommandRouter→ComputerManager→ MCP Server 完成本机工具直调，WebSocketCommandDispatcher依托 AIPTaskExecutionProtocol+asyncio.Future完成跨设备远程调度；两者共享generate_error_results()的统一异常翻译、Result结果模型与超时策略。理解这一模块，即可掌握 UFO 从"Agent 决定做什么"到"命令真正被谁、在哪里、如何执行"的完整链路，也能在自定义会话或新执行后端时快速接入。

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