AI 必须在断网时工作：哪些功能应留在设备上？

断网时，哪些工作必须继续？

远程温室采集环境数据并发送给异常提示服务。如果互联网中断十二小时，问题不只是显示错误信息，还要决定是否继续测量、保存历史和提供本地提示。网络不稳定的设备同样面临这个问题。本文帮助读者在远程服务、本地设备和混合架构之间做选择：关键工作应摆脱哪些依赖，需要预留多少存储，以及如何恢复数据而不混淆过去与现在。

答案不是“把所有 AI 都搬到开发板上”。应分开即时功能、归档和可延迟分析。实际执行的模拟说明，十二小时数据所需空间可能不大，但容量预算错误仍会丢失部分数据；网络恢复也不会立刻清空积压。我们没有测试传感器、固件、模型、闪存或真实协议。场景是假设的，计算可复现，设备表现仍需实测。这既不是温室自动控制设计，也不是农学试验。

把有用决策与可用连接分开

最小路径包括采集测量值、检查格式和质量、执行所选本地功能并明确报告状态。简单告警可能只需要明确阈值，只有需求需要额外能力且评估证明有益时才应使用模型。本地 AI 还需要预处理、参数、辅助数据和运行时：下载权重不会消除隐藏的远程调用。具体检验应在测试环境断网，记录哪些依赖阻止功能，而不是仅凭架构图推断成功。

传感器故障或缺少关键更新时，本地功能不能假装一切正常。应区分有效测量、数据不足、模型不可用和连接缺失，让操作员了解局限。本假设场景的本地输出是提示，不是水泵命令：执行器需要单独的约束和验证。可等待的分析，例如多座温室历史比较，可以留在服务器。边界应由必须持续的工作决定，而非对云端或嵌入式的偏好。

离线十二小时需要多少存储？

选择开发板前先定义数据约定。假设每十秒生成一条记录，序列化后每条 64 字节，包含标识、时间、数值和状态。这是设定大小，不是 Python 对象、网络包或数据库页的大小。实际序列化必须计入额外完整性信息或元数据。十秒也不是作物采样建议，只是明确频率与空间的关系。

公式回答“中断期间产生多少记录”。lambda 是每秒记录数，D 是中断秒数，b 是每条字节数。B 为需保存的记录数量，S 为这些记录的数据空间。区间包含起点但不含终点：在 0、10、…、43190 秒采样。注意十进制与二进制单位不同：1 KiB 等于 1024 字节。

256 KiB 区域只能容纳 4096 条 64 字节记录，小于所需 270 KiB，尽管两个数字看起来接近。采用保留最新、删除最旧策略，十二小时结束时已缺少 224 条。这保护了当前数据，却丢失历史，并不必然优于拒绝新数据。如果目标是重建早期事件，丢掉的可能正是关键记录。溢出策略必须由历史数据的用途决定。

270 KiB 只是积压数据本身，不是整个平台需求。还需分别考虑模型权重、激活与峰值 RAM、协议缓冲区、记录索引和持久存储。标称空闲空间可能含保留区域，或需要按块擦除。因此，记录计算不证明某块开发板能运行应用。有用的第一步是明确预算，再用实际格式、固件和物理介质验证。

网络恢复了，为什么历史还没同步？

假设服务器稳定接收每秒两条有效记录。这是设定的服务能力，不是无线电实测或链路比特率，必须包含达到有用结果路径上的实际瓶颈，重传和等待可能降低它。同时本地仍每秒产生 0.1 条记录。积压减少速度是输出与输入之差，不是单看输出。可类比一边排水一边有水龙头注水：它解释收支平衡，不模拟故障或数据包。

连续近似得到约 37.9 分钟，而不是除以二所得的 36 分钟。要求 mu 大于 lambda；两者相等时旧积压不会减少，输出更慢时还会增长。公式假设速率恒定、服务立即可用且不再中断。突发流量或重连暂停需要更丰富的时间模型。实际应在支撑正常流量之外预留历史恢复余量：正常时刚好够用的连接，可能永远无法清理故障积压。

模拟：三种容量，相同事件

为检验公式，采用在确定时刻发生到达与离开的离散事件。一步为半秒，初始及每二十步生成带编号记录。到第 86400 步即十二小时为止，网络不移走记录；半秒后首次完成发送。此后每步完成一条，直到队列第一次为空。到达与离开重合时先加入新记录，缓冲区满时删除最旧记录。不含随机性，因此不需要种子。

容量 8192 时，峰值为 4321 条而不是 4320 条，因为恢复时刻、首次发送之前又到达一个样本。恢复后 2274 秒，即 37 分 54 秒，队列第一次清空，与连续近似一致。此时生成并上传了 4548 条，其中中断期间 4320 条，恢复期间又生成 228 条。上传数量超过最初积压不是重复，因为采集没有停止。模型在此终止，并未模拟当天剩余时间。

容量 4096 时丢失 225 条：中断时已丢 224 条，恢复后首次完成前又挤掉一条。恢复只需 2155.5 秒，比无损情况快。容量 1024 时仅需 538.5 秒，却缺少 3297 条。必须正确解释：“同步更快”可能掩盖数据删除。表格还列出生成和上传数；各行时长不同，因为每次在首次清空时结束，不能当作相同时间窗口比较总产量。

图中小队列的平坦部分并不表示传感器停止生成数据，而是每个新到达都删除旧数据。十二小时后曲线下降，只有最大容量在此协议下保存全部记录。坐标轴是模拟时间和记录数量，不是 RAM 或网络实测。序列化区域容量不等于 Python 进程内存：deque 含对象和额外元数据，不能拿来代表开发板测量。

已保存数据与已接收消息不是同一种保证

