UFO 设备信息提供器（Device Info Provider）：星群场景下的智能设备发现与任务路由基石

UFO 设备信息提供器（Device Info Provider）：星群场景下的智能设备发现与任务路由基石

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

本篇技术指南系统讲解 UFO（UFO³: Weaving the Digital Agent Galaxy）客户端中的Device Info Provider模块：它如何在设备注册时自动采集系统信息并推送给服务器，如何基于平台能力做特征检测与设备分类，进而支撑星群（Constellation）多设备场景下的智能任务分配与设备选择。读完本文，你将掌握DeviceSystemInfo数据模型、DeviceInfoProvider的采集流程与优雅降级机制、psutil与 IP 探测等底层实现细节，以及如何将设备信息接入 WebSocket 注册流程与服务器端 AgentProfile，用于构建真实的设备注册与能力路由链路。

一、模块定位：注册即采集、路由零延迟

Device Info Provider 的核心设计理念是"注册时主动采集、注册中随包推送"：设备信息在客户端注册阶段被主动收集并随注册消息一起推送到服务器，而不是在任务到达时才临时查询。这一"推（Push）模型"显著降低了任务路由时的延迟，使服务器在设备上线瞬间即可依据其平台、硬件、网络与能力特征做出立即的任务分配决策。

从源码来看，该模块位于 ufo/client/device_info_provider.py，由两个核心类型构成：

• DeviceSystemInfo：一个轻量级的dataclass，承载全部设备描述信息，提供to_dict()用于序列化传输；
• DeviceInfoProvider：提供collect_system_info()静态入口及一系列私有静态采集方法。

官方文档（documents/docs/client/device_info.md）总结了该模块的五大核心能力：

在支持的平台上，桌面/笔记本类系统（Windows、Linux、macOS）为完整支持（Full Support），可检测 GUI、CLI、浏览器、文件系统、办公软件及应用类特征；移动端与 IoT 端处于规划状态（Planned），未来将检测触控、移动应用、传感器等能力。

二、数据模型：DeviceSystemInfo字段全解析

DeviceSystemInfo的设计目标是在最小化注册开销的前提下捕获必要信息。其类定义位于 ufo/client/device_info_provider.py：

@dataclass class DeviceSystemInfo: # Basic identification device_id: str platform: str # windows, linux, darwin, android, ios, web os_version: str # Hardware information (simplified) cpu_count: int memory_total_gb: float # Network information hostname: str ip_address: str # Capability information supported_features: List[str] = field(default_factory=list) # Platform type categorization platform_type: str = "computer" # computer, mobile, web, iot # Schema version for future compatibility schema_version: str = "1.0" # Custom metadata (optional, can be loaded from config) custom_metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization""" return asdict(self)

各字段的完整参考如下：

值得注意的实现细节：supported_features、platform_type、schema_version、custom_metadata均通过field(default_factory=...)设置了默认值，使DeviceSystemInfo可以被部分初始化（如异常降级场景下仅提供device_id），这与错误处理机制相辅相成。to_dict()直接基于dataclasses.asdict实现，保证序列化后的字典与字段一一对应。

单元测试 tests/unit/test_device_info_provider.py 验证了该数据类的行为：在仅传入部分字段时，schema_version默认取"1.0"（见test_device_system_info_creation），to_dict()输出完整字典结构（见test_to_dict）。

三、采集流程：从系统调用到结构化信息

3.1 自动采集入口

collect_system_info()是唯一公开入口，签名如下：

@staticmethod def collect_system_info( client_id: str, custom_metadata: Optional[Dict[str, Any]] = None ) -> DeviceSystemInfo

典型用法：

