跨设备 AI Agent 的容错进化：从「全局重规划」到「分层恢复」

📰 每日要闻

• A股周五收盘：上证指数 4090 点(-0.43%)，创业板指 4252 点(+2.05%)创历史新高，科创50飙涨3.84%，两市成交3.31万亿

• 美伊核协议6月19日瑞士正式签署，6月美股波动加剧，标普500一度回调约4%，AI主导行情降温

• Nathan Lambert发文呼吁不应禁止开源AI；MCP之父Sean Lynch称「MCP的理想形态可能只是一个Auth Gateway」

• AI陪伴机器人ZuzuZoos完成数千万元Pre-A轮融资，moody前高管搭档大疆骨干创业

从一次跨设备任务的崩溃说起

上周我在折腾一个 AI Agent 自动化流程——让它在 Linux 服务器上拉取数据、处理后推送到 Android 手机通知栏、最后在手机上打开 App 截图确认。三步操作，两台设备。

第二步挂了。FCM 推送通道临时抽风。然后 Agent 做了一件让我血压上升的事——重规划了整条任务链。服务器上的数据又拉了一遍。

就好比你点外卖，骑手在楼下被保安拦了。合理操作是换个门进。但外卖平台的做法是：把订单取消，重新下单，重新让餐厅做一份。

这周读到的 H-RePlan 论文（arXiv:2606.20487）精确命中了这个痛点。但我今天不想只转述论文——我打算基于它的思路，写一个真正能跑的最小化分层恢复框架，顺便指出论文没提到的几个工程坑。

30秒理解核心思想

H-RePlan 的核心就一个判断：这个故障该在哪一层修？

任务执行失败

↓

本地还有替代策略？

↓

✅ 有 → 设备内切换策略，不上报 （延迟：ms级 | token消耗：0）

❌ 没有 → 上报编排层，附带结构化故障描述 （延迟：秒级 | token消耗：一次LLM调用）

和分布式系统里的「就近恢复」一个道理。但 Agent 领域之前没人系统地做这件事。

一个能跑的最小实现（Python）

论文的伪代码太抽象了。下面我写一个约 100 行的可运行实现，核心包括三个组件：策略注册、设备级恢复器、编排级决策器。

组件一：策略注册与故障抽象

from dataclasses import dataclass
from enum import Enum
from typing import (
Callable, Optional, List
)
import asyncio, timeclass FailureScope(Enum):
STRATEGY = "strategy"
DEVICE = "device"
GLOBAL = "global"@dataclass
class FailureReport:
scope: FailureScope
failed_strategy: str
error: str
state_modified: bool
alternatives_left: int
# 关键字段：编排层据此
# 决定是否需要全局重规划
needs_global_replan: bool = False@dataclass
class Strategy:
name: str
executor: Callable
# 优先级越小越优先
priority: int = 0
# 超时（秒）
timeout: float = 30.0
# 连续失败次数 → 动态降权
fail_count: int = 0
last_fail_ts: float = 0

注意 FailureReport 里的 state_modified 字段——这是论文里一笔带过但工程上最要命的字段。如果执行到一半状态已经被改了（比如消息已发出、数据库已写入），你不能简单换策略重来，得先做补偿。

组件二：设备级恢复器

class DeviceRecoverer:
"""设备内分层恢复器"""def __init__(self, device_id: str):
self.device_id = device_id
# task → [Strategy]
self.registry: dict[
str, List[Strategy]
] = {}def register(
self, task: str,
strategies: List[Strategy]
):
# 按优先级排序
self.registry[task] = sorted(
strategies,
key=lambda s: s.priority
)async def execute_with_recovery(
self, task: str, **kwargs
) -> tuple[any, FailureReport|None]:
strategies = self._rank(task)
last_err = Nonefor i, strat in enumerate(
strategies
):
try:
result = await asyncio.wait_for(
strat.executor(**kwargs),
timeout=strat.timeout
)
# 成功：重置计数
strat.fail_count = 0
return result, None
except Exception as e:
strat.fail_count += 1
strat.last_fail_ts = (
time.time()
)
last_err = e# 所有策略耗尽
report = FailureReport(
scope=FailureScope.DEVICE,
failed_strategy=(
strategies[-1].name
),
error=str(last_err),
state_modified=False,
alternatives_left=0,
needs_global_replan=True
)
return None, reportdef _rank(
self, task: str
) -> List[Strategy]:
"""动态排序：最近老失败的降权"""
strategies = self.registry[task]
now = time.time()
return sorted(
strategies,
key=lambda s: (
s.priority
+ s.fail_count * 2
# 5分钟内失败过，
# 额外惩罚
+ (5 if (
now - s.last_fail_ts
) < 300 else 0)
)
)

重点看 _rank 方法。论文里策略排序是静态的，但在真实系统里你需要动态降权：一个策略如果最近 5 分钟内连续失败了 3 次，就不应该还排在前面。这种「记忆」机制论文没提，但不加就是一个坑——你的 Agent 会反复撞墙。

组件三：编排级决策器

class Orchestrator:
"""编排层：只处理设备搞不定的"""def __init__(self, devices: dict[
str, DeviceRecoverer
]):
self.devices = devicesasync def run_task_chain(
self,
plan: List[dict]
):
"""
plan = [
{"device": "linux",
"task": "fetch_data",
"args": {...}},
{"device": "android",
"task": "push_notify",
"args": {...}},
]
"""
results = []
for step in plan:
dev = self.devices[
step["device"]
]
result, report = (
await dev.execute_with_recovery(
step["task"],
**step["args"]
)
)if report is None:
results.append(result)
continue# 设备搞不定了
if report.state_modified:
# 需要补偿!
await self._compensate(
results, step
)# 重规划：把任务给另一台
new_plan = await self._replan(
step, report, plan
)
if new_plan:
return await self.run_task_chain(
new_plan
)
raise RuntimeError(
f"Unrecoverable: "
f"{report.error}"
)
return results