2019 年 3 月 7 日的 OASIS MQTT 5.0 标准区分交付级别。QoS 1 可能产生重复；协议确认不代表最终应用已完成工作。这些是协议性质，不是我们执行 MQTT 测试所得的结果。

归档服务应为每条记录设计稳定应用标识，例如设备、启动代次和一致持久化的计数器。服务器通过唯一约束保存标识，使重试不新增第二行。应测试写入后回复丢失的情况。仅使用重启后归零计数器，可能把两个不同样本认成同一条。本地删除策略还应说明收到何种确认才可释放空间，“包已发出”不是归档确认。

SQLite 原子提交文档说明，掉电后的行为还依赖存储和刷新操作假设。回滚模式与 WAL 模式机制不同。我们未测试 SQLite 或断电。

本队列只存在于模拟进程，不能证明重启存活、闪存完整性或写入原子性。真实设计应区分采集、持久化、发送和确认，重启必须重建状态而不删除未确认数据。数据库与日志应根据资源评估，不是可凭口号互换的方案。微控制器与嵌入式 Linux 可能需要不同策略。关键验证是在受控位置切断供电和通信，使用非运营数据与适合设备的协议。

不能让历史数据指挥现在

重连会把旧测量送到服务器。每条记录应分别保留采集与接收时间，以及质量和本地功能版本，否则几分钟上传的十二小时数据可能被误认为刚发生的事件。历史恢复应与当前决策分开。不要把旧命令自动当新请求重放，因为告警或动作可能已经失效。这不是安全执行器控制设计，而是区分需求不同的两种数据用途。

选择平台前应验证什么？

现在可以具体比较方案。纯云端便于更新中央分析，但功能依赖网络，远端队列不能保存设备从未发送的数据。纯本地可能自主工作，但更新、资源和维护仍是问题。混合方式由最小本地功能处理当前数据并保存记录，服务器提供可延迟分析。若足够，最小功能可以是不用 AI 的规则。模拟只比较队列，不衡量三种架构的诊断质量或经济价值。

验证需在选定设备上声明格式和版本，分别测量权重与峰值 RAM、可用存储和归档开销；测量同步期间本地功能平均延迟、分位数与截止时间；以瓦表示功率、以焦耳表示每次推理或有用决策能耗，区分采集和无线电。超过预算的中断用于检验饱和策略，更新与重启后重复用于检验状态连续性。这些是无结果的拟议实验，不是照片中电路板的隐含实测。

读代码，回答最初的问题

simulate 用 deque 有序保存数据，生成递增标识，满时删除最旧记录。移除只代表假设的服务完成，不是 MQTT 包或数据库确认。断言检查计数守恒、丢失与上传集合不相交，并验证 225 条丢失和无损恢复 2274 秒。没有重复只是模拟性质，不是网络保证。步数决定程序计算量，Python 内存和运行时间不是嵌入式基准。

操作性结论是：把产生有效本地提示所需功能留在设备，并声明局限；计算数据与状态预算；让恢复能力高于正常流入。本例 256 KiB 数据空间不足十二小时，小队列更快清空是因为丢了数据。8192 条容量在模型中无损，不证明抗断电或产品适用。架构选择已有理由，开发板与模型选择仍需缺失测试。本文不宣布 EL-AI 嵌入式产品或安装项目。

来源与可复现代码

OASIS — MQTT 5.0, 7 March 2019; sections 4.3–4.6.

SQLite — Atomic Commit; storage assumptions and sections 8–9.

import json,sys
from collections import deque
from pathlib import Path
# One tick is 0.5 seconds. New sample precedes completion at coincident events.
OUTAGE_TICKS=86400
RECORD_BYTES=64
def simulate(capacity):
 q=deque();lost=[];uploaded=[];maxq=0;trace=[];generated=0;end=None
 for tick in range(120001):
  if tick%20==0:
   if len(q)>=capacity:lost.append(q.popleft())
   q.append(generated);generated+=1;maxq=max(maxq,len(q))
  if tick>OUTAGE_TICKS and q:uploaded.append(q.popleft())
  if tick%120==0:trace.append([tick/2,len(q)])
  if tick>OUTAGE_TICKS and not q:end=tick/2;break
 trace.append([end,0])
 assert generated==len(lost)+len(uploaded)+len(q)
 assert not(set(lost)&set(uploaded)) and len(uploaded)==len(set(uploaded))
 return dict(capacityRecords=capacity,capacityBytes=capacity*RECORD_BYTES,lost=len(lost),uploaded=len(uploaded),generated=generated,maxQueue=maxq,drainSeconds=end-OUTAGE_TICKS/2,trace=trace)
runs=[simulate(c) for c in [1024,4096,8192]]
assert runs[2]['lost']==0 and runs[2]['drainSeconds']==2274 and runs[2]['maxQueue']==4321
assert runs[1]['lost']==225
assert abs(4320/(2-.1)-2273.684210526316)<1e-9
assert 4096*64==262144 and 4320*64==276480
result=dict(python=sys.version.split()[0],seed=None,scope='Discrete deterministic buffer simulation only; no sensor, model, MQTT client, flash or device tested',samplePeriodSeconds=10,assumedRecordBytes=64,outageSeconds=43200,assumedServiceRecordsPerSecond=2,fluidApproxDrainSeconds=4320/(2-.1),runs=runs)
Path(__file__).with_name('results.json').write_text(json.dumps(result,indent=2),encoding='utf8')
print(json.dumps([{k:v for k,v in x.items() if k!='trace'} for x in runs]))

代码、数据与说明 · JSON. 教学计算使用 Python 3.14.0，图使用 Matplotlib 3.11.2。分析由 AI 辅助，不声称经过同行评审或人工审核。原创 ImageGen 封面仅作示意，不记录 EL-AI 人员、场所或实际安装。来源查阅于 2026 年 10 月 27 日。