一步是怎么走的:Processor 模板 + 策略流水线 + 记忆黑板

30 秒导读: 上一章 讲的是 UFO² 的大循环——Session/Round 和 HostAgent → AppAgent 的状态机。本章把镜头拉近到一个 agent 的「一步」内部:当状态机说"轮到你走一步了",这一步具体经历了什么。答案是一条固定四阶段的流水线——看(采集)→ 想(问 LLM)→ 做(执行)→ 记(落记忆)——每一阶段都是一块可插拔的策略积木。

---

1. 这是什么(零基础也能懂)

一句话定义: UFO 里每个 agent 走的「一步」(一次 ReAct:观察 → 推理 → 行动)不是一坨面条代码,而是一条标准化的四阶段流水线,由 ProcessorTemplate 这个"模板方法"骨架统一驱动。

它要解决的问题。 GUI agent 的一步天然很重:要截屏、要读控件树、要拼一个巨大的多模态提示、要调 LLM、要解析回复、要真的去点鼠标、还要把这一切记下来给下一步用。如果每种 agent(HostAgent、AppAgent、各种定制 agent)都把这七八件事揉在一个几百行的 process() 里,就会出现大量重复、难以复用、改一处崩一片。

它的解法——把"流程"和"内容"分开:

• 流程(骨架)固定 —— 谁先谁后、出错怎么办、成本怎么累加,由 ProcessorTemplate 一次写死。
• 内容(每阶段干什么)可换 —— 每个阶段挂一个"策略(Strategy)"对象,像插件一样按 agent 类型替换。

一句话直觉/类比。 把它想成一条工厂流水线:传送带(模板)是固定的,四个工位依次是"拍照 → 请专家看照片做决定 → 按决定操作 → 记录工单";每个工位站着谁(哪个策略)可以随产品线更换,但传送带的走法永远不变。工位之间靠一张共享工单(ProcessingContext)传递半成品。

本章聚焦编排/流水线机制本身。截屏和控件树到底怎么采(感知)见 03 眼睛;动作怎么真正落到 UI/API(命令模式)见 04 手脚。

---

2. 顶层全景(它大概怎么转)

2.1 四阶段流水线

一个 agent 的一步,就是 ProcessorTemplate.process() 按 ProcessingPhase 枚举顺序,依次跑挂在每个阶段上的策略。

阶段枚举定义在 ufo/agents/processors/context/processing_context.py:24 的 ProcessingPhase——完整有 6 个,但 App/Host agent 只用其中 4 个核心阶段:

阶段(ProcessingPhase)	白话职责	AppAgent 挂的策略
DATA_COLLECTION	看:截屏 + 采控件信息	ComposedStrategy(截屏 + 控件两个策略合成)
LLM_INTERACTION	想:拼提示、问 LLM、解析回复	AppLLMInteractionStrategy
ACTION_EXECUTION	做:把回复里的动作派发执行	AppActionExecutionStrategy
MEMORY_UPDATE	记:写 MemoryItem、更新 Blackboard	AppMemoryUpdateStrategy

SETUP 和 CLEANUP 也在枚举里,但当前 App/Host 处理器没有为它们注册策略——process() 会跳过没有策略的阶段(见 processor_framework.py:354 的 if phase in self.strategies)。

2.2 一张图:一步的数据流

