UFO3 設(shè)備代理 Strategy Layer 之 ProcessorTemplate 處理器框架深度解析策略編排、中間件與上下文管理【免費(fèi)下載鏈接】UFOUFO3: Weaving the Digital Agent Galaxy項(xiàng)目地址: https://gitcode.com/GitHub_Trending/uf/UFO導(dǎo)讀本文基于 UFO3UFO3: Weaving the Digital Agent Galaxy開源倉庫中的 ProcessorLevel-2設(shè)計(jì)文檔深入解析設(shè)備代理三層架構(gòu)中Strategy LayerLevel-2的核心編排組件 ——ProcessorTemplate。讀者將掌握處理器框架如何通過模板方法模式編排數(shù)據(jù)采集、LLM 推理、動(dòng)作執(zhí)行、記憶更新四大階段如何注冊(cè)策略與中間件如何通過依賴驗(yàn)證與本地/全局上下文分離保證復(fù)雜工作流可靠執(zhí)行以及如何面向 Windows、Linux 等平臺(tái)擴(kuò)展出自定義處理器。文中的類與流程均已對(duì)照 processor_framework.py、processing_context.py、processing_middleware.py、app_agent_processor.py 等實(shí)際源碼逐條印證。定位Strategy Layer 中的處理器框架在 UFO3 的設(shè)備代理 三層架構(gòu) 中Level-1 State LayerFSM決定何時(shí)、做什么狀態(tài)遷移與代理交接Level-2 Strategy Layer執(zhí)行邏輯決定如何做其中ProcessorTemplate負(fù)責(zé)將模塊化策略編排為完整執(zhí)行流程Level-3 Command Layer系統(tǒng)接口執(zhí)行確定性系統(tǒng)操作命令分發(fā)與 MCP 工具。Strategy Layer 由四部分組成Processor 框架本主題、Processing Strategies見 strategy.md、中間件系統(tǒng)日志、指標(biāo)、錯(cuò)誤處理等橫切關(guān)注點(diǎn)、依賴驗(yàn)證與上下文管理。設(shè)計(jì)哲學(xué)ProcessorTemplate遵循模板方法模式——process()定義工作流骨架子類通過_setup_strategies()/_setup_middleware()配置各階段的策略與中間件策略與中間件均在初始化時(shí)注入從而實(shí)現(xiàn)不改動(dòng)核心框架即可擴(kuò)展。ProcessorTemplate 框架結(jié)構(gòu)ProcessorTemplate是定義執(zhí)行工作流的抽象基類平臺(tái)專屬處理器AppAgentProcessor、HostAgentProcessor、LinuxAgentProcessor通過繼承它來配置各自平臺(tái)的策略與中間件。源碼位于 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)實(shí)際源碼中的定義與文檔一致processing_context.py。抽象基類核心職責(zé)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() # 子類按階段注冊(cè)策略與中間件 self._setup_strategies() self._setup_middleware() # 創(chuàng)建本地/全局分離的處理上下文 self.processing_context self._create_processing_context() # 初始化即校驗(yàn)策略鏈依賴盡早暴露配置錯(cuò)誤 self._validate_strategy_chain()子類必須實(shí)現(xiàn)的兩個(gè)抽象方法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需要說明的源碼差異文檔中的示例代碼為教學(xué)意圖的簡化示意實(shí)際倉庫在初始化順序上略有不同 —— 源碼在_setup_strategies()/_setup_middleware()之后才創(chuàng)建processing_context并調(diào)用_validate_strategy_chain()且校驗(yàn)失敗默認(rèn)僅記錄 warning 日志_validate_strategy_chain捕獲異常并打印Strategy dependency issue而文檔版本描述為直接拋出ProcessingException。讀者以 processor_framework.py 的實(shí)際實(shí)現(xiàn)為準(zhǔn)。上下文創(chuàng)建與公共數(shù)據(jù)注入源碼為處理器上下文提供了更強(qiáng)的初始化支持processor_framework.pydef _get_common_context_data(self) - Dict[str, Any]: 所有處理器共有的本地上下文初始化數(shù)據(jù) 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()追加平臺(tái)專屬數(shù)據(jù)如AppAgentProcessor注入subtask、application_process_name、app_root見 app_agent_processor.py。_create_local_context()會(huì)嘗試用合并后的數(shù)據(jù)構(gòu)造上下文實(shí)例若目標(biāo)上下文類不接受某些參數(shù)則依次降級(jí)為僅公共數(shù)據(jù)乃至空構(gòu)造 手動(dòng) setattr保證框架健壯性?？蚣苁找嬉恢碌墓ぷ髁魉刑幚砥髯裱粓?zhí)行模式跨平臺(tái)行為可預(yù)測(cè)平臺(tái)定制化子類只配置策略不修改核心框架框架可復(fù)用編排邏輯被所有處理器共享消除重復(fù)代碼中間件支持日志、指標(biāo)、錯(cuò)誤處理等橫切關(guān)注點(diǎn)統(tǒng)一作用于每次策略執(zhí)行可測(cè)試性每個(gè)階段可用 mock 策略與 mock 上下文獨(dú)立測(cè)試。策略注冊(cè)Strategy Registration處理器通過為每個(gè)階段注冊(cè)策略來配置工作流。文檔給出的注冊(cè)關(guān)系如下例 1Windows AppAgent 處理器以下代碼與 app_agent_processor.py 的實(shí)際實(shí)現(xiàn)一致含 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 - 組合多個(gè)策略 self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy( strategies[ AppScreenshotCaptureStrategy(), # 截取應(yīng)用截圖 AppControlInfoStrategy() # 提取 UI Automation 樹 ], nameAppDataCollectionStrategy, fail_fastTrue, ) # Phase 2: LLM_INTERACTION - 單策略LLM 失敗應(yīng)觸發(fā)恢復(fù) self.strategies[ProcessingPhase.LLM_INTERACTION] AppLLMInteractionStrategy( fail_fastTrue ) # Phase 3: ACTION_EXECUTION - 動(dòng)作失敗可優(yōu)雅處理 self.strategies[ProcessingPhase.ACTION_EXECUTION] AppActionExecutionStrategy( fail_fastFalse ) # Phase 4: MEMORY_UPDATE - 記憶更新失敗不應(yīng)中斷流程 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 - 執(zhí)行 shell 命令 self.strategies[ProcessingPhase.ACTION_EXECUTION] LinuxActionExecutionStrategy() # Phase 4: MEMORY_UPDATE - 記錄命令歷史 self.strategies[ProcessingPhase.MEMORY_UPDATE] LinuxMemoryUpdateStrategy()注冊(cè)最佳實(shí)踐需要多數(shù)據(jù)源的階段如 DATA_COLLECTION使用ComposedStrategy職責(zé)單一集中的階段如 LLM_INTERACTION使用單個(gè)策略除非需要初始化/清理資源否則不要注冊(cè) SETUP / CLEANUP 階段通過覆寫processor_context_class定義平臺(tái)專屬數(shù)據(jù)結(jié)構(gòu)見 app_agent_processing_context.py 中AppAgentProcessorContext的截圖路徑、過濾控件、LLM 響應(yīng)等字段。中間件系統(tǒng)Middleware System中間件為所有策略執(zhí)行提供統(tǒng)一的橫切關(guān)注點(diǎn)日志、指標(biāo)、錯(cuò)誤處理、緩存等在每個(gè)階段執(zhí)行前后以及出錯(cuò)時(shí)被調(diào)用。接口定義見 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內(nèi)置中間件EnhancedLoggingMiddleware框架內(nèi)置的EnhancedLoggingMiddleware提供完整的執(zhí)行期日志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}) # 把本地上下文總耗時(shí)、各階段耗時(shí)寫入日志文件 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)} )關(guān)鍵特性上下文感知日志記錄 round/step 信息便于溯源結(jié)果摘要記錄總執(zhí)行時(shí)間與各階段耗時(shí)分解持久化日志將結(jié)構(gòu)化上下文數(shù)據(jù)通過to_dict(selectiveTrue)選取關(guān)鍵字段序列化為 JSON 寫入日志文件增強(qiáng)錯(cuò)誤處理區(qū)分ProcessingException攜帶 phase、context_data、original_exception與普通異常完整堆棧捕獲traceback.format_exception輸出完整調(diào)用棧便于調(diào)試。AppAgentProcessor還提供了AppAgentLoggingMiddleware繼承自EnhancedLoggingMiddleware在標(biāo)準(zhǔn)日志基礎(chǔ)上疊加 rich 控制臺(tái)面板展示Round/Step/Agent與starting_message()的啟動(dòng)信息、成功/失敗彩色輸出等見 app_agent_processor.py。中間件執(zhí)行順序處理前按順序?yàn)槊總€(gè)中間件調(diào)用before_process()策略執(zhí)行策略按階段依次執(zhí)行處理后按逆序?yàn)槊總€(gè)中間件調(diào)用after_process()對(duì)應(yīng)源碼 processor_framework.py 的reversed(self.middleware_chain)出錯(cuò)時(shí)對(duì)所有中間件調(diào)用on_error()。中間件收益關(guān)注點(diǎn)分離橫切邏輯與策略邏輯解耦、可復(fù)用同一中間件跨處理器使用、非侵入增刪中間件無需改動(dòng)策略。工作流執(zhí)行Workflow Executionprocess()是ProcessorTemplate的模板方法負(fù)責(zé)編排完整執(zhí)行流程processor_framework.py。執(zhí)行順序# ProcessorTemplate.process() 內(nèi)按 ProcessingPhase 枚舉順序執(zhí)行 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 ]階段執(zhí)行規(guī)則可選階段SETUP 與 CLEANUP 可選未注冊(cè)策略則跳過順序執(zhí)行階段按固定順序串行執(zhí)行不支持并行依賴驗(yàn)證執(zhí)行前通過StrategyDependencyValidator校驗(yàn)源碼中為_validate_strategy_dependencies_runtime見 processor_framework.py執(zhí)行結(jié)果校驗(yàn)策略執(zhí)行后還會(huì)調(diào)用_validate_strategy_provides_runtime檢查聲明的 provides與實(shí)際返回字段的一致性缺失或多余均記錄 warning見 strategy_dependency.pyFail-Fast vs 繼續(xù)執(zhí)行由策略的fail_fast設(shè)置決定錯(cuò)誤處理方式。當(dāng)策略返回失敗結(jié)果時(shí)處理器會(huì)構(gòu)造異常并觸發(fā)中間件on_error后終止對(duì)應(yīng)源碼break邏輯當(dāng)策略拋出ProcessingException時(shí)外層 catch 會(huì)將其 phase 與 context_data 寫入失敗結(jié)果并再次觸發(fā)on_error中間件上下文即時(shí)更新每個(gè)策略的輸出通過update_local()立即寫入ProcessingContext供下一策略讀取階段結(jié)果追蹤每個(gè)階段的ProcessingResult含成功標(biāo)志、執(zhí)行耗時(shí)、數(shù)據(jù)鍵、錯(cuò)誤被記錄在phase_results有序字典中最終隨combined_result.data[phase_results]/phase_results_summary返回processing_context.py。上下文收尾finalize_finalize_processing_context()負(fù)責(zé)把本地?cái)?shù)據(jù)提升到全局上下文累加 LLM 成本到CURRENT_ROUND_COST/SESSION_COST自增CURRENT_ROUND_STEP與SESSION_STEPprocessor_framework.py。子類可覆寫以定制收尾邏輯。ProcessingContext本地與全局?jǐn)?shù)據(jù)分離ProcessingContext為所有策略提供統(tǒng)一的數(shù)據(jù)訪問分離本地處理器專屬與全局會(huì)話級(jí)數(shù)據(jù)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 # 模塊系統(tǒng)全局上下文 local_context: BasicProcessorContext # 處理器本地?cái)?shù)據(jù) 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 自動(dòng)轉(zhuǎn)大寫優(yōu)先匹配 ContextNames 枚舉否則查內(nèi)部字典 ... def update_local(self, data: Dict[str, Any]) - None: 批量更新本地上下文 ... def require_local(self, field_name: str, expected_type: Type None) - Any: 獲取必需字段缺失或類型不符時(shí)拋出 ProcessingException 并在 context_data 中附帶 available_keys 便于排查。 ...上下文分離的設(shè)計(jì)依據(jù)全局上下文會(huì)話級(jí)、跨組件共享用戶請(qǐng)求REQUEST會(huì)話 ID、round 號(hào)、step 號(hào)配置設(shè)置命令分發(fā)器引用command_dispatcher黑板Blackboard引用。本地上下文處理器專屬、臨時(shí)數(shù)據(jù)截圖數(shù)據(jù)screenshot、screenshot_pathUI 控件信息control_infoLLM 解析響應(yīng)parsed_response動(dòng)作執(zhí)行結(jié)果results臨時(shí)處理數(shù)據(jù)。此外BasicProcessorContextprocessing_context.py統(tǒng)一承載agent_type、session_step/round_step/round_num、action、arguments、parsed_response、llm_cost、execution_times、custom_data等公共字段update_from_dict()對(duì)未知字段自動(dòng)落入custom_data保證向后兼容。get_context_summary()提供面向日志/調(diào)試的摘要視圖。平臺(tái)專屬處理器一覽不同代理類型實(shí)現(xiàn)平臺(tái)專屬處理器PlatformProcessor ClassDATA_COLLECTIONLLM_INTERACTIONACTION_EXECUTIONMEMORY_UPDATEWindows AppAgentAppAgentProcessor截圖 UI 樹UI 元素選擇UI Automation 命令UI 交互歷史Windows HostAgentHostAgentProcessor桌面截圖 應(yīng)用列表應(yīng)用選擇啟動(dòng)應(yīng)用、創(chuàng)建 AppAgent應(yīng)用選擇歷史LinuxLinuxAgentProcessor截圖 shell 輸出Shell 命令生成Shell 命令執(zhí)行命令歷史倉庫中實(shí)際存在的處理器與上下文實(shí)現(xiàn)包括app_agent_processor.py、host_agent_processor.py、customized_agent_processor.py以及 app_agent_processing_context.py、host_agent_processing_context.py。平臺(tái)專屬策略分別位于 strategies/app_agent_processing_strategy.py、strategies/host_agent_processing_strategy.py、strategies/linux_agent_strategy.py、strategies/mobile_agent_strategy.py。設(shè)計(jì)要點(diǎn)與最佳實(shí)踐處理器設(shè)計(jì)指南1. 清晰的階段劃分每個(gè)階段職責(zé)單一 —— DATA_COLLECTION 采集原始數(shù)據(jù)、LLM_INTERACTION 推理決策、ACTION_EXECUTION 執(zhí)行命令、MEMORY_UPDATE 持久化狀態(tài)。2. 合理的策略組合多源數(shù)據(jù)采集使用ComposedStrategy。實(shí)際實(shí)現(xiàn)processing_strategy.py會(huì)順序執(zhí)行組件策略、將成功策略的結(jié)果即時(shí)寫入共享上下文供后續(xù)組件讀取、自動(dòng)聚合全部依賴與 provides 元數(shù)據(jù)并支持 fail-fast / 繼續(xù)執(zhí)行兩種錯(cuò)誤模式self.strategies[ProcessingPhase.DATA_COLLECTION] ComposedStrategy([ AppScreenshotCaptureStrategy(), AppControlInfoStrategy() ])3. 中間件承載橫切關(guān)注點(diǎn)不要在策略內(nèi)部實(shí)現(xiàn)日志/指標(biāo)邏輯。4. 依賴驗(yàn)證利用StrategyDependencyValidator自動(dòng)校驗(yàn)。依賴可通過兩種方式聲明詳見 strategy_dependency.py覆寫get_dependencies()/get_provides()返回StrategyDependency列表使用depends_on(...)/provides(...)裝飾器自動(dòng)登記到StrategyMetadataRegistry供運(yùn)行時(shí)校驗(yàn)與 provides 一致性檢查使用。5. 自定義上下文類按平臺(tái)定義專屬上下文。倉庫真實(shí)示例AppAgentProcessorContextapp_agent_processing_context.py包含subtask、app_root、application_process_name、多路徑截圖字段、filtered_controls、annotation_dict、response_text、prompt_message等應(yīng)用專屬數(shù)據(jù)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 主鏈路 -改變階段順序不要重排階段會(huì)破壞依賴鏈 -策略內(nèi)部存狀態(tài)不要在執(zhí)行期間修改策略實(shí)例狀態(tài)請(qǐng)改用上下文 -直接修改代理屬性不要在處理器中直接改 agent 屬性應(yīng)通過記憶系統(tǒng)等正規(guī)通道。與其它層的集成集成點(diǎn)層/組件關(guān)系A(chǔ)gentStateLevel-1 State狀態(tài)調(diào)用processor.process()執(zhí)行工作流ProcessingStrategyLevel-2 Strategy處理器注冊(cè)并執(zhí)行策略CommandDispatcherLevel-3 CommandACTION_EXECUTION 策略使用分發(fā)器下發(fā)命令Memory/Blackboard記憶系統(tǒng)MEMORY_UPDATE 策略更新代理記憶Global Context模塊系統(tǒng)處理器通過上下文讀取請(qǐng)求、寫回結(jié)果相關(guān)文檔狀態(tài)層見 state.md策略層見 strategy.md命令層見 command.md記憶系統(tǒng)見 memory.md平臺(tái)處理器總覽見 agent_types.md。關(guān)鍵結(jié)論P(yáng)rocessorTemplate工作流編排的抽象框架遵循模板方法模式策略注冊(cè)通過_setup_strategies()配置階段專屬策略中間件系統(tǒng)日志、錯(cuò)誤處理等橫切關(guān)注點(diǎn)統(tǒng)一作用于所有策略執(zhí)行工作流執(zhí)行按 DATA_COLLECTION → LLM_INTERACTION → ACTION_EXECUTION → MEMORY_UPDATE 順序編排并在執(zhí)行前后分別校驗(yàn)依賴與 provides 一致性依賴驗(yàn)證StrategyDependencyValidator結(jié)合StrategyMetadataRegistry保證策略在所需數(shù)據(jù)就緒后才執(zhí)行上下文管理本地處理器級(jí)與全局會(huì)話級(jí)數(shù)據(jù)分離require_local/get_global提供類型安全訪問平臺(tái)可擴(kuò)展子類化即可創(chuàng)建平臺(tái)專屬處理器Windows / Linux / 移動(dòng)端核心框架無需改動(dòng)。Processor 作為 Strategy Layer 的編排核心通過協(xié)調(diào)策略執(zhí)行、中間件應(yīng)用與上下文管理支撐代理在多樣平臺(tái)上可靠、高效地執(zhí)行復(fù)雜工作流?！久赓M(fèi)下載鏈接】UFOUFO3: Weaving the Digital Agent Galaxy項(xiàng)目地址: https://gitcode.com/GitHub_Trending/uf/UFO創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考