from ufo.client.device_info_provider import DeviceInfoProvider # Collect system information system_info = DeviceInfoProvider.collect_system_info( client_id="device_windows_001", custom_metadata=None # Or load from config ) # Result: DeviceSystemInfo object print(system_info.platform) # "windows" print(system_info.cpu_count) # 8 print(system_info.memory_total_gb) # 16.0 print(system_info.supported_features) # ["gui", "cli", "browser", ...] # Convert to dict for transmission device_dict = system_info.to_dict()

3.2 底层采集链路

从 ufo/client/device_info_provider.py 的源码看，采集过程对系统调用做了完整封装，每个采集方法独立 try/except：

采集完成的调用链为：collect_system_info()→ 并行执行各私有方法 → 构造DeviceSystemInfo。整个流程对应文档中的序列图：基础信息（platform、os_version）、硬件信息（cpu_count、memory）、网络信息（hostname、ip_address）三个分组并行收集，随后进入特征检测与平台分类阶段，最终返回结构化对象。

仓库还提供了可直接运行的演示脚本 tests/demo_device_info.py，它调用collect_system_info("demo_device_001", custom_metadata={"demo": True, "purpose": "testing"})并逐字段打印平台、版本、平台类型、CPU、内存、主机名、IP、Schema 版本、能力列表与自定义元数据，最后以json.dumps(..., indent=2)输出字典表示，是快速体验该模块的入口。

四、能力特征检测：基于平台的能力画像

4.1 各平台特征

_detect_features()（ufo/client/device_info_provider.py）根据platform.system()的结果返回能力列表：

Windows：