怎么读这张图:从上到下是时间顺序;中间那根竖线 ProcessingContext 是贯穿全程的共享黑板,每个阶段把产物写进去,下一阶段从里面读。

                          一个 agent 的「一步」= ProcessorTemplate.process()

   ┌──────────────────────────────────────────────────────────────────────┐
   │  middleware.before_process()   (日志/性能:开场)                        │
   └──────────────────────────────────────────────────────────────────────┘
        │
        ▼
   ① DATA_COLLECTION ──写─▶┐
     截屏 + 控件信息          │
        │                    │
        ▼                    │      ┌───────────────────────────┐
   ② LLM_INTERACTION ─读写─▶ ├────▶ │   ProcessingContext        │
     拼提示→调LLM→解析回复    │      │  (共享黑板 / 传送带工单)     │
        │                    │      │                            │
        ▼                    │      │  local_context: 本步私有    │
   ③ ACTION_EXECUTION ─读写▶ │      │  global_context: 跨步共享   │
     派发动作到 UI/API        │      └───────────────────────────┘
        │                    │
        ▼                    │
   ④ MEMORY_UPDATE ───读写─▶─┘
     写 MemoryItem + Blackboard
        │
        ▼
   ┌──────────────────────────────────────────────────────────────────────┐
   │  _finalize_processing_context()  (累加成本、step+1、提拔字段到全局)      │
   │  middleware.after_process()      (日志/性能:收尾,逆序执行)            │
   └──────────────────────────────────────────────────────────────────────┘

2.3 部件一句话职责

部件	干什么	在哪
ProcessorTemplate	模板方法:定义四阶段流程骨架、成本累加、错误处理	core/processor_framework.py:45
ProcessingPhase	阶段枚举,决定执行顺序	context/processing_context.py:24
ProcessingContext	共享黑板:统一封装 local + global 上下文	context/processing_context.py:171
BasicProcessorContext	本步私有数据的 dataclass 容器(可被子类扩展)	context/processing_context.py:74
ProcessingStrategy / BaseProcessingStrategy	单阶段策略的接口/基类	strategies/processing_strategy.py:17,29
ComposedStrategy	把多个策略压进"一个阶段一个策略"的槽	strategies/processing_strategy.py:140
StrategyMetadataRegistry + 装饰器	记录每个策略的 provides/depends_on,做依赖校验	core/strategy_dependency.py:44,481,510
ProcessorMiddleware	环绕整条流水线的中间件(日志/性能)	core/processing_middleware.py:25

2.4 谁来敲响这条流水线

上一章的状态机走到一步时,agent 会新建一个 processor 实例并 await 它。以 AppAgent 为例(ufo/agents/agent/app_agent.py:380):

# ufo/agents/agent/app_agent.py:380  handle() 内,示意保留真实调用
self.processor = self._processor_cls(agent=self, global_context=context)
await self.processor.process()                         # 跑完四阶段
self.status = self.processor.processing_context.get_local("status")  # 读出这一步的状态

status 就是流水线跑完后写回黑板的值(由动作执行阶段产出),状态机据它决定下一步去哪——这正是本章流水线与 01 状态机 的接缝。

---

3. 核心原理(逐个机制,由浅入深)

3.1 模板方法:流程写死一次,内容留空给子类

要解决的小问题: 让"四阶段 + 错误处理 + 成本累加"只写一遍,同时允许每种 agent 定制"每阶段具体干什么"。

思路。 这是经典的模板方法(Template Method)模式:基类 ProcessorTemplate 把不变的流程编排成 process(),把可变的部分抽成两个抽象钩子——_setup_strategies()(挂哪些策略)和 _setup_middleware()(挂哪些中间件)——强制子类实现。

原理演示(示意,非源码):

# 示意:模板方法的精神——流程固定,填空可变
class ProcessorTemplate:
    def __init__(self, agent, global_context):
        self.strategies = {}
        self._setup_strategies()      # 子类填:每个阶段挂什么策略
        self._setup_middleware()      # 子类填:挂什么中间件
        self._validate_strategy_chain()  # 初始化时先校验一遍依赖

    async def process(self):          # 骨架:所有 agent 共用
        for phase in ProcessingPhase: # 严格按枚举顺序
            if phase in self.strategies:
                result = await self.strategies[phase].execute(self.agent, ctx)
                ctx.update_local(result.data)   # 产物写回黑板,喂给下一阶段

