跳到主要内容

数据截至 (上游 commit 96983c73ed09)

第 2 章 · 一个 Step 内部:四阶段策略管线

这章讲什么: 上一章的 agent.handle() 里究竟发生了什么。答案是:跑一条固定顺序、可插拔的四段管线。这是 UFO 最工程化的一块,也是它能同时支持 Windows / Linux / Android / OpenAI Operator 的原因。

2.1 要解决的小问题​

每一步 agent 要做的事其实很多:截图、枚举控件、合并去重、画标注、拼 prompt、调模型、重试解析、执行动作、记录记忆、更新黑板、写结构化日志……

如果全塞进一个 process() 方法,就会得到一个八百行、没法给 Linux 复用的巨兽。 事实上仓库里确实有这样体量的文件——只不过它们被拆成了策略类(ufo/agents/processors/strategies/app_agent_processing_strategy.py 有 1784 行,但里面是 5 个独立类)。

2.2 思路:模板方法 + 阶段枚举​

阶段是个枚举​

# 真实定义,见 ufo/agents/processors/context/processing_context.py:24
class ProcessingPhase(Enum):
    SETUP = "setup"
    DATA_COLLECTION = "data_collection"
    LLM_INTERACTION = "llm_interaction"
    ACTION_EXECUTION = "action_execution"
    MEMORY_UPDATE = "memory_update"
    CLEANUP = "cleanup"

枚举里有六个阶段,但 AppAgent / HostAgent 都只注册了中间四个(SETUP 和 CLEANUP 是留给扩展的)。

模板方法按枚举顺序跑​

ProcessorTemplate.process 就是那个模板方法(ufo/agents/processors/core/processor_framework.py:336-441):

# 示意,非源码 —— 演示模板方法的骨架
for mw in self.middleware_chain:            # 前置中间件
    await mw.before_process(self, ctx)

for phase in ProcessingPhase:               # 按枚举顺序,注册了才跑
    if phase not in self.strategies:
        continue
    strategy = self.strategies[phase]
    self._validate_strategy_dependencies_runtime(strategy, ctx)  # 先查依赖齐不齐
    result = await strategy.execute(self.agent, ctx)
    if not result.success:
        break                               # 失败就停,交给错误中间件
    ctx.update_local(result.data)           # 产出并入局部上下文,给下一段用

这里有一个关键约定: 段与段之间不直接调用,只通过 ctx.update_local(result.data) 传值。上一段把产出写进局部上下文,下一段自己去取。这让任意一段都能被单独替换。

2.3 四段各干什么(以 AppAgent 为例)​

注册代码只有二十来行(ufo/agents/processors/app_agent_processor.py:82-109):

阶段策略类干什么fail_fast
DATA_COLLECTIONComposedStrategy(截图 + 控件)截当前窗口、枚举并合并控件、画 SoM 标注图True
LLM_INTERACTIONAppLLMInteractionStrategy拼 prompt、调模型、带重试地解析 JSONTrue
ACTION_EXECUTIONAppActionExecutionStrategy把模型给的动作转成 Command 并执行False
MEMORY_UPDATEAppMemoryUpdateStrategy写记忆项、更新黑板、写结构化日志False

fail_fast 的分级很讲究​

注释把理由写清楚了(ufo/agents/processors/app_agent_processor.py:87-109):

  • 截图/控件失败 → 后面全是空的,必须快失败。
  • 模型调用失败 → 没有动作可执行,必须快失败。
  • 动作执行失败 → 这很正常(控件消失了、点歪了),记下失败结果继续走,让模型下一步自己看着办。
  • 记忆更新失败 → 不该影响任务,吞掉。

这条分级把「哪种失败该中断」的判断从散落的 try/except 提升成了一个构造参数。

HostAgent 的四段形状一样,内容不同​

HostAgentProcessor._setup_strategies(ufo/agents/processors/host_agent_processor.py:95-112)注册的是 DesktopDataCollectionStrategy / HostLLMInteractionStrategy / HostActionExecutionStrategy / HostMemoryUpdateStrategy——截的是整个桌面而不是单个窗口,执行的动作是选应用而不是点控件。