注意 _compensate 这个方法——这是我加的，论文里完全没讨论。现实中一旦 state_modified=True（比如你已经往数据库里写了一条记录），你就不能简单地换一台设备重来。得先把脏数据回滚或标记为无效。这是 Agent 容错最容易翻车的地方。

论文没告诉你的三个工程坑

坑一：状态补偿不是可选项

论文的故障抽象只区分「可恢复/不可恢复」，但实际你还需要区分「幂等/非幂等」操作。幂等的（查询、读文件）随便重试。非幂等的（发消息、写数据库、调支付接口）必须有补偿逻辑。

我的建议：每个 Strategy 额外注册一个 compensator 回调。不幂等的操作必须提供，否则 state_modified=True 时直接 fast-fail 到编排层。

坑二：超时竞争

当策略 A 超时后你切到策略 B，策略 A 可能还在执行中。如果 A 最终成功了，你就有两份执行结果。这在 GUI 自动化场景特别常见：屏幕点击发出去了但响应慢，Agent 以为超时换了别的方案，结果原始点击最后生效了。

解决方案：加一个 cancel token。策略切换时先取消前一个策略的执行上下文。在 Python 里就是 asyncio.Task.cancel()，在 Android 端侧就是 CoroutineScope.cancel()。

坑三：跨设备状态同步的延迟

编排层看到的设备状态是有延迟的。设备 A 说"我搞定了"，但网络延迟导致编排层还没收到确认就把任务又分配给了设备 B。

H-RePlan 论文假设通信是即时的。在真实系统里，你需要一个简单的分布式锁或者 lease 机制：

# 任务 lease：防止重复分配
class TaskLease:
def __init__(
self, ttl_sec=60
):
self.leases: dict[
str, float
] = {}def acquire(
self, task_id: str,
device_id: str
) -> bool:
now = time.time()
key = f"{task_id}"
if key in self.leases:
if (now -
self.leases[key]
) < self.ttl_sec:
return False
self.leases[key] = now
return Truedef release(
self, task_id: str
):
self.leases.pop(
task_id, None
)

和 Android Intelligence System 的交汇

Google I/O 2026 宣布 Android 从 OS 转型为 Intelligence System。翻译成人话：Agent 从第三方变成一等公民，App 从主角变成 Agent 的工具。

在这个新架构里，H-RePlan 的分层思想天然适配：

端侧模型负责快速的策略级决策——FCM 挂了换 WebSocket，API 超时降级到 CLI——这种决策不需要大模型，3B 参数量绰绰有余，延迟还能控制在 20ms 以内。只有真正搞不定的（设备不可达、需要重新分配任务给另一台设备）才上云。

目前 ExecuTorch 和 llama.cpp 在旗舰 Android 设备上跑 3B-7B 模型已经能做到 20-30 tokens/s。对于策略决策这种短输入短输出的场景，够用了。

Benchmark 数据的一点独立分析

论文的 HeraBench 结果值得单独说说：

几个值得注意的点：

• 全局重规划的 ROI 很差：花了 3.2 倍的 token（≈成本），只多拿了 16% 的完成率。这说明大量故障其实是策略级的，根本不需要全局重规划

• H-RePlan 的 perfect-pass（58%）最有价值：完成率只说明"做完了"，perfect-pass 说明"做对了"。分层恢复不仅效率高，还减少了不必要的状态污染

• 论文没测的：高并发下的表现。HeraBench 是串行测试。真实系统里多个设备可能同时报故障，编排层的决策队列会不会成为瓶颈？这是开放问题

我的判断和预测

三个观点：

1. 容错精细化是 2026 下半年 Agent 框架的主战场。LangGraph、CrewAI 目前的错误处理还停留在"retry N 次"。谁先把 scope-aware recovery 做进框架，谁就能啃下企业级复杂场景。

2. MCP 如果真变成 Auth Gateway，策略切换会变得极其便宜。Sean Lynch 那句话不是随口说的。如果工具连接变成一个轻量的认证+路由层，Agent 在运行时切换执行策略（换个 API、换个 endpoint）的成本就趋近于零。这会让分层恢复变得更加 practical。

3. 端侧 + 云端的双层决策会成为 Android Agent 的标准架构。Google 已经铺好了路（Gemini Nano + AI Core APIs），论文给了理论基础。接下来就是谁先把工程做出来的问题。

💡 One more thing：如果你现在在做 Agent 项目，最低成本的改进就是——给你的工具调用加一个策略池，给失败信息加结构化字段。不需要做跨设备系统，单设备也能立竿见影。

参考资料

• H-RePlan: Beyond Global Replanning: Hierarchical Recovery for Cross-Device Agent Systems (arXiv:2606.20487)

• LedgerAgent: Structured State for Policy-Adherent Tool-Calling Agents (arXiv, 2026-06)

• Google I/O 2026: Android Intelligence System (Android Developers Blog)

• On-Device LLMs on Android 2026 (Aleph Zero Labs)

• Sean Lynch on MCP as Auth Gateway (Simon Willison, 2026-06-19)