真实实现。 骨架在 ProcessorTemplate.process()(processor_framework.py:336)。它遍历 ProcessingPhase,对每个已注册策略调用 strategy.execute(...),拿到 ProcessingResult 后 self.processing_context.update_local(result.data)(processor_framework.py:395)把产物合并进黑板供后续阶段消费。

子类只填两个钩子。AppAgent 的填法在 AppAgentProcessor._setup_strategies(app_agent_processor.py:82):

# ufo/agents/processors/app_agent_processor.py:87  真实注册,示意省略参数
self.strategies[ProcessingPhase.DATA_COLLECTION] = ComposedStrategy(
    strategies=[AppScreenshotCaptureStrategy(), AppControlInfoStrategy()],
    name="AppDataCollectionStrategy", fail_fast=True)
self.strategies[ProcessingPhase.LLM_INTERACTION]  = AppLLMInteractionStrategy(fail_fast=True)
self.strategies[ProcessingPhase.ACTION_EXECUTION] = AppActionExecutionStrategy(fail_fast=False)
self.strategies[ProcessingPhase.MEMORY_UPDATE]    = AppMemoryUpdateStrategy(fail_fast=False)

关键细节——fail_fast 是每阶段独立的容错策略。 注意上面四行的 fail_fast 不同:采集/LLM 用 True(失败即抛,触发恢复),动作/记忆用 False(失败只记日志不中断)。动作执行本就可能因界面漂移局部失败,记忆更新失败也不该毁掉整步——这个取舍直接写在注册处。HostAgent 的填法几乎同构(host_agent_processor.py:95),只是换成 DesktopDataCollectionStrategy 等 Host 专属策略。

3.2 ProcessingContext:一块 local + global 的双层黑板

要解决的小问题: 策略之间要传数据(截屏路径 → LLM;解析结果 → 动作执行),但又不能把一切都塞进永久全局状态。

思路——两层上下文:

层	生命周期	装什么	底层类型
local_context	一步之内	本步半成品:截屏路径、控件、解析回复、动作结果	BasicProcessorContext 及子类
global_context	跨步/跨 round	会话级状态:请求、step 计数、累计成本、选中的应用	Context(见 01 章)

ProcessingContext(processing_context.py:171)把这两层包在一起,对策略暴露统一接口。它有个巧妙设计:重写了 __getattr__/__setattr__(:196/:204),让 ctx.clean_screenshot_path 这种属性访问自动落到 local_context 上;不认识的字段则塞进 custom_data 兜底。策略里也常用更直白的 ctx.get_local(key) / ctx.update_local(dict)(:223/:261)。

local_context 是一个可扩展的 dataclass。 基类 BasicProcessorContext(:74)已经带了 session_step、llm_cost、action、parsed_response 等通用字段;AppAgent 用子类 AppAgentProcessorContext(context/app_agent_processing_context.py:8)加了 clean_screenshot_path、annotation_dict、control_log 等一大票 app 专属字段。processor 用哪个 context 类,由类属性 processor_context_class 指定(app_agent_processor.py:76),ProcessorTemplate 在 _create_local_context 里据此实例化(processor_framework.py:124)。

收尾时的"提拔"。 一步结束,_finalize_processing_context(processor_framework.py:192)负责把 local 里该长期保留的东西写回 global:累加 llm_cost 到 CURRENT_ROUND_COST 和 SESSION_COST,并把 SESSION_STEP/CURRENT_ROUND_STEP 各 +1。HostAgent 还额外重写了它(host_agent_processor.py:137),把选中的 subtask、target(选中的应用窗口)提拔到全局,好让下一步的 AppAgent 能读到——这正是 Host→App 交接的数据通道。

3.3 策略即插件 + provides/consumes 声明式契约

要解决的小问题: 阶段 B 要用阶段 A 的产物,但两者只通过一块字典黑板通信。怎么保证 B 需要的字段 A 真的给了?写错字段名怎么早点发现?

思路——每个策略声明自己的"进出口"。 用两个装饰器把契约挂到类上:

