UFO³ 设备代理 Strategy Layer 之 ProcessorTemplate 处理器框架深度解析策略编排、中间件与上下文管理【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO导读本文基于 UFO³UFO³: Weaving the Digital Agent Galaxy开源仓库中的 ProcessorLevel-2设计文档深入解析设备代理三层架构中Strategy LayerLevel-2的核心编排组件 ——ProcessorTemplate。读者将掌握处理器框架如何通过模板方法模式编排数据采集、LLM 推理、动作执行、记忆更新四大阶段如何注册策略与中间件如何通过依赖验证与本地/全局上下文分离保证复杂工作流可靠执行以及如何面向 Windows、Linux 等平台扩展出自定义处理器。文中的类与流程均已对照 processor_framework.py、processing_context.py、processing_middleware.py、app_agent_processor.py 等实际源码逐条印证。定位Strategy Layer 中的处理器框架在 UFO³ 的设备代理 三层架构 中Level-1 State LayerFSM决定何时、做什么状态迁移与代理交接Level-2 Strategy Layer执行逻辑决定如何做其中ProcessorTemplate负责将模块化策略编排为完整执行流程Level-3 Command Layer系统接口执行确定性系统操作命令分发与 MCP 工具。Strategy Layer 由四部分组成Processor 框架本主题、Processing Strategies见 strategy.md、中间件系统日志、指标、错误处理等横切关注点、依赖验证与上下文管理。设计哲学ProcessorTemplate遵循模板方法模式——process()定义工作流骨架子类通过_setup_strategies()/_setup_middleware()配置各阶段的策略与中间件策略与中间件均在初始化时注入从而实现不改动核心框架即可扩展。ProcessorTemplate 框架结构ProcessorTemplate是定义执行工作流的抽象基类平台专属处理器AppAgentProcessor、HostAgentProcessor、LinuxAgentProcessor通过继承它来配置各自平台的策略与中间件。源码位于 processor_framework.py。阶段枚举 ProcessingPhasefrom enum import Enum class ProcessingPhase(Enum): Enumeration of processor execution phases SETUP setup # Initialization (optional) DATA_COLLECTION data_collection # Gather context from device LLM_INTERACTION llm_interaction # LLM reasoning and decision ACTION_EXECUTION action_execution # Execute commands on device MEMORY_UPDATE memory_update # Update memory and blackboard CLEANUP cleanup # Cleanup (optional)实际源码中的定义与文档一致processing_context.py。抽象基类核心职责class ProcessorTemplate(ABC): Abstract processor template defining workflow orchestration framework. Responsibilities: 1. Strategy Registration: Configure strategies for each phase 2. Middleware Management: Setup cross-cutting concern handlers 3. Dependency Validation: Ensure strategy data flow is valid 4. Workflow Execution: Orchestrate strategy execution in phase order 5. Context Management: Create and manage ProcessingContext Subclasses must implement: - _setup_strategies(): Register strategies for processing phases - _setup_middleware(): Register middleware (optional) # Subclasses can override to use custom context class processor_context_class: Type[BasicProcessorContext] BasicProcessorContext def __init__(self, agent: BasicAgent, global_context: Context): self.agent agent self.global_context global_context self.strategies: Dict[ProcessingPhase, ProcessingStrategy] {} self.middleware_chain: List[ProcessorMiddleware] [] self.logger logging.getLogger(self.__class__.__name__) self.dependency_validator StrategyDependencyValidator() # 子类按阶段注册策略与中间件 self._setup_strategies() self._setup_middleware() # 创建本地/全局分离的处理上下文 self.processing_context self._create_processing_context() # 初始化即校验策略链依赖尽早暴露配置错误 self._validate_strategy_chain()子类必须实现的两个抽象方法abstractmethod def _setup_strategies(self) - None: Setup strategies for each processing phase. 示例 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ ScreenshotStrategy(), UITreeStrategy() ]) self.strategies[ProcessingPhase.LLM_INTERACTION] LLMStrategy() self.strategies[ProcessingPhase.ACTION_EXECUTION] ActionStrategy() self.strategies[ProcessingPhase.MEMORY_UPDATE] MemoryStrategy() pass def _setup_middleware(self) - None: Setup middleware for cross-cutting concerns. Default: no middleware. 示例 self.middleware_chain [ LoggingMiddleware(), MetricsMiddleware(), ErrorHandlingMiddleware() ] pass需要说明的源码差异文档中的示例代码为教学意图的简化示意实际仓库在初始化顺序上略有不同 —— 源码在_setup_strategies()/_setup_middleware()之后才创建processing_context并调用_validate_strategy_chain()且校验失败默认仅记录 warning 日志_validate_strategy_chain捕获异常并打印Strategy dependency issue而文档版本描述为直接抛出ProcessingException。读者以 processor_framework.py 的实际实现为准。上下文创建与公共数据注入源码为处理器上下文提供了更强的初始化支持processor_framework.pydef _get_common_context_data(self) - Dict[str, Any]: 所有处理器共有的本地上下文初始化数据 return { command_dispatcher: self.global_context.command_dispatcher, agent_name: self.agent.name, session_step: self.global_context.get(ContextNames.SESSION_STEP), round_step: self.global_context.get(ContextNames.CURRENT_ROUND_STEP), round_num: self.global_context.get(ContextNames.CURRENT_ROUND_ID), request: self.global_context.get(ContextNames.REQUEST), log_path: self.global_context.get(ContextNames.LOG_PATH), }子类可覆写_get_processor_specific_context_data()追加平台专属数据如AppAgentProcessor注入subtask、application_process_name、app_root见 app_agent_processor.py。_create_local_context()会尝试用合并后的数据构造上下文实例若目标上下文类不接受某些参数则依次降级为仅公共数据乃至空构造 手动 setattr保证框架健壮性。框架收益一致的工作流所有处理器遵循同一执行模式跨平台行为可预测平台定制化子类只配置策略不修改核心框架框架可复用编排逻辑被所有处理器共享消除重复代码中间件支持日志、指标、错误处理等横切关注点统一作用于每次策略执行可测试性每个阶段可用 mock 策略与 mock 上下文独立测试。策略注册Strategy Registration处理器通过为每个阶段注册策略来配置工作流。文档给出的注册关系如下例 1Windows AppAgent 处理器以下代码与 app_agent_processor.py 的实际实现一致含 fail_fast 配置from ufo.agents.processors.core.processor_framework import ProcessorTemplate, ProcessingPhase from ufo.agents.processors.strategies.processing_strategy import ComposedStrategy class AppAgentProcessor(ProcessorTemplate): Processor for Windows AppAgent (UI Automation) processor_context_class AppAgentProcessorContext # 自定义上下文类型 def _setup_strategies(self): Configure strategies for Windows UI automation workflow # Phase 1: DATA_COLLECTION - 组合多个策略 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy( strategies[ AppScreenshotCaptureStrategy(), # 截取应用截图 AppControlInfoStrategy() # 提取 UI Automation 树 ], nameAppDataCollectionStrategy, fail_fastTrue, ) # Phase 2: LLM_INTERACTION - 单策略LLM 失败应触发恢复 self.strategies[ProcessingPhase.LLM_INTERACTION] AppLLMInteractionStrategy( fail_fastTrue ) # Phase 3: ACTION_EXECUTION - 动作失败可优雅处理 self.strategies[ProcessingPhase.ACTION_EXECUTION] AppActionExecutionStrategy( fail_fastFalse ) # Phase 4: MEMORY_UPDATE - 记忆更新失败不应中断流程 self.strategies[ProcessingPhase.MEMORY_UPDATE] AppMemoryUpdateStrategy( fail_fastFalse ) def _setup_middleware(self): Configure middleware for logging and metrics self.middleware_chain [AppAgentLoggingMiddleware()]例 2Linux Agent 处理器class LinuxAgentProcessor(ProcessorTemplate): Processor for Linux Agent (Shell Commands) processor_context_class LinuxAgentProcessorContext def _setup_strategies(self): Configure strategies for Linux shell workflow # Phase 1: DATA_COLLECTION - 截图 shell 输出 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ CustomizedScreenshotCaptureStrategy(), ShellOutputStrategy() ]) # Phase 2: LLM_INTERACTION - 生成 shell 命令 self.strategies[ProcessingPhase.LLM_INTERACTION] CustomizedLLMInteractionStrategy() # Phase 3: ACTION_EXECUTION - 执行 shell 命令 self.strategies[ProcessingPhase.ACTION_EXECUTION] LinuxActionExecutionStrategy() # Phase 4: MEMORY_UPDATE - 记录命令历史 self.strategies[ProcessingPhase.MEMORY_UPDATE] LinuxMemoryUpdateStrategy()注册最佳实践需要多数据源的阶段如 DATA_COLLECTION使用ComposedStrategy职责单一集中的阶段如 LLM_INTERACTION使用单个策略除非需要初始化/清理资源否则不要注册 SETUP / CLEANUP 阶段通过覆写processor_context_class定义平台专属数据结构见 app_agent_processing_context.py 中AppAgentProcessorContext的截图路径、过滤控件、LLM 响应等字段。中间件系统Middleware System中间件为所有策略执行提供统一的横切关注点日志、指标、错误处理、缓存等在每个阶段执行前后以及出错时被调用。接口定义见 processing_middleware.py。ProcessorMiddleware 接口from abc import ABC, abstractmethod class ProcessorMiddleware(ABC): Abstract base for processor middleware. Middleware wraps strategy execution to provide cross-cutting concerns such as logging, metrics collection, error handling, caching, etc. def __init__(self, name: Optional[str] None): self.name name or self.__class__.__name__ abstractmethod async def before_process(self, processor: ProcessorTemplate, context: ProcessingContext) - None: Called before processing starts. pass abstractmethod async def after_process(self, processor: ProcessorTemplate, result: ProcessingResult) - None: Called after processing completes. pass abstractmethod async def on_error(self, processor: ProcessorTemplate, error: Exception) - None: Called when an error occurs during processing. pass内置中间件EnhancedLoggingMiddleware框架内置的EnhancedLoggingMiddleware提供完整的执行期日志processing_middleware.pyclass EnhancedLoggingMiddleware(ProcessorMiddleware): Enhanced logging middleware that handles different types of errors appropriately def __init__(self, log_level: int logging.INFO, name: Optional[str] None): super().__init__(name) self.logger logging.getLogger(f{self.__class__.__name__}.{self.name}) self.log_level log_level async def before_process(self, processor, context): Log processing start with context information round_num context.get(round_num, 0) round_step context.get(round_step, 0) self.logger.log( self.log_level, fStarting processing: Round {round_num 1}, Step {round_step 1}, fProcessor: {processor.__class__.__name__} ) async def after_process(self, processor, result): Log processing completion with result summary and save to file if result.success: self.logger.log( self.log_level, fProcessing completed successfully in {result.execution_time:.2f}s ) else: self.logger.warning(fProcessing completed with failure: {result.error}) # 把本地上下文总耗时、各阶段耗时写入日志文件 local_logger processor.processing_context.global_context.get(ContextNames.LOGGER) local_context processor.processing_context.local_context local_context.total_time result.execution_time phrase_time_cost {} for phrase, phrase_result in processor.processing_context.phase_results.items(): phrase_time_cost[phrase.name] phrase_result.execution_time local_context.execution_times phrase_time_cost safe_obj to_jsonable_python(local_context.to_dict(selectiveTrue)) local_logger.write(json.dumps(safe_obj, ensure_asciiFalse)) async def on_error(self, processor, error): Enhanced error logging with context information if isinstance(error, ProcessingException): self.logger.error( fProcessingException in {processor.__class__.__name__}:\n f Phase: {error.phase}\n Message: {str(error)}\n f Context: {error.context_data}\n f Original Exception: {error.original_exception} ) if error.original_exception: self.logger.info( fOriginal traceback:\n{traceback.format_exception(error.original_exception)} ) else: self.logger.error( fUnexpected error in {processor.__class__.__name__}: {str(error)}\n fError type: {type(error).__name__}\n fTraceback:\n{traceback.format_exception(error)} )关键特性上下文感知日志记录 round/step 信息便于溯源结果摘要记录总执行时间与各阶段耗时分解持久化日志将结构化上下文数据通过to_dict(selectiveTrue)选取关键字段序列化为 JSON 写入日志文件增强错误处理区分ProcessingException携带 phase、context_data、original_exception与普通异常完整堆栈捕获traceback.format_exception输出完整调用栈便于调试。AppAgentProcessor还提供了AppAgentLoggingMiddleware继承自EnhancedLoggingMiddleware在标准日志基础上叠加 rich 控制台面板展示Round/Step/Agent与starting_message()的启动信息、成功/失败彩色输出等见 app_agent_processor.py。中间件执行顺序处理前按顺序为每个中间件调用before_process()策略执行策略按阶段依次执行处理后按逆序为每个中间件调用after_process()对应源码 processor_framework.py 的reversed(self.middleware_chain)出错时对所有中间件调用on_error()。中间件收益关注点分离横切逻辑与策略逻辑解耦、可复用同一中间件跨处理器使用、非侵入增删中间件无需改动策略。工作流执行Workflow Executionprocess()是ProcessorTemplate的模板方法负责编排完整执行流程processor_framework.py。执行顺序# ProcessorTemplate.process() 内按 ProcessingPhase 枚举顺序执行 execution_order [ ProcessingPhase.SETUP, # Optional: Initialize resources ProcessingPhase.DATA_COLLECTION, # Gather device/environment context ProcessingPhase.LLM_INTERACTION, # LLM reasoning and decision-making ProcessingPhase.ACTION_EXECUTION, # Execute actions on device ProcessingPhase.MEMORY_UPDATE, # Update memory and blackboard ProcessingPhase.CLEANUP # Optional: Cleanup resources ]阶段执行规则可选阶段SETUP 与 CLEANUP 可选未注册策略则跳过顺序执行阶段按固定顺序串行执行不支持并行依赖验证执行前通过StrategyDependencyValidator校验源码中为_validate_strategy_dependencies_runtime见 processor_framework.py执行结果校验策略执行后还会调用_validate_strategy_provides_runtime检查声明的 provides与实际返回字段的一致性缺失或多余均记录 warning见 strategy_dependency.pyFail-Fast vs 继续执行由策略的fail_fast设置决定错误处理方式。当策略返回失败结果时处理器会构造异常并触发中间件on_error后终止对应源码break逻辑当策略抛出ProcessingException时外层 catch 会将其 phase 与 context_data 写入失败结果并再次触发on_error中间件上下文即时更新每个策略的输出通过update_local()立即写入ProcessingContext供下一策略读取阶段结果追踪每个阶段的ProcessingResult含成功标志、执行耗时、数据键、错误被记录在phase_results有序字典中最终随combined_result.data[phase_results]/phase_results_summary返回processing_context.py。上下文收尾finalize_finalize_processing_context()负责把本地数据提升到全局上下文累加 LLM 成本到CURRENT_ROUND_COST/SESSION_COST自增CURRENT_ROUND_STEP与SESSION_STEPprocessor_framework.py。子类可覆写以定制收尾逻辑。ProcessingContext本地与全局数据分离ProcessingContext为所有策略提供统一的数据访问分离本地处理器专属与全局会话级数据processing_context.py。dataclass class ProcessingContext: Processing context with local and global data separation. :param global_context: Global context (shared across all components) :param local_context: Local context (processor-specific data) global_context: Context # 模块系统全局上下文 local_context: BasicProcessorContext # 处理器本地数据 phase_results: OrderedDict[ProcessingPhase, ProcessingResult] field(default_factoryOrderedDict) # --- 主要接口直接类型安全的属性访问 --- def __getattr__(self, name): if hasattr(self.local_context, name): return getattr(self.local_context, name) raise AttributeError(...) # --- 向后兼容接口 --- def get_local(self, key: str, defaultNone) - Any: 先从本地上下文的字段取其次查 custom_data ... def get_global(self, key: str, defaultNone) - Any: key 自动转大写优先匹配 ContextNames 枚举否则查内部字典 ... def update_local(self, data: Dict[str, Any]) - None: 批量更新本地上下文 ... def require_local(self, field_name: str, expected_type: Type None) - Any: 获取必需字段缺失或类型不符时抛出 ProcessingException 并在 context_data 中附带 available_keys 便于排查。 ...上下文分离的设计依据全局上下文会话级、跨组件共享用户请求REQUEST会话 ID、round 号、step 号配置设置命令分发器引用command_dispatcher黑板Blackboard引用。本地上下文处理器专属、临时数据截图数据screenshot、screenshot_pathUI 控件信息control_infoLLM 解析响应parsed_response动作执行结果results临时处理数据。此外BasicProcessorContextprocessing_context.py统一承载agent_type、session_step/round_step/round_num、action、arguments、parsed_response、llm_cost、execution_times、custom_data等公共字段update_from_dict()对未知字段自动落入custom_data保证向后兼容。get_context_summary()提供面向日志/调试的摘要视图。平台专属处理器一览不同代理类型实现平台专属处理器PlatformProcessor ClassDATA_COLLECTIONLLM_INTERACTIONACTION_EXECUTIONMEMORY_UPDATEWindows AppAgentAppAgentProcessor截图 UI 树UI 元素选择UI Automation 命令UI 交互历史Windows HostAgentHostAgentProcessor桌面截图 应用列表应用选择启动应用、创建 AppAgent应用选择历史LinuxLinuxAgentProcessor截图 shell 输出Shell 命令生成Shell 命令执行命令历史仓库中实际存在的处理器与上下文实现包括app_agent_processor.py、host_agent_processor.py、customized_agent_processor.py以及 app_agent_processing_context.py、host_agent_processing_context.py。平台专属策略分别位于 strategies/app_agent_processing_strategy.py、strategies/host_agent_processing_strategy.py、strategies/linux_agent_strategy.py、strategies/mobile_agent_strategy.py。设计要点与最佳实践处理器设计指南1. 清晰的阶段划分每个阶段职责单一 —— DATA_COLLECTION 采集原始数据、LLM_INTERACTION 推理决策、ACTION_EXECUTION 执行命令、MEMORY_UPDATE 持久化状态。2. 合理的策略组合多源数据采集使用ComposedStrategy。实际实现processing_strategy.py会顺序执行组件策略、将成功策略的结果即时写入共享上下文供后续组件读取、自动聚合全部依赖与 provides 元数据并支持 fail-fast / 继续执行两种错误模式self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ AppScreenshotCaptureStrategy(), AppControlInfoStrategy() ])3. 中间件承载横切关注点不要在策略内部实现日志/指标逻辑。4. 依赖验证利用StrategyDependencyValidator自动校验。依赖可通过两种方式声明详见 strategy_dependency.py覆写get_dependencies()/get_provides()返回StrategyDependency列表使用depends_on(...)/provides(...)装饰器自动登记到StrategyMetadataRegistry供运行时校验与 provides 一致性检查使用。5. 自定义上下文类按平台定义专属上下文。仓库真实示例AppAgentProcessorContextapp_agent_processing_context.py包含subtask、app_root、application_process_name、多路径截图字段、filtered_controls、annotation_dict、response_text、prompt_message等应用专属数据dataclass class AppAgentProcessorContext(BasicProcessorContext): Extended context for Windows AppAgent agent_type: str AppAgent screenshot: str screenshot_path: str control_info: str control_elements: List[Dict] field(default_factorylist) parsed_response: Dict field(default_factorydict) action: List[Dict[str, Any]] field(default_factorylist) arguments: Dict field(default_factorydict) results: str !!! warning 常见陷阱 -跳过必需阶段不要跳过 DATA_COLLECTION → LLM → ACTION → MEMORY 主链路 -改变阶段顺序不要重排阶段会破坏依赖链 -策略内部存状态不要在执行期间修改策略实例状态请改用上下文 -直接修改代理属性不要在处理器中直接改 agent 属性应通过记忆系统等正规通道。与其它层的集成集成点层/组件关系AgentStateLevel-1 State状态调用processor.process()执行工作流ProcessingStrategyLevel-2 Strategy处理器注册并执行策略CommandDispatcherLevel-3 CommandACTION_EXECUTION 策略使用分发器下发命令Memory/Blackboard记忆系统MEMORY_UPDATE 策略更新代理记忆Global Context模块系统处理器通过上下文读取请求、写回结果相关文档状态层见 state.md策略层见 strategy.md命令层见 command.md记忆系统见 memory.md平台处理器总览见 agent_types.md。关键结论ProcessorTemplate工作流编排的抽象框架遵循模板方法模式策略注册通过_setup_strategies()配置阶段专属策略中间件系统日志、错误处理等横切关注点统一作用于所有策略执行工作流执行按 DATA_COLLECTION → LLM_INTERACTION → ACTION_EXECUTION → MEMORY_UPDATE 顺序编排并在执行前后分别校验依赖与 provides 一致性依赖验证StrategyDependencyValidator结合StrategyMetadataRegistry保证策略在所需数据就绪后才执行上下文管理本地处理器级与全局会话级数据分离require_local/get_global提供类型安全访问平台可扩展子类化即可创建平台专属处理器Windows / Linux / 移动端核心框架无需改动。Processor 作为 Strategy Layer 的编排核心通过协调策略执行、中间件应用与上下文管理支撑代理在多样平台上可靠、高效地执行复杂工作流。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 UFO³ 设备代理 Strategy Layer 之 ProcessorTemplate 处理器框架深度解析：策略编排、中间件与上下文管理
UFO³ 设备代理 Strategy Layer 之 ProcessorTemplate 处理器框架深度解析策略编排、中间件与上下文管理【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO导读本文基于 UFO³UFO³: Weaving the Digital Agent Galaxy开源仓库中的 ProcessorLevel-2设计文档深入解析设备代理三层架构中Strategy LayerLevel-2的核心编排组件 ——ProcessorTemplate。读者将掌握处理器框架如何通过模板方法模式编排数据采集、LLM 推理、动作执行、记忆更新四大阶段如何注册策略与中间件如何通过依赖验证与本地/全局上下文分离保证复杂工作流可靠执行以及如何面向 Windows、Linux 等平台扩展出自定义处理器。文中的类与流程均已对照 processor_framework.py、processing_context.py、processing_middleware.py、app_agent_processor.py 等实际源码逐条印证。定位Strategy Layer 中的处理器框架在 UFO³ 的设备代理 三层架构 中Level-1 State LayerFSM决定何时、做什么状态迁移与代理交接Level-2 Strategy Layer执行逻辑决定如何做其中ProcessorTemplate负责将模块化策略编排为完整执行流程Level-3 Command Layer系统接口执行确定性系统操作命令分发与 MCP 工具。Strategy Layer 由四部分组成Processor 框架本主题、Processing Strategies见 strategy.md、中间件系统日志、指标、错误处理等横切关注点、依赖验证与上下文管理。设计哲学ProcessorTemplate遵循模板方法模式——process()定义工作流骨架子类通过_setup_strategies()/_setup_middleware()配置各阶段的策略与中间件策略与中间件均在初始化时注入从而实现不改动核心框架即可扩展。ProcessorTemplate 框架结构ProcessorTemplate是定义执行工作流的抽象基类平台专属处理器AppAgentProcessor、HostAgentProcessor、LinuxAgentProcessor通过继承它来配置各自平台的策略与中间件。源码位于 processor_framework.py。阶段枚举 ProcessingPhasefrom enum import Enum class ProcessingPhase(Enum): Enumeration of processor execution phases SETUP setup # Initialization (optional) DATA_COLLECTION data_collection # Gather context from device LLM_INTERACTION llm_interaction # LLM reasoning and decision ACTION_EXECUTION action_execution # Execute commands on device MEMORY_UPDATE memory_update # Update memory and blackboard CLEANUP cleanup # Cleanup (optional)实际源码中的定义与文档一致processing_context.py。抽象基类核心职责class ProcessorTemplate(ABC): Abstract processor template defining workflow orchestration framework. Responsibilities: 1. Strategy Registration: Configure strategies for each phase 2. Middleware Management: Setup cross-cutting concern handlers 3. Dependency Validation: Ensure strategy data flow is valid 4. Workflow Execution: Orchestrate strategy execution in phase order 5. Context Management: Create and manage ProcessingContext Subclasses must implement: - _setup_strategies(): Register strategies for processing phases - _setup_middleware(): Register middleware (optional) # Subclasses can override to use custom context class processor_context_class: Type[BasicProcessorContext] BasicProcessorContext def __init__(self, agent: BasicAgent, global_context: Context): self.agent agent self.global_context global_context self.strategies: Dict[ProcessingPhase, ProcessingStrategy] {} self.middleware_chain: List[ProcessorMiddleware] [] self.logger logging.getLogger(self.__class__.__name__) self.dependency_validator StrategyDependencyValidator() # 子类按阶段注册策略与中间件 self._setup_strategies() self._setup_middleware() # 创建本地/全局分离的处理上下文 self.processing_context self._create_processing_context() # 初始化即校验策略链依赖尽早暴露配置错误 self._validate_strategy_chain()子类必须实现的两个抽象方法abstractmethod def _setup_strategies(self) - None: Setup strategies for each processing phase. 示例 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ ScreenshotStrategy(), UITreeStrategy() ]) self.strategies[ProcessingPhase.LLM_INTERACTION] LLMStrategy() self.strategies[ProcessingPhase.ACTION_EXECUTION] ActionStrategy() self.strategies[ProcessingPhase.MEMORY_UPDATE] MemoryStrategy() pass def _setup_middleware(self) - None: Setup middleware for cross-cutting concerns. Default: no middleware. 示例 self.middleware_chain [ LoggingMiddleware(), MetricsMiddleware(), ErrorHandlingMiddleware() ] pass需要说明的源码差异文档中的示例代码为教学意图的简化示意实际仓库在初始化顺序上略有不同 —— 源码在_setup_strategies()/_setup_middleware()之后才创建processing_context并调用_validate_strategy_chain()且校验失败默认仅记录 warning 日志_validate_strategy_chain捕获异常并打印Strategy dependency issue而文档版本描述为直接抛出ProcessingException。读者以 processor_framework.py 的实际实现为准。上下文创建与公共数据注入源码为处理器上下文提供了更强的初始化支持processor_framework.pydef _get_common_context_data(self) - Dict[str, Any]: 所有处理器共有的本地上下文初始化数据 return { command_dispatcher: self.global_context.command_dispatcher, agent_name: self.agent.name, session_step: self.global_context.get(ContextNames.SESSION_STEP), round_step: self.global_context.get(ContextNames.CURRENT_ROUND_STEP), round_num: self.global_context.get(ContextNames.CURRENT_ROUND_ID), request: self.global_context.get(ContextNames.REQUEST), log_path: self.global_context.get(ContextNames.LOG_PATH), }子类可覆写_get_processor_specific_context_data()追加平台专属数据如AppAgentProcessor注入subtask、application_process_name、app_root见 app_agent_processor.py。_create_local_context()会尝试用合并后的数据构造上下文实例若目标上下文类不接受某些参数则依次降级为仅公共数据乃至空构造 手动 setattr保证框架健壮性。框架收益一致的工作流所有处理器遵循同一执行模式跨平台行为可预测平台定制化子类只配置策略不修改核心框架框架可复用编排逻辑被所有处理器共享消除重复代码中间件支持日志、指标、错误处理等横切关注点统一作用于每次策略执行可测试性每个阶段可用 mock 策略与 mock 上下文独立测试。策略注册Strategy Registration处理器通过为每个阶段注册策略来配置工作流。文档给出的注册关系如下例 1Windows AppAgent 处理器以下代码与 app_agent_processor.py 的实际实现一致含 fail_fast 配置from ufo.agents.processors.core.processor_framework import ProcessorTemplate, ProcessingPhase from ufo.agents.processors.strategies.processing_strategy import ComposedStrategy class AppAgentProcessor(ProcessorTemplate): Processor for Windows AppAgent (UI Automation) processor_context_class AppAgentProcessorContext # 自定义上下文类型 def _setup_strategies(self): Configure strategies for Windows UI automation workflow # Phase 1: DATA_COLLECTION - 组合多个策略 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy( strategies[ AppScreenshotCaptureStrategy(), # 截取应用截图 AppControlInfoStrategy() # 提取 UI Automation 树 ], nameAppDataCollectionStrategy, fail_fastTrue, ) # Phase 2: LLM_INTERACTION - 单策略LLM 失败应触发恢复 self.strategies[ProcessingPhase.LLM_INTERACTION] AppLLMInteractionStrategy( fail_fastTrue ) # Phase 3: ACTION_EXECUTION - 动作失败可优雅处理 self.strategies[ProcessingPhase.ACTION_EXECUTION] AppActionExecutionStrategy( fail_fastFalse ) # Phase 4: MEMORY_UPDATE - 记忆更新失败不应中断流程 self.strategies[ProcessingPhase.MEMORY_UPDATE] AppMemoryUpdateStrategy( fail_fastFalse ) def _setup_middleware(self): Configure middleware for logging and metrics self.middleware_chain [AppAgentLoggingMiddleware()]例 2Linux Agent 处理器class LinuxAgentProcessor(ProcessorTemplate): Processor for Linux Agent (Shell Commands) processor_context_class LinuxAgentProcessorContext def _setup_strategies(self): Configure strategies for Linux shell workflow # Phase 1: DATA_COLLECTION - 截图 shell 输出 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ CustomizedScreenshotCaptureStrategy(), ShellOutputStrategy() ]) # Phase 2: LLM_INTERACTION - 生成 shell 命令 self.strategies[ProcessingPhase.LLM_INTERACTION] CustomizedLLMInteractionStrategy() # Phase 3: ACTION_EXECUTION - 执行 shell 命令 self.strategies[ProcessingPhase.ACTION_EXECUTION] LinuxActionExecutionStrategy() # Phase 4: MEMORY_UPDATE - 记录命令历史 self.strategies[ProcessingPhase.MEMORY_UPDATE] LinuxMemoryUpdateStrategy()注册最佳实践需要多数据源的阶段如 DATA_COLLECTION使用ComposedStrategy职责单一集中的阶段如 LLM_INTERACTION使用单个策略除非需要初始化/清理资源否则不要注册 SETUP / CLEANUP 阶段通过覆写processor_context_class定义平台专属数据结构见 app_agent_processing_context.py 中AppAgentProcessorContext的截图路径、过滤控件、LLM 响应等字段。中间件系统Middleware System中间件为所有策略执行提供统一的横切关注点日志、指标、错误处理、缓存等在每个阶段执行前后以及出错时被调用。接口定义见 processing_middleware.py。ProcessorMiddleware 接口from abc import ABC, abstractmethod class ProcessorMiddleware(ABC): Abstract base for processor middleware. Middleware wraps strategy execution to provide cross-cutting concerns such as logging, metrics collection, error handling, caching, etc. def __init__(self, name: Optional[str] None): self.name name or self.__class__.__name__ abstractmethod async def before_process(self, processor: ProcessorTemplate, context: ProcessingContext) - None: Called before processing starts. pass abstractmethod async def after_process(self, processor: ProcessorTemplate, result: ProcessingResult) - None: Called after processing completes. pass abstractmethod async def on_error(self, processor: ProcessorTemplate, error: Exception) - None: Called when an error occurs during processing. pass内置中间件EnhancedLoggingMiddleware框架内置的EnhancedLoggingMiddleware提供完整的执行期日志processing_middleware.pyclass EnhancedLoggingMiddleware(ProcessorMiddleware): Enhanced logging middleware that handles different types of errors appropriately def __init__(self, log_level: int logging.INFO, name: Optional[str] None): super().__init__(name) self.logger logging.getLogger(f{self.__class__.__name__}.{self.name}) self.log_level log_level async def before_process(self, processor, context): Log processing start with context information round_num context.get(round_num, 0) round_step context.get(round_step, 0) self.logger.log( self.log_level, fStarting processing: Round {round_num 1}, Step {round_step 1}, fProcessor: {processor.__class__.__name__} ) async def after_process(self, processor, result): Log processing completion with result summary and save to file if result.success: self.logger.log( self.log_level, fProcessing completed successfully in {result.execution_time:.2f}s ) else: self.logger.warning(fProcessing completed with failure: {result.error}) # 把本地上下文总耗时、各阶段耗时写入日志文件 local_logger processor.processing_context.global_context.get(ContextNames.LOGGER) local_context processor.processing_context.local_context local_context.total_time result.execution_time phrase_time_cost {} for phrase, phrase_result in processor.processing_context.phase_results.items(): phrase_time_cost[phrase.name] phrase_result.execution_time local_context.execution_times phrase_time_cost safe_obj to_jsonable_python(local_context.to_dict(selectiveTrue)) local_logger.write(json.dumps(safe_obj, ensure_asciiFalse)) async def on_error(self, processor, error): Enhanced error logging with context information if isinstance(error, ProcessingException): self.logger.error( fProcessingException in {processor.__class__.__name__}:\n f Phase: {error.phase}\n Message: {str(error)}\n f Context: {error.context_data}\n f Original Exception: {error.original_exception} ) if error.original_exception: self.logger.info( fOriginal traceback:\n{traceback.format_exception(error.original_exception)} ) else: self.logger.error( fUnexpected error in {processor.__class__.__name__}: {str(error)}\n fError type: {type(error).__name__}\n fTraceback:\n{traceback.format_exception(error)} )关键特性上下文感知日志记录 round/step 信息便于溯源结果摘要记录总执行时间与各阶段耗时分解持久化日志将结构化上下文数据通过to_dict(selectiveTrue)选取关键字段序列化为 JSON 写入日志文件增强错误处理区分ProcessingException携带 phase、context_data、original_exception与普通异常完整堆栈捕获traceback.format_exception输出完整调用栈便于调试。AppAgentProcessor还提供了AppAgentLoggingMiddleware继承自EnhancedLoggingMiddleware在标准日志基础上叠加 rich 控制台面板展示Round/Step/Agent与starting_message()的启动信息、成功/失败彩色输出等见 app_agent_processor.py。中间件执行顺序处理前按顺序为每个中间件调用before_process()策略执行策略按阶段依次执行处理后按逆序为每个中间件调用after_process()对应源码 processor_framework.py 的reversed(self.middleware_chain)出错时对所有中间件调用on_error()。中间件收益关注点分离横切逻辑与策略逻辑解耦、可复用同一中间件跨处理器使用、非侵入增删中间件无需改动策略。工作流执行Workflow Executionprocess()是ProcessorTemplate的模板方法负责编排完整执行流程processor_framework.py。执行顺序# ProcessorTemplate.process() 内按 ProcessingPhase 枚举顺序执行 execution_order [ ProcessingPhase.SETUP, # Optional: Initialize resources ProcessingPhase.DATA_COLLECTION, # Gather device/environment context ProcessingPhase.LLM_INTERACTION, # LLM reasoning and decision-making ProcessingPhase.ACTION_EXECUTION, # Execute actions on device ProcessingPhase.MEMORY_UPDATE, # Update memory and blackboard ProcessingPhase.CLEANUP # Optional: Cleanup resources ]阶段执行规则可选阶段SETUP 与 CLEANUP 可选未注册策略则跳过顺序执行阶段按固定顺序串行执行不支持并行依赖验证执行前通过StrategyDependencyValidator校验源码中为_validate_strategy_dependencies_runtime见 processor_framework.py执行结果校验策略执行后还会调用_validate_strategy_provides_runtime检查声明的 provides与实际返回字段的一致性缺失或多余均记录 warning见 strategy_dependency.pyFail-Fast vs 继续执行由策略的fail_fast设置决定错误处理方式。当策略返回失败结果时处理器会构造异常并触发中间件on_error后终止对应源码break逻辑当策略抛出ProcessingException时外层 catch 会将其 phase 与 context_data 写入失败结果并再次触发on_error中间件上下文即时更新每个策略的输出通过update_local()立即写入ProcessingContext供下一策略读取阶段结果追踪每个阶段的ProcessingResult含成功标志、执行耗时、数据键、错误被记录在phase_results有序字典中最终随combined_result.data[phase_results]/phase_results_summary返回processing_context.py。上下文收尾finalize_finalize_processing_context()负责把本地数据提升到全局上下文累加 LLM 成本到CURRENT_ROUND_COST/SESSION_COST自增CURRENT_ROUND_STEP与SESSION_STEPprocessor_framework.py。子类可覆写以定制收尾逻辑。ProcessingContext本地与全局数据分离ProcessingContext为所有策略提供统一的数据访问分离本地处理器专属与全局会话级数据processing_context.py。dataclass class ProcessingContext: Processing context with local and global data separation. :param global_context: Global context (shared across all components) :param local_context: Local context (processor-specific data) global_context: Context # 模块系统全局上下文 local_context: BasicProcessorContext # 处理器本地数据 phase_results: OrderedDict[ProcessingPhase, ProcessingResult] field(default_factoryOrderedDict) # --- 主要接口直接类型安全的属性访问 --- def __getattr__(self, name): if hasattr(self.local_context, name): return getattr(self.local_context, name) raise AttributeError(...) # --- 向后兼容接口 --- def get_local(self, key: str, defaultNone) - Any: 先从本地上下文的字段取其次查 custom_data ... def get_global(self, key: str, defaultNone) - Any: key 自动转大写优先匹配 ContextNames 枚举否则查内部字典 ... def update_local(self, data: Dict[str, Any]) - None: 批量更新本地上下文 ... def require_local(self, field_name: str, expected_type: Type None) - Any: 获取必需字段缺失或类型不符时抛出 ProcessingException 并在 context_data 中附带 available_keys 便于排查。 ...上下文分离的设计依据全局上下文会话级、跨组件共享用户请求REQUEST会话 ID、round 号、step 号配置设置命令分发器引用command_dispatcher黑板Blackboard引用。本地上下文处理器专属、临时数据截图数据screenshot、screenshot_pathUI 控件信息control_infoLLM 解析响应parsed_response动作执行结果results临时处理数据。此外BasicProcessorContextprocessing_context.py统一承载agent_type、session_step/round_step/round_num、action、arguments、parsed_response、llm_cost、execution_times、custom_data等公共字段update_from_dict()对未知字段自动落入custom_data保证向后兼容。get_context_summary()提供面向日志/调试的摘要视图。平台专属处理器一览不同代理类型实现平台专属处理器PlatformProcessor ClassDATA_COLLECTIONLLM_INTERACTIONACTION_EXECUTIONMEMORY_UPDATEWindows AppAgentAppAgentProcessor截图 UI 树UI 元素选择UI Automation 命令UI 交互历史Windows HostAgentHostAgentProcessor桌面截图 应用列表应用选择启动应用、创建 AppAgent应用选择历史LinuxLinuxAgentProcessor截图 shell 输出Shell 命令生成Shell 命令执行命令历史仓库中实际存在的处理器与上下文实现包括app_agent_processor.py、host_agent_processor.py、customized_agent_processor.py以及 app_agent_processing_context.py、host_agent_processing_context.py。平台专属策略分别位于 strategies/app_agent_processing_strategy.py、strategies/host_agent_processing_strategy.py、strategies/linux_agent_strategy.py、strategies/mobile_agent_strategy.py。设计要点与最佳实践处理器设计指南1. 清晰的阶段划分每个阶段职责单一 —— DATA_COLLECTION 采集原始数据、LLM_INTERACTION 推理决策、ACTION_EXECUTION 执行命令、MEMORY_UPDATE 持久化状态。2. 合理的策略组合多源数据采集使用ComposedStrategy。实际实现processing_strategy.py会顺序执行组件策略、将成功策略的结果即时写入共享上下文供后续组件读取、自动聚合全部依赖与 provides 元数据并支持 fail-fast / 继续执行两种错误模式self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ AppScreenshotCaptureStrategy(), AppControlInfoStrategy() ])3. 中间件承载横切关注点不要在策略内部实现日志/指标逻辑。4. 依赖验证利用StrategyDependencyValidator自动校验。依赖可通过两种方式声明详见 strategy_dependency.py覆写get_dependencies()/get_provides()返回StrategyDependency列表使用depends_on(...)/provides(...)装饰器自动登记到StrategyMetadataRegistry供运行时校验与 provides 一致性检查使用。5. 自定义上下文类按平台定义专属上下文。仓库真实示例AppAgentProcessorContextapp_agent_processing_context.py包含subtask、app_root、application_process_name、多路径截图字段、filtered_controls、annotation_dict、response_text、prompt_message等应用专属数据dataclass class AppAgentProcessorContext(BasicProcessorContext): Extended context for Windows AppAgent agent_type: str AppAgent screenshot: str screenshot_path: str control_info: str control_elements: List[Dict] field(default_factorylist) parsed_response: Dict field(default_factorydict) action: List[Dict[str, Any]] field(default_factorylist) arguments: Dict field(default_factorydict) results: str !!! warning 常见陷阱 -跳过必需阶段不要跳过 DATA_COLLECTION → LLM → ACTION → MEMORY 主链路 -改变阶段顺序不要重排阶段会破坏依赖链 -策略内部存状态不要在执行期间修改策略实例状态请改用上下文 -直接修改代理属性不要在处理器中直接改 agent 属性应通过记忆系统等正规通道。与其它层的集成集成点层/组件关系AgentStateLevel-1 State状态调用processor.process()执行工作流ProcessingStrategyLevel-2 Strategy处理器注册并执行策略CommandDispatcherLevel-3 CommandACTION_EXECUTION 策略使用分发器下发命令Memory/Blackboard记忆系统MEMORY_UPDATE 策略更新代理记忆Global Context模块系统处理器通过上下文读取请求、写回结果相关文档状态层见 state.md策略层见 strategy.md命令层见 command.md记忆系统见 memory.md平台处理器总览见 agent_types.md。关键结论ProcessorTemplate工作流编排的抽象框架遵循模板方法模式策略注册通过_setup_strategies()配置阶段专属策略中间件系统日志、错误处理等横切关注点统一作用于所有策略执行工作流执行按 DATA_COLLECTION → LLM_INTERACTION → ACTION_EXECUTION → MEMORY_UPDATE 顺序编排并在执行前后分别校验依赖与 provides 一致性依赖验证StrategyDependencyValidator结合StrategyMetadataRegistry保证策略在所需数据就绪后才执行上下文管理本地处理器级与全局会话级数据分离require_local/get_global提供类型安全访问平台可扩展子类化即可创建平台专属处理器Windows / Linux / 移动端核心框架无需改动。Processor 作为 Strategy Layer 的编排核心通过协调策略执行、中间件应用与上下文管理支撑代理在多样平台上可靠、高效地执行复杂工作流。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考