2.4 组合策略:一个阶段里塞两件事​

AppAgent 的第一阶段其实是两件事(截图、抓控件),但阶段枚举只有一个槽位。解法是 ComposedStrategy(ufo/agents/processors/strategies/processing_strategy.py:140):

# 真实用法,见 ufo/agents/processors/app_agent_processor.py:87
self.strategies[ProcessingPhase.DATA_COLLECTION] = ComposedStrategy(
    strategies=[AppScreenshotCaptureStrategy(), AppControlInfoStrategy()],
    name="AppDataCollectionStrategy",
    fail_fast=True,
)

ComposedStrategy 会把子策略的 depends_on / provides 元数据汇总上来(_collect_strategy_metadata,:185),所以外层的依赖校验照样有效。组合不会让声明失真——这是它跟「随手写个 for 循环」的区别。

2.5 声明式依赖:装饰器 + 两次校验​

要解决的小问题​

段间靠字典传值,好处是解耦,坏处是打错一个 key,报错会在很远的地方冒出来(比如第三段拿到 None)。

做法:让每个策略把「我要什么、我给什么」写在类头上​

# 真实写法,见 ufo/agents/processors/strategies/app_agent_processing_strategy.py:1279
@depends_on("parsed_response", "log_path", "session_step")
@provides("execution_result", "action_info", "control_log", "status",
          "selected_control_screenshot_path")
class AppActionExecutionStrategy(BaseProcessingStrategy):
    ...

装饰器把这两串名字登记进一个全局元数据表 StrategyMetadataRegistry(ufo/agents/processors/core/strategy_dependency.py:44、:481 depends_on、:510 provides)。

校验发生两次​

怎么读这张图:上面一条是构造时跑一次;下面一条是每个阶段执行前后各跑一次。

  构造 Processor 时
     └─▶ _validate_strategy_chain()
            检查:后面阶段要的字段,前面阶段有人产出吗?

  每个阶段执行时
     ├─▶ 执行前 _validate_strategy_dependencies_runtime()
     │      检查:上下文里这些字段真的有值吗?
     └─▶ 执行后 _validate_strategy_provides_runtime()
            检查:你说要给的,真的给了吗?

三个方法都在 ufo/agents/processors/core/processor_framework.py:238、:265、:312。

这是本仓库最值得抄的一处设计: 用字典传值的松散管线,配上声明式契约,既保住了可插拔性,又把「传错 key」从运行期偶发 bug 变成了构造期的确定性报错。

2.6 局部上下文 vs 全局上下文​

管线里有两个上下文,别混:

局部上下文(local_context)全局上下文(Context)
生命周期一个 Step整个 Session
装什么截图路径、控件列表、模型响应、本步动作结果请求、子任务、当前窗口、累计成本、步数、日志器
谁写各策略的 result.data只有 _finalize_processing_context 挑选性地提升
定义ufo/agents/processors/context/processing_context.py:75 BasicProcessorContextufo/module/context.py:24 ContextNames

ProcessingContext 是把两者包在一起的门面(ufo/agents/processors/context/processing_context.py:172),get_local 只看局部,get 会退回全局。

为什么要分? 因为一个 Step 里产生的临时数据(比如十几个截图路径)不该污染整个会话。管线跑完后,只有精挑的几个字段(如 status)被提升到全局——这也是 agent.process() 结尾那句 self.status = self.processor.processing_context.get_local("status") 的来历(ufo/agents/agent/app_agent.py:368-385)。

每个 Processor 还能声明自己要往局部上下文里预置什么。AppAgent 预置的是子任务名、应用进程名、应用 root 名(_get_processor_specific_context_data,ufo/agents/processors/app_agent_processor.py:116-129)。

2.7 中间件:横切关注点​

中间件链有三个钩子:before_process / after_process / on_error(ufo/agents/processors/core/processing_middleware.py:25 ProcessorMiddleware)。