• @depends_on("clean_screenshot_path", ...) —— 我执行前,黑板上必须有这些字段(consumes)。
• @provides("control_info", "annotation_dict", ...) —— 我执行后,承诺往黑板写这些字段。

真实例子(app_agent_processing_strategy.py:425):

# ufo/agents/processors/strategies/app_agent_processing_strategy.py:425
@depends_on("clean_screenshot_path", "application_window_info")
@provides("control_info", "annotation_dict", "control_filter_time",
          "control_info_recorder", "annotated_screenshot_path", "annotated_screenshot_url")
class AppControlInfoStrategy(BaseProcessingStrategy):
    ...

这就把整条流水线连成了一张数据流图:上一策略 provides 的字段,恰好是下一策略 depends_on 的字段。

装饰器把契约存进注册表。 depends_on/provides(strategy_dependency.py:481/:510)不是简单加属性,而是把元数据登记进 StrategyMetadataRegistry(:44)这个全局字典,按类名索引。这样即便实例被 ComposedStrategy 包了一层,校验器也能按类查到它的进出口。

三处校验,分三个时机:

时机	校验什么	代码位置
processor 初始化	整条链:后面策略要的字段,前面有策略给吗?	_validate_strategy_chain → validate_strategy_chain(strategy_dependency.py:234)
每阶段执行前	运行时:黑板上此刻真有必需字段吗?	_validate_strategy_dependencies_runtime(processor_framework.py:265)
每阶段执行后	一致性:声明 provides 的字段真返回了吗?有没有多返回?	validate_provides_consistency(strategy_dependency.py:559)

注意:链校验发现问题只 warning 不中断(processor_framework.py:257),因为有些字段来自 global 或初始 setup(校验器会跳过 global_/command_/log_/session_ 前缀的字段,见 :266)。这是一种"防呆但不拦路"的软契约。

ComposedStrategy——把多个策略塞进一个阶段。 框架规定"一个阶段一个策略",但采集阶段实际要做两件事(截屏 + 控件)。ComposedStrategy(processing_strategy.py:140)是个"策略的容器策略":它顺序执行内部子策略,每个子策略的产物 context.update_local(result.data) 立刻回灌黑板供下一个子策略用(:255),并把所有子策略的 provides/depends_on 聚合成自己的对外契约(_collect_strategy_metadata,:185)。对模板来说,它就是"一个"策略。

统一的错误处理。 所有策略继承 BaseProcessingStrategy.handle_error(processing_strategy.py:108):fail_fast=True 时抛 ProcessingException(触发中间件的 on_error);fail_fast=False 时只记日志、返回一个 success=False 的 ProcessingResult。模板的 process() 看到失败结果会触发 on_error 中间件并 break 掉后续阶段(processor_framework.py:379)。

3.4 中间件栈:环绕整条流水线的日志与性能

要解决的小问题: 记日志、写落盘、报错渲染这些"横切关注点",不该散落进每个策略。

思路——洋葱式中间件。 ProcessorMiddleware(processing_middleware.py:25)定义三个钩子:before_process(开场)、after_process(收尾)、on_error(出错)。process() 在四阶段之前顺序跑所有 before_process,四阶段之后逆序跑 after_process(processor_framework.py:411,reversed(...))——典型的洋葱包裹顺序。

内置实现干两件实事。 EnhancedLoggingMiddleware(processing_middleware.py:69):

• after_process 里把本步各阶段的 execution_time 汇总进 local_context.execution_times,再把整个 local_context 序列化成 JSON 落盘(:114-126)——这就是 UFO 日志目录里每步一行的结构化 response 记录。
• on_error 里区分 ProcessingException(带 phase/context_data 的富错误)和普通异常,打印不同详细度的堆栈(:130)。