features = [ "gui", # Graphical user interface "cli", # Command line interface "browser", # Web browser support "file_system", # File system operations "office", # Office applications (Word, Excel, etc.) "windows_apps" # Windows-specific applications ]

Linux：

features = [ "gui", # Graphical user interface (X11/Wayland) "cli", # Bash/shell "browser", # Firefox, Chrome, etc. "file_system", # Linux file system "office", # LibreOffice, etc. "linux_apps" # Linux-specific applications ]

macOS：

features = [ "gui", # macOS GUI "cli", # Terminal "browser", # Safari, Chrome, etc. "file_system", # macOS file system "office" # Office for Mac ]

4.2 检测逻辑与判定依据

源码中的实际判定逻辑比文档示例更完整：桌面平台统一追加"gui"、"cli"、"browser"、"file_system"、"office"五个通用特征，随后按平台细分追加平台专属特征——Windows 追加"windows_apps"，Linux 追加"linux_apps"，macOS 追加"macos_apps"；而android/ios分支（未来移动端支持）会返回mobile_touch、mobile_apps、camera、gps等移动特征。

对应测试（tests/unit/test_device_info_provider.py 的test_detect_features_windows、test_detect_features_linux、test_detect_features_macos）分别 mock 了platform.system返回"Windows"、"Linux"、"Darwin"，验证windows_apps、linux_apps、macos_apps的追加逻辑。集成测试 tests/integration/test_device_info_flow.py 的test_multiple_devices_different_info还展示了 Windows/Linux/macOS 三台设备各自携带windows_apps、docker/kubernetes、macos_apps等特征注册后，服务器通过get_all_devices_info()可检索所有设备的差异化能力画像。

4.3 平台类型分类

_get_platform_type()将设备归类为computer（windows/linux/darwin）、mobile（android/ios）或unknown（其他），对应测试test_get_platform_type_computer验证了三类桌面平台均归为"computer"。

五、错误处理与优雅降级

Device Info Provider 的核心健壮性设计是：任何检测方法失败都不会让采集崩溃，而是返回最小可用信息。

try: # Attempt full collection return DeviceSystemInfo(...) except Exception as e: logger.error(f"Error collecting system info: {e}", exc_info=True) # Return minimal info on error return DeviceSystemInfo( device_id=client_id, platform="unknown", os_version="unknown", cpu_count=0, memory_total_gb=0.0, hostname="unknown", ip_address="unknown", supported_features=[], platform_type="unknown", custom_metadata=custom_metadata or {} )

该兜底逻辑在源码中真实存在（ufo/client/device_info_provider.py），且device_id与custom_metadata即使在异常路径也会被保留。单个方法的失败行为如下表：

单元测试test_collect_system_info_handles_errors用@patch("platform.system", side_effect=Exception(...))模拟了平台检测整体崩溃的场景，断言返回对象的platform == "unknown"、cpu_count == 0、memory_total_gb == 0.0，同时device_id保持传入值。

六、关键实现细节：内存检测与 IP 探测

6.1 内存检测依赖psutil（可选依赖）

内存检测依赖第三方库psutil，这是一个可选依赖——若未安装，内存将报告为0.0并输出 warning 日志，其余字段不受影响。

pip install psutil

检测代码（与文档示例一致，源码位于 ufo/client/device_info_provider.py）：

@staticmethod def _get_memory_total_gb() -> float: """Get total memory in GB""" try: import psutil total_memory = psutil.virtual_memory().total return round(total_memory / (1024**3), 2) # Convert to GB, round to 2 decimals except ImportError: logger.warning("psutil not installed, memory info unavailable") return 0.0 except Exception: return 0.0

注意两点实现细节：其一，单位换算使用1024**3（GiB 语义）并round(..., 2)保留两位小数；其二，import psutil被放在方法内部（函数级导入），从而把 psutil 与模块顶层解耦，保证未安装 psutil 时整个模块仍可正常导入。单元测试test_collect_system_info_success中 mock 的total = 17179869184字节正好等于 16 GB，验证了memory_total_gb == 16.0的换算结果。

6.2 IP 地址探测的两级策略

IP 检测采用两阶段方案以提升可靠性：

主方法（UDP Socket 连接）：

# Connect to external address (doesn't actually send data) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) # Google DNS ip = s.getsockname()[0] s.close()

该技巧利用 UDP 的"无连接"特性：connect()仅在本机路由表中建立目标条目，并不会真正发包，却能借助系统路由选出最合适的出网网卡，从而拿到本机在该网络下的真实 IP。注意它依赖外网可达（此处指向8.8.8.8），在内网或无外网环境下可能失败。

兜底方法（主机名解析）：

# If primary fails, resolve via hostname ip = socket.gethostbyname(socket.gethostname())

最终兜底：两级均失败则返回"unknown"。这一完整的三层逻辑在源码 ufo/client/device_info_provider.py 中通过嵌套 try/except 实现。

七、集成实战：从客户端注册到服务器路由

7.1 客户端：WebSocket 注册时的自动推送

Device Info Provider 被 UFO 的 WebSocket 客户端在register_client()阶段调用（ufo/client/websocket.py），实现"Push 模型"——设备信息随注册消息一次性上报：

# In websocket client's register_client() from ufo.client.device_info_provider import DeviceInfoProvider system_info = DeviceInfoProvider.collect_system_info( self.ufo_client.client_id, custom_metadata=None ) metadata = { "system_info": system_info.to_dict(), "registration_time": datetime.now(timezone.utc).isoformat() } await self.registration_protocol.register_as_device( device_id=self.ufo_client.client_id, metadata=metadata, platform=self.ufo_client.platform )

源码中的关键细节（与文档示例略有差异但语义一致）：

• 客户端侧刻意传custom_metadata=None，注释说明"服务器会在配置后补充自定义元数据"，即服务器配置是元数据的权威来源；
• 若采集抛异常，客户端不会中断注册，而是仅上报registration_time，保证注册流程的健壮性；
• 注册成功后会connected_event.set()通知上层，失败则抛出RuntimeError触发重连逻辑。

完整的注册流程可参考 WebSocket Client。

7.2 服务器：AgentProfile 存储与查询

服务器端在收到注册消息后，从metadata["system_info"]中提取设备信息并写入在线客户端注册表。以 ufo/server/services/client_connection_manager.py 为例：

# Server-side AgentProfile integration device_info = registration_data["metadata"]["system_info"] agent_profile.add_device(device_id, device_info)

ClientConnectionManager（服务器端对应类，兼容WSManager角色）提供了三个关键查询接口：

• get_device_system_info(device_id)：按 ID 查询单台设备的系统信息（client_connection_manager.py）；
• get_all_devices_info()：返回全部在线设备的能力画像字典（client_connection_manager.py）；
• _merge_device_info(system_info, server_config)：将设备自动采集的信息与服务器配置合并（client_connection_manager.py）。

合并策略值得展开：服务器配置整体并入custom_metadata；additional_features与设备自动检测的supported_features做并集去重；tags单独提升为顶层字段。服务器侧设备配置文件示例见 config/galaxy/devices.yaml，其metadata段（如os、performance、logs_file_path、dev_path、warning_log_pattern等）即典型的环境特定配置来源。此外_load_device_configs()同时支持 YAML 与 JSON 两种格式，按devices: {device_id: {...}}结构加载。

7.3 星群客户端按需查询设备信息

集成测试 tests/integration/test_device_info_flow.py 完整演示了另一条链路：星群（Constellation）客户端通过ClientMessageType.DEVICE_INFO_REQUEST请求某台设备的系统信息，服务器以ServerMessageType.DEVICE_INFO_RESPONSE返回result（含platform、cpu_count等字段）；当目标设备不存在时，服务器返回status == TaskStatus.ERROR且错误信息包含"not found"（见test_request_device_info_not_found）。test_device_info_with_server_config则验证了服务器配置（tags、tier、additional_features、max_concurrent_tasks）与设备自报信息合并后，custom_metadata["tier"] == "enterprise"、supported_features同时包含自报的gui与配置的advanced_automation。

服务器端的权限控制同样值得注意：ufo/server/ws/handler.py中对DEVICE_INFO_REQUEST做了函数级鉴权，仅允许 constellation 类型客户端发起，防止普通设备越权探测他人设备信息。服务器侧整体处理流程详见 Server Quick Start。

八、最佳实践清单

1. 为环境追踪添加自定义元数据

custom_meta = { "environment": os.getenv("ENVIRONMENT", "development"), "version": "1.0.0", "deployment_region": "us-west-2", "cost_center": "engineering" } system_info = DeviceInfoProvider.collect_system_info( client_id="device_001", custom_metadata=custom_meta )

2. 安装 psutil 以获得准确内存检测

pip install psutil

3. 使用描述性客户端 ID

# Include environment and location in client_id client_id = f"device_{platform}_{env}_{location}_{instance_id}" # Example: "device_windows_prod_us-west_001"

4. 记录采集结果

system_info = DeviceInfoProvider.collect_system_info(...) logger.info( f"Collected device info: " f"platform={system_info.platform}, " f"cpu={system_info.cpu_count}, " f"memory={system_info.memory_total_gb}GB, " f"features={system_info.supported_features}" )

5. 发送前校验关键字段

system_info = DeviceInfoProvider.collect_system_info(...) # Validate essential fields assert system_info.device_id, "Device ID required" assert system_info.platform != "unknown", "Platform detection failed" assert system_info.cpu_count > 0, "CPU detection failed"

6. 合理规划元数据权威性：客户端侧collect_system_info建议传custom_metadata=None，把环境标签、机房、运维联系信息等交给服务器配置文件管理，便于集中维护与更新。

九、延伸阅读

• WebSocket Client：了解设备信息如何在注册流程中发挥作用
• Quick Start：将你的设备连接到服务器
• MCP Integration：理解客户端工具能力
• Server Quick Start：学习服务器端注册处理
• 设备信息采集演示脚本：可直接运行的完整采集示例
• Device Info 单元测试 与 Device Info 集成测试：覆盖成功采集、元数据传递、异常降级、服务器合并与按需查询的完整验证链路

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