注意 after_process 是逆序执行的(processor_framework.py:409),这是标准的洋葱模型。

目前只挂了日志中间件,但它承担的活不少:

  • 前置:打印一个带轮次/步数/agent 名的面板(AppAgentLoggingMiddleware.before_process,ufo/agents/processors/app_agent_processor.py:148)。
  • 后置:成功时打印子任务与动作摘要、记录 LLM 花费。
  • 出错:打印红色错误行。

有个小细节值得注意:面板里带 emoji,而老版 Windows 控制台编码可能炸。代码专门写了 _safe_console_text,发现控制台编码不含 utf 就把非 ASCII 字符剥掉(ufo/agents/processors/app_agent_processor.py:40-47)。

2.8 模型调用的重试策略​

第二阶段调模型时,失败重试的判据不是 HTTP 错误,而是**「解析不出 JSON」**(ufo/agents/processors/strategies/app_agent_processing_strategy.py:1221-1268):

# 示意,非源码 —— 演示重试的判据
for attempt in range(config.json_parsing_retry):     # 默认 3 次
    response_text, cost = await run_in_executor(agent.get_response, prompt)
    try:
        agent.response_to_dict(response_text)        # 解析得动才算成功
        return response_text, cost
    except Exception:
        continue                                     # 解不动就重来

两处细节:

  • LLM 调用是同步的,被丢进 loop.run_in_executor 跑。源码注释写明理由:不这么做会阻塞事件循环,导致 WebSocket 心跳超时断连(:1236-1238)。
  • 重试次数来自 JSON_PARSING_RETRY(config/ufo/system.yaml:24),跟 LLM 自身的 MAX_RETRY 是两个东西。

2.9 平台扩展点​

同一套骨架换策略就能换平台。仓库里已有的几套:

策略文件面向
strategies/app_agent_processing_strategy.pyWindows 应用窗口
strategies/host_agent_processing_strategy.pyWindows 桌面(选应用)
strategies/linux_agent_strategy.pyLinux(走 shell)
strategies/mobile_agent_strategy.pyAndroid
strategies/customized_agent_processing_strategy.py自定义 agent
galaxy/agents/processors/strategies/base_constellation_strategy.pyGalaxy 的 DAG 生成/编辑

注意最后一行: Galaxy 那个「改 DAG」的 agent,用的是同一套 BaseProcessingStrategy 基类和同一个 ProcessorTemplate。这不是巧合,是刻意的——详见第 5 章。

2.10 代码地图​

主题文件路径符号名
模板方法主体ufo/agents/processors/core/processor_framework.pyProcessorTemplate、ProcessorTemplate.process
三处依赖校验ufo/agents/processors/core/processor_framework.py_validate_strategy_chain、_validate_strategy_dependencies_runtime、_validate_strategy_provides_runtime
阶段枚举与结果类型ufo/agents/processors/context/processing_context.pyProcessingPhase、ProcessingResult
双层上下文ufo/agents/processors/context/processing_context.pyProcessingContext、BasicProcessorContext
策略基类与组合ufo/agents/processors/strategies/processing_strategy.pyBaseProcessingStrategy、ComposedStrategy
依赖装饰器与注册表ufo/agents/processors/core/strategy_dependency.pydepends_on、provides、StrategyMetadataRegistry、StrategyDependencyValidator
中间件基类ufo/agents/processors/core/processing_middleware.pyProcessorMiddleware、EnhancedLoggingMiddleware
AppAgent 管线装配ufo/agents/processors/app_agent_processor.pyAppAgentProcessor._setup_strategies、AppAgentLoggingMiddleware
HostAgent 管线装配ufo/agents/processors/host_agent_processor.pyHostAgentProcessor._setup_strategies
模型调用与重试ufo/agents/processors/strategies/app_agent_processing_strategy.pyAppLLMInteractionStrategy._get_llm_response、_parse_app_response
全局上下文键名ufo/module/context.pyContextNames