AppAgent 用子类 AppAgentLoggingMiddleware(app_agent_processor.py:132)在此之上加了 Rich 面板的彩色进度("🚀 Round X, Step Y")和成功/失败的用户友好提示。中间件在 _setup_middleware 里注册(app_agent_processor.py:111)。

3.5 LLM 交互:多后端抽象 + 结构化响应 schema

要解决的小问题: 拼一个多模态巨型提示、可靠地调到某个 LLM 后端、把自由文本回复变成程序能用的结构。

这是 LLM_INTERACTION 阶段(AppLLMInteractionStrategy,app_agent_processing_strategy.py:824)干的活,分五小步:收集截屏字符串 → 检索知识/示例(RAG)→ 用 agent.message_constructor 拼提示 → 调 LLM → 解析。

多后端抽象。 真正的 LLM 调用统一走 ufo/llm/llm_call.py。get_completion(:14)→ get_completions(:34)按 agent 类型查到 API_TYPE/API_MODEL,再由 BaseService.get_service(...) 分发到具体后端服务(OpenAI、Azure、本地模型等)。它内建主备切换:主引擎抛异常时,若 use_backup_engine,自动改用 AgentType.BACKUP 重试一次(:99-109)。

非阻塞 + JSON 解析重试。 策略里不直接调 get_completion,而是 agent.get_response,并特意用 loop.run_in_executor 把同步 LLM 调用丢进线程池(app_agent_processing_strategy.py:1238),避免长响应阻塞事件循环、拖垮 06 AIP 的 WebSocket 心跳。外面还套了 json_parsing_retry 次重试(:1234):每次拿到回复先试 response_to_dict 解析,解析不了就重来。

结构化响应 schema。 回复不是裸字典,而是 pydantic 模型 AppAgentResponse(schemas/response_schema.py:25)。解析时 AppAgentResponse.model_validate(response_dict)(app_agent_processing_strategy.py:1285)做 schema 校验,不合规直接报错。这个 schema 的字段定义了 agent"能说什么":

字段	含义
observation / thought	观察与推理(ReAct 的思考痕迹)
plan / comment	后续计划、给用户的说明
action	动作:Union[List[ActionCommandInfo], ActionCommandInfo, None]
save_screenshot	是否把本步截屏存进 Blackboard 及原因

注意 action 是 List 或单个 的联合类型——这是下一节"多动作"的地基。解析成功后,策略把 parsed_response.model_dump() 摊平进 ProcessingResult.data(:945),于是 plan/action/thought 等字段自动进了黑板,供动作执行和记忆阶段消费。

3.6 记忆与协作:MemoryItem + 跨 agent 的 Blackboard

要解决的小问题: 这一步做了什么,得记下来给同一个 agent 的下一步看(短期记忆),也得共享给别的 agent看(协作)。

两个层次:

• 私有记忆 Memory —— 每个 agent 自己的一串 MemoryItem(memory/memory.py:100,:12)。MemoryItem 是个很灵活的 dataclass:set_value 动态加字段并登记进 _memory_attributes(:55),to_dict 只导出登记过的键。LLM 交互阶段就靠 agent.memory.get_latest_item() 回看上一步的 plan 和成功动作,拼进新提示(app_agent_processing_strategy.py:999,:1131)。
• 共享黑板 Blackboard —— 所有 agent 都能读写的公告板(memory/blackboard.py:36),分四类:questions、requests、trajectories、screenshots(:45-48)。

MEMORY_UPDATE 阶段(AppMemoryUpdateStrategy,app_agent_processing_strategy.py:1543)把这一步固化下来,四步走:攒 additional_memory(动作、成本、结果等)→ 造 MemoryItem → agent.add_memory(memory_item) 存进私有记忆 → _update_blackboard(:1732)把关键轨迹和(可选)截屏推上公告板。

Blackboard 怎么变成提示。 下一个 agent 拼提示时调 blackboard.blackboard_to_prompt()(blackboard.py:274),把四类内容格式化成带 [Questions & Answers:]、[Request History:]、[Step Trajectories...] 标签的多模态消息块(截屏还会转成 image_url)。这就是"上一个 app 做了什么,下一个 app 立刻知道"的物理实现——UFO 多 agent 协作的共享内存。

推上黑板的轨迹只保留 history_keys 配置里的字段(:1751),截屏则由 LLM 自己在 save_screenshot 里决定要不要存及原因(:1590)——省得公告板被无关信息淹没。

3.7 投机式多动作:一次 LLM 调用,走好几步

要解决的小问题: LLM 调用又慢又贵。如果每次只能点一下,填个表单十个字段就得来回问 LLM 十次。

思路——让 LLM 一口气给出一串动作(action sequence),一次执行掉。 前面 AppAgentResponse.action 是 List 联合类型,就是为此:当 LLM 对接下来几步很有把握(比如"点用户名框 → 输入 → 点密码框 → 输入 → 点登录"),它可以把这一连串动作放进一个回复里。

这是"投机(speculative)"的: 模型基于当前这一张截屏预测未来几步,赌中间界面不会突变。赌对了就省下好几次截屏+LLM 往返;赌错了,后续动作在真实界面上失败,靠动作执行阶段的容错兜住。

由配置开关驱动提示模板。 ufo_config.system.action_sequence 为真时,提示构造切到 system_as 模板(prompter/agent_prompter.py:232),并把示例里的单动作改写成 ActionList 格式(action2action_sequence,:393),教会模型输出动作数组。

执行侧无脑循环。 AppActionExecutionStrategy._execute_app_action(app_agent_processing_strategy.py:1405)把单个 action 统一包成列表,逐个转成 Command,一批丢给 command_dispatcher 执行(:1423-1437):

# ufo/agents/processors/strategies/app_agent_processing_strategy.py:1423  示意精简
if isinstance(actions, ActionCommandInfo):
    actions = [actions]                    # 单个也当列表处理
commands = [self._action_to_command(a) for a in actions if a.function]
execution_result = await command_dispatcher.execute_commands(commands)  # 一批执行

动作到底怎么落到真实 UI/API(Command 的命令模式、Puppeteer 混合自动化),是 04 手脚 的主题,这里只关心"一次回复 → 一批动作"的编排收益。

---

4. 巧妙之处(可借鉴的技术)

• 模板方法 + 策略注册字典 —— 流程只写一遍,内容按 agent 类型换插件。加一种新 agent 只需实现两个钩子(_setup_strategies/_setup_middleware)+ 一个 context 子类。见 processor_framework.py:45。
• provides/consumes 声明式契约 + 三时机校验 —— 用装饰器把策略的数据进出口显式化,初始化时静态查链、运行时前查依赖后查一致性,把"字段名打错"这类隐蔽 bug 提前暴露成 warning。见 strategy_dependency.py:481,510,559。
• ComposedStrategy 消解"一阶段一策略"的僵硬 —— 既守住框架不变量,又允许一个阶段做多件事,还自动聚合子策略的契约。见 processing_strategy.py:140。
• LLM 调用丢进线程池 —— 同步 SDK 调用用 run_in_executor 包裹,避免阻塞 async 事件循环拖垮 WebSocket 心跳,是 async agent 里易被忽视的正确姿势。见 app_agent_processing_strategy.py:1238。
• 投机式多动作 —— 用 schema 的联合类型 + 配置开关 + 提示改写,把"减少 LLM 往返"做成可开关的优化,执行侧零特判(单/多统一成列表)。见 schemas/response_schema.py:34 与 agent_prompter.py:393。
• pydantic schema 当护栏 —— LLM 回复必须过 model_validate,不合规即失败,把"自由文本"收敛成"可信结构"。见 app_agent_processing_strategy.py:1285。

5. 边界与局限

• 链校验是软的,不拦路。 依赖不满足在初始化时只 warning(processor_framework.py:257),真正的硬失败要等运行时。前缀白名单(global_/command_/session_...)是启发式的,命名不守约的全局字段可能被误判。
• SETUP/CLEANUP 阶段形同虚设。 枚举里有,但当前处理器没注册策略,想做"阶段前置准备/统一清理"得自己补策略。
• 投机式多动作会踩空。 一串动作基于同一张旧截屏预测,中途界面若跳变(弹窗、加载),后面的动作可能作用到错误目标——正确性依赖动作执行阶段和感知层的容错,不是零风险优化。
• MemoryItem._memory_attributes 是类级可变默认值。 定义为类属性 _memory_attributes = [](memory/memory.py:17),set_value 往里 append(:63)——这是 Python 里常见的"可变类属性被实例共享"隐患,阅读时需留意其实际作用域(inferred:此处未见每实例覆盖,行为依赖各子类如 ImageMemoryItem 显式重定义)。
• 本章不覆盖感知与动作的落地实现。 截屏/控件检测/视觉 grounding 见 03;命令模式与混合 GUI+API 自动化见 04。

6. 横向对比

• 与 01 主循环: 01 是宏观状态机(Session/Round、Host↔App 交接),本章是微观一步的内部流水线。状态机在 app_agent.py:380 处 await processor.process(),读回 status 决定下一跳——两章在此接缝。
• 与 05 Galaxy 星座: 本章是"一台电脑内、一个 agent 的一步";05 把请求拆成跨设备 DAG,是更外层的编排。相同的"声明式依赖/契约"思想,在 05 里放大成跨设备的任务依赖图。
• 通用模式: "模板方法固定流程 + 策略插件填内容 + 共享黑板传数据 + 声明式契约校验",是任何多阶段 agent 流水线都可借鉴的骨架。

7. 代码地图(导航索引)

主题	文件	关键符号
四阶段流程骨架	ufo/agents/processors/core/processor_framework.py	ProcessorTemplate、ProcessorTemplate.process、_finalize_processing_context
阶段枚举	ufo/agents/processors/context/processing_context.py	ProcessingPhase、ProcessingResult
共享上下文/黑板	ufo/agents/processors/context/processing_context.py	ProcessingContext、BasicProcessorContext、get_local、update_local
App 专属上下文	ufo/agents/processors/context/app_agent_processing_context.py	AppAgentProcessorContext、selected_keys
策略接口/基类/合成	ufo/agents/processors/strategies/processing_strategy.py	ProcessingStrategy、BaseProcessingStrategy、ComposedStrategy、handle_error
依赖契约与校验	ufo/agents/processors/core/strategy_dependency.py	depends_on、provides、StrategyMetadataRegistry、StrategyDependencyValidator、validate_provides_consistency
中间件栈	ufo/agents/processors/core/processing_middleware.py	ProcessorMiddleware、EnhancedLoggingMiddleware
App 处理器装配	ufo/agents/processors/app_agent_processor.py	AppAgentProcessor._setup_strategies、AppAgentLoggingMiddleware
Host 处理器装配	ufo/agents/processors/host_agent_processor.py	HostAgentProcessor._setup_strategies、_finalize_processing_context
四个 App 策略	ufo/agents/processors/strategies/app_agent_processing_strategy.py	AppScreenshotCaptureStrategy、AppControlInfoStrategy、AppLLMInteractionStrategy、AppActionExecutionStrategy、AppMemoryUpdateStrategy
LLM 多后端调用	ufo/llm/llm_call.py	get_completion、get_completions(备用引擎切换)
结构化响应 schema	ufo/agents/processors/schemas/response_schema.py	AppAgentResponse、HostAgentResponse
记忆	ufo/agents/memory/memory.py	MemoryItem、Memory
共享黑板	ufo/agents/memory/blackboard.py	Blackboard、blackboard_to_prompt、add_trajectories、add_image
投机式多动作	ufo/prompter/agent_prompter.py	action2action_sequence、system_prompt_construction