之前在 Termux 里跑并发任务时经常遇到一个奇怪现象任务一多手机背面就开始发烫紧接着 CPU 频率像被一只手按住所有进程一起变慢。这个现象就是 thermal throttling热节流。它不会直接报错但会让你的多进程系统从“流畅”变成“寸步难行”。本文将围绕这个问题记录一套在 Termux 中通过 SQLite 搭建 10 层 agent mesh 的完整方案包含可运行的 Python 代码、数据库表设计、温度监控方法和进程调度策略。无论你是想在旧手机上搭建一个实验性的多 Agent 任务系统还是准备把 Termux 当作低功耗开发环境这篇文章都值得收藏。1. 背景为什么要在手机上跑 10 层 Agent Mesh1.1 这个实验解决什么问题日常开发中当我们提到“多 Agent 协作”第一反应往往是 K8s、Docker Compose、Redis 消息队列这些重型基础设施。但如果你手里的设备只是一台安卓手机没有 root 权限也没有完整的 Linux 内核管理能力这些方案基本都跑不起来。Termux 提供了一个轻量级的 Linux 环境但它对进程管理、系统服务、网络端口绑定都有一定的限制。本文要探讨的是如何不依赖复杂基础设施仅用 Termux SQLite Python 就搭建出一个 10 层任务处理网格并且保证长时间运行不触发热节流。一个典型的 agent mesh 场景是这样的系统里有多个 worker 节点每个节点处理特定类型的任务。任务从第一层进入经过抽取、清洗、转换、计算、聚合等多个阶段最终输出结果。每一层都可能有多个 agent 并发处理任务层与层之间通过共享的存储介质传递数据。在传统架构中这个存储介质可能是 Redis、RabbitMQ 或 Kafka但在 Termux 这种轻量环境里SQLite 反而成了最合适的选择。1.2 SQLite 为什么适合做 Agent 总线很多人对 SQLite 的印象还停留在“单机小数据库”觉得它撑不起并发场景。实际上SQLite 在 WALWrite-Ahead Logging模式下多进程并发读、单进程写的能力足以支撑中小规模的 Agent 任务调度。关键在于你怎么设计表结构和事务粒度。SQLite 相比消息队列有几个不可替代的优势零配置Termux 里直接安装就能用。数据库即文件备份、迁移、查看都非常方便。支持事务、索引、唯一约束可以避免任务被重复消费。一个文件就能保存全部任务状态排查问题时用 db browser 打开就能看。缺点当然也有写入并发能力有限单条记录大小不宜过大。但对于 agent mesh 这种以任务状态流转为主、消息体通常只有几 KB 的场景SQLite 的瓶颈完全够用。1.3 热节流是最大的敌人在手机上跑任务最怕的不是 CPU 不够快而是“发挥不出来”。手机 SoC 有严格的温度控制策略当芯片温度超过阈值时系统会主动降低 CPU 频率、限制核心调度甚至关闭高性能核心。这个机制在 x86 设备上通常叫 PROCHOTProcessor Hot信号在 ARM 设备上则是通过 thermal_zone 温度节点与 governor 策略共同实现。热节流带来的问题非常隐蔽你的进程并没有崩溃但运行速度可能会降到原来的十分之一。如果 agent 之间有超时机制还会引发连锁的任务堆积和重复处理。所以在 Termux 中搭建多 Agent 系统第一优先级不是“跑满 CPU”而是“控制温度”。2. 环境准备Termux 里的最小开发栈2.1 安装 Termux 并更新软件源Termux 的安装渠道比较特殊Google Play 上的版本已经停止维护推荐从 F-Droid 或 GitHub Releases 下载最新 APK。安装完成后首先执行pkg update pkg upgrade -y这一步会更新软件源和已安装包。需要注意如果你的 Termux 是刚安装的pkg命令可能需要先初始化存储目录执行termux-setup-storage授权存储权限。2.2 安装 SQLite、Python 与监控工具本文的实战部分以 Python 为例需要安装以下包pkg install python sqlite clang -y pkg install termux-api -y这里说明一下每个包的用途python用来编写 Agent 工作循环。sqlite提供 SQLite 命令行工具方便调试和查询。clangPython 的某些依赖需要编译提前安装可以避免报错。termux-api用来读取电池温度、充电状态等系统信息是热节流保护模块的基础。如果你希望 Agent 输出更规范的日志还可以安装pandas、pydantic这类包但本文为了保持依赖最小化只用 Python 标准库。pip install --upgrade pip国内网络环境下pip 源如果不稳定可以临时使用清华镜像pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple版本方面需要提醒一下Termux 的软件源是滚动更新的不同时间安装的 Python 和 SQLite 版本可能不同。本文的代码只使用 SQLite 的基础语法和 Python 标准库只要你的 SQLite 版本在 3.8 以上支持 WAL 模式基本不存在兼容性问题。2.3 规划项目目录建议按下面的结构组织项目agent-mesh/ ├── db/ │ └── mesh.db ├── logs/ │ └── agent.log ├── scripts/ │ ├── init_db.py │ ├── agent_worker.py │ ├── thermal_guard.py │ └── dispatch.py └── README.md在 Termux 中执行mkdir -p agent-mesh/db agent-mesh/logs agent-mesh/scripts cd agent-mesh这样目录结构就准备好了。3. 理解热节流手机 SoC 的自我保护和监控方法3.1 Thermal Throttling 的硬件逻辑热节流不是一个单纯的操作系统概念它本质上是硬件层面的保护机制。手机 SoC 内集成了多个温度传感器分布在 CPU 集群、GPU、充电管理芯片等位置。当某个传感器的温度超过设定阈值时SoC 的电源管理单元会通过 DVFS动态电压频率调整降低核心电压和频率。所以你在top里看到的 CPU 占用率 100%实际运行速度可能已经低到不可用。更麻烦的是热节流的触发条件不止 CPU 负载还包括环境温度、屏幕亮度、充电状态。如果你的手机一边充电一边跑任务温度会迅速上升。3.2 在 Termux 中监控温度与频率在 Termux 中最直接的监控方式是读取系统温度节点。Android 系统通常会把温度传感器信息暴露在/sys/class/thermal/目录下。执行for zone in /sys/class/thermal/thermal_zone*/; do echo -n $zone: cat $zone/temp done温度值的单位通常是毫摄氏度也就是 65000 表示 65 摄氏度。为了方便我们可以用 Python 封装一个温度采集函数 文件路径: agent-mesh/scripts/thermal_guard.py 作用: 采集电池温度和 CPU 温度作为热节流保护模块的基础。 import os import subprocess import time def read_cpu_temp(): 读取第一个可用的 thermal_zone 温度单位转换为摄氏度。 for zone_idx in range(10): base f/sys/class/thermal/thermal_zone{zone_idx} temp_path os.path.join(base, temp) if os.path.exists(temp_path): try: with open(temp_path, r) as f: raw f.read().strip() return int(raw) / 1000.0 except (ValueError, IOError): continue return None def read_battery_temp(): 通过 termux-battery-status 读取电池温度。 try: result subprocess.run( [termux-battery-status], capture_outputTrue, textTrue, timeout5, ) output result.stdout # 简化解析这里只取 temperature 字段 import json data json.loads(output) return data.get(temperature) except Exception: return None if __name__ __main__: if os.environ.get(TERMUX_VERSION): print(fCPU 温度: {read_cpu_temp()}°C) print(f电池温度: {read_battery_temp()}°C) else: print(fCPU 温度: {read_cpu_temp()}°C)注意/sys/class/thermal/中的传感器节点在不同手机上有差异有的设备是thermal_zone0对应 CPU有的设备是thermal_zone1对应电池。这个脚本会遍历前 10 个 zone取第一个有效值实际使用时建议你先手动执行上面的 shell 命令确定哪个节点对应 CPU。3.3 目标把温度压在阈值以下经验数据表明手机 CPU 在 80°C 以下时通常可以保持较高的运行频率超过 85°C 后热节流会明显加强。为了保证 agent mesh 的吞吐稳定我们设定的目标温度建议不高于 75°C。具体的控制策略后文会详细展开。4. 10-Tier Agent Mesh 架构设计4.1 什么是 Agent MeshAgent mesh 指的是一种多节点协作模式每个节点是一个轻量级 Agent负责从队列中取出任务执行特定操作然后把结果写入下一个队列。这些 Agent 之间不直接通信而是通过共享数据库协调。这样做的好处是解耦每个 Agent 只知道自己的输入和输出不需要了解其他 Agent 的内部状态。“10-tier”并不是说必须有 10 台设备而是把任务处理流程纵向切分为 10 层每一层代表一种不同的处理阶段。本文的例子中各层职责如下层级职责示例操作tier_1原始任务接收写入原始数据tier_2数据清洗去空格、去重tier_3字段标准化格式化时间、枚举值tier_4关键字提取简单文本扫描tier_5规则过滤按规则丢弃任务tier_6计算处理数值计算或聚合tier_7结果分组按目录字段归组tier_8汇总统计生成统计摘要tier_9格式转换转 JSON、CSV 等tier_10输出归档写入结果表每一层可以运行一个或多个 Agent层数越多每个 Agent 的职责越单一代码越容易维护。4.2 数据模型10 层队列表怎么设计在 SQLite 中设计 10 层任务表时有两种思路一种是物理建 10 张表另一种是只建一张任务表通过tier字段区分。本文推荐第二种因为任务在各层之间流转时本质上只是修改tier字段的值不需要跨表复制性能更好代码也更简洁。核心表结构如下-- 文件路径: agent-mesh/db/schema.sql PRAGMA journal_mode WAL; CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, tier INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT pending, payload TEXT, current_agent TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)), attempts INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_tier_status ON tasks(tier, status); CREATE TABLE IF NOT EXISTS agent_heartbeats ( agent_id TEXT PRIMARY KEY, last_seen TEXT NOT NULL DEFAULT (datetime(now)), current_tier INTEGER, pid INTEGER );tasks表中tier当前任务所在层级1 到 10。status状态pending表示等待处理processing表示正在被某个 Agent 处理done表示完成failed表示失败。payload任务数据建议存储 JSON 字符串。attempts尝试次数用于防止任务无限重试。updated_at记录任务最近一次状态变更时间是任务超时回收的重要依据。agent_heartbeats表用于记录每个 Agent 的心跳监控 Agent 是否存活。生产环境中你可以进一步细化表结构比如增加task_type字段支持不同类型的任务流但本文的简化设计足够跑通整个流程。4.3 为什么选择“数据库轮询”而不是消息队列在电脑上搭 agent mesh你大概率会用 Celery、RQ 或 RabbitMQ。但在 Termux 环境里这些方案都有各自的痛点Celery 的 Broker 依赖 Redis 或 RabbitMQTermux 里编译安装较麻烦。RQ 相对轻量但同样需要 Redis。直接使用消息队列还需要额外的端口监听、持久化保障对手机来说负载偏高。数据库轮询的优点是实现简单、无额外依赖、进程重启后任务状态不丢。缺点是需要自己控制轮询间隔避免空转消耗 CPU。本文的 Agent 会采用“IDLE 退避”策略没有任务时逐步拉长轮询间隔从 1 秒逐步增加到 10 秒。5. 完整实战用 SQLite Python 搭建 Agent Mesh5.1 初始化数据库与 10 层任务表编写初始化脚本确保数据库和表结构正确创建 文件路径: agent-mesh/scripts/init_db.py 作用: 初始化数据库表结构并写入几条演示任务。 import sqlite3 import os import json import time DB_DIR os.path.join(os.path.dirname(__file__), .., db) DB_PATH os.path.join(DB_DIR, mesh.db) SCHEMA PRAGMA journal_mode WAL; CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, tier INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT pending, payload TEXT, current_agent TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)), attempts INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_tier_status ON tasks(tier, status); CREATE TABLE IF NOT EXISTS agent_heartbeats ( agent_id TEXT PRIMARY KEY, last_seen TEXT NOT NULL DEFAULT (datetime(now)), current_tier INTEGER, pid INTEGER ); def init_db(): os.makedirs(DB_DIR, exist_okTrue) conn sqlite3.connect(DB_PATH) conn.executescript(SCHEMA) conn.commit() conn.close() print(f[init_db] 数据库初始化完成: {DB_PATH}) def seed_tasks(count20): 写入 count 条演示任务全部从 tier_1 开始。 conn sqlite3.connect(DB_PATH) tasks [ ( 1, pending, json.dumps({content: f任务 {i}, value: i}), ) for i in range(count) ] conn.executemany( INSERT INTO tasks (tier, status, payload) VALUES (?, ?, ?), tasks, ) conn.commit() conn.close() print(f[init_db] 已写入 {count} 条演示任务) if __name__ __main__: init_db() seed_tasks(20)执行python scripts/init_db.py如果你本机没有安装sqlite3命令行工具也可以用 Python 的sqlite3模块直接查看数据python -c import sqlite3; print(sqlite3.connect(db/mesh.db).execute(select count(*) from tasks).fetchone())预期输出为(20,)表示有 20 条任务进入队列。5.2 Agent 工作循环每个 Agent 都是一个独立的 Python 进程它周期性地执行以下操作从指定 tier 取出一条 pending 任务并把状态改为 processing。处理任务更新 payload。把任务写入下一层tier 1状态改为 pending。如果已经是第 10 层则把状态改为 done。更新心跳。为了避免多个 Agent 同时取到同一条任务这里使用 SQLite 的事务和条件更新来完成“原子领取”。代码如下 文件路径: agent-mesh/scripts/agent_worker.py 作用: 单个 Agent 工作循环从指定 tier 领取任务并处理后进入下一层。 用法: python agent_worker.py --tier 1 --agent agent-a import argparse import json import os import sqlite3 import time DB_DIR os.path.join(os.path.dirname(__file__), .., db) DB_PATH os.path.join(DB_DIR, mesh.db) def db_connect(): conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row return conn def claim_task(conn, tier, agent_id): 原子地领取一条 pending 任务返回任务行或 None。 cursor conn.execute( UPDATE tasks SET status processing, current_agent ?, updated_at datetime(now), attempts attempts 1 WHERE id ( SELECT id FROM tasks WHERE tier ? AND status pending ORDER BY created_at ASC LIMIT 1 ) RETURNING id, tier, status, payload, attempts , (agent_id, tier), ) row cursor.fetchone() conn.commit() return row def process_payload(payload, tier): 核心处理函数这里用简单的时间戳和字段追加来模拟真实计算。 data json.loads(payload) # 模拟计算耗时避免 CPU 空转 time.sleep(0.2) data[processed_tier] tier data[processed_at] time.strftime(%Y-%m-%d %H:%M:%S) data[value] (data.get(value, 0) or 0) * 2 return json.dumps(data, ensure_asciiFalse) def write_next_tier(conn, row, next_tier, agent_id): 把任务写入下一层或者标记为完成。 if next_tier 10: conn.execute( UPDATE tasks SET status done, current_agent ?, updated_at datetime(now) WHERE id ? , (agent_id, row[id]), ) else: conn.execute( UPDATE tasks SET tier ?, status pending, current_agent NULL, payload ?, updated_at datetime(now) WHERE id ? , (next_tier, row[payload], row[id]), ) conn.commit() def heartbeat(conn, agent_id, tier): 更新 Agent 心跳让运维方知道进程还活着。 pid os.getpid() conn.execute( INSERT INTO agent_heartbeats (agent_id, last_seen, current_tier, pid) VALUES (?, datetime(now), ?, ?) ON CONFLICT(agent_id) DO UPDATE SET last_seen datetime(now), current_tier excluded.current_tier, pid excluded.pid , (agent_id, tier, pid), ) conn.commit() def run_agent(tier, agent_id): 主循环持续领取任务并处理。 idle_sleep 1.0 while True: conn db_connect() try: row claim_task(conn, tier, agent_id) if row is None: # 没有任务时退避降低 CPU 空转 idle_sleep min(idle_sleep 1.0, 10.0) time.sleep(idle_sleep) continue idle_sleep max(idle_sleep - 1.0, 1.0) row[payload] process_payload(row[payload], tier) write_next_tier(conn, row, tier 1, agent_id) heartbeat(conn, agent_id, tier) print(f[{agent_id}] 任务 {row[id]} - tier {tier 1}) finally: conn.close() if __name__ __main__: parser argparse.ArgumentParser(descriptionRun an agent worker) parser.add_argument(--tier, typeint, requiredTrue, helpAgent 监听的任务层级1-10) parser.add_argument(--agent, typestr, requiredTrue, helpAgent 唯一标识) args parser.parse_args() run_agent(args.tier, args.agent)代码中有几个关键点需要展开解释。claim_task函数使用了UPDATE ... WHERE id (SELECT ... LIMIT 1)的原子更新模式。在 SQLite 中这个子查询加更新的组合在同一事务里是原子性的可以有效避免两个 Agent 同时领取同一条任务。如果你的 SQLite 版本较老不支持RETURNING子句SQLite 3.35 以上才支持可以先用SELECT查出任务 id再通过带条件的UPDATE把状态从pending改为processing然后检查影响行数是否为 1# 兼容旧版本 SQLite 的领取方式 cursor conn.execute( SELECT id, payload FROM tasks WHERE tier ? AND status pending ORDER BY created_at ASC LIMIT 1 , (tier,), ) row cursor.fetchone() if row: cur2 conn.execute( UPDATE tasks SET status processing, current_agent ?, updated_at datetime(now) WHERE id ? AND status pending , (agent_id, row[id]), ) if cur2.rowcount 1: conn.commit() # 返回任务行process_payload函数是 Agent 的真正业务逻辑所在。本文用一个 0.2 秒的 sleep 和数值乘 2 来模拟真实计算你在实际项目中可以替换成文本解析、网络请求、图像压缩等操作。write_next_tier函数把当前层处理完的任务送入下一层。当层级增加到第 11 层时说明整个流水线已经执行完毕任务状态被置为done。heartbeat函数每隔一段时间把 Agent 的信息写入心跳表。主循环中的idle_sleep实现了退避策略任务持续到达时保持 1 秒轮询一次没有任务时逐步拉长到 10 秒。这个设计对控温非常重要。5.3 启动多节点并验证分层流转启动 10 个 Agent分别监听 1 到 10 层cd agent-mesh for i in $(seq 1 10); do python scripts/agent_worker.py --tier $i --agent agent-$i logs/agent-$i.log 21 done执行后用ps查看进程ps aux | grep agent_worker每个 tier 都有一个对应的 Agent 进程。然后查看任务流转情况sqlite3 db/mesh.db select tier, status, count(*) from tasks group by tier, status;预期输出大致如下1|pending|0 2|pending|4 3|pending|6 ... 10|pending|2由于各层处理速度不同任务会逐渐从 tier_1 流向 tier_10。最终所有任务都应该变成10|done|20也就是说20 条任务全部走完了 10 层流水线。5.4 热节流保护模块在 5.2 节的 Agent 循环里我们其实还没有加入真正的温度控制逻辑。为了长时间运行不触发热节流需要引入一个保护模块在每次循环开始前检查温度如果超过阈值就主动“躺平”一段时间让 SoC 降温。下面是在agent_worker.py中加入温度控制的改造思路# 在 agent_worker.py 中 import thermal_guard from thermal_guard import read_cpu_temp, read_battery_temp def get_current_temperature(): 返回当前系统温度优先使用电池温度因为最贴近真实手感。 battery_temp read_battery_temp() if battery_temp is not None: return battery_temp return read_cpu_temp() or 0.0 TEMPERATURE_LIMIT 72.0 SLEEP_BACKOFF 30.0 def wait_if_hot(agent_id): 如果温度过高Agent 进入冷却休眠。 temp get_current_temperature() if temp TEMPERATURE_LIMIT: print(f[{agent_id}] 温度过高: {temp:.1f}°C, 冷却休眠 {SLEEP_BACKOFF}s) time.sleep(SLEEP_BACKOFF) return True return False然后在主循环开头调用def run_agent_with_thermal_guard(tier, agent_id): idle_sleep 1.0 while True: conn db_connect() try: # 热节流保护温度过高时暂停处理 if wait_if_hot(agent_id): continue row claim_task(conn, tier, agent_id) ... finally: conn.close()这个模块的工作逻辑是温度超过 72°C 时所有 Agent 会进入 30 秒的冷却休眠不领取新任务让系统负载降下来。这个策略虽然简单但在手机这种被动散热设备上非常有效。更精细的做法是动态调整休眠时长比如温度越高休眠越长def wait_if_hot_dynamic(agent_id): temp get_current_temperature() if temp TEMPERATURE_LIMIT: extra int((temp - TEMPERATURE_LIMIT) * 2) sleep_time 15 extra print(f[{agent_id}] 温度 {temp:.1f}°C, 冷却休眠 {sleep_time}s) time.sleep(sleep_time) return True return False这里(temp - TEMPERATURE_LIMIT) * 2的意思是温度每超过阈值 1°C休眠时间增加 2 秒。假设温度 76°C休眠时间就是 15 8 23 秒温度 80°C休眠时间就是 31 秒。这种梯度休眠策略能让温度回落更平缓同时不至于完全停止所有 Agent。需要注意的是wait_if_hot只是软件层面的自我保护。如果你的手机本身散热能力很差或者你正在边充电边跑任务软件的降载效果会很有限。更稳妥的做法是结合硬件策略比如关闭充电、降低屏幕亮度这些内容在最佳实践部分会继续展开。运行热保护版 Agentcd agent-mesh python scripts/agent_worker.py --tier 1 --agent agent-hot-1观察日志当温度接近阈值时你会看到类似输出[agent-hot-1] 温度 73.2°C, 冷却休眠 20s5.5 验证热节流是否被有效避免启动带热保护的 Agent 集群后持续运行一段时间然后用下面的命令记录温度曲线while true; do temp$(cat /sys/class/thermal/thermal_zone0/temp) echo $(date %H:%M:%S) $((temp / 1000))°C echo $(date %H:%M:%S) $((temp / 1000))°C logs/temperature.log sleep 5 done运行 10 分钟后打开logs/temperature.log如果温度始终在 72°C 附近波动而不是持续攀升说明热保护模块生效了。如果温度仍然一路走高到 85°C 以上说明你的业务逻辑计算量太大或者 Agent 数量过多需要减少并发 Agent 数量或者降低每层 Agent 的轮询频率。6. 常见问题与排查思路在 Termux 中搭建和运行 SQLite agent mesh最常遇到的问题有下面这些。问题现象常见原因解决思路数据库显示database is locked多个进程同时尝试写入事务冲突设置timeout10并确保每个连接及时关闭改用 WAL 模式Agent 领取任务后任务丢失进程被系统杀掉事务未提交确保领取任务和更新状态在同一个事务内使用attemps字段做超时回收温度迅速飙升到 85°C 以上Agent 数量太多或轮询间隔太短减少并发 Agent调大idle_sleep在温度过高时休眠任务卡在processing不继续流转Agent 进程崩溃或手机息屏后进程被冻结添加超时回收机制定期把updated_at超过 5 分钟的处理中任务重置为pendingtermux-api命令返回空未安装 Termux:API 应用或未授权需要在 F-Droid 额外安装 Termux:API APK并在系统设置中授权Python 进程在后台运行几分钟后被杀死Android 后台进程管理策略使用termux-wake-lock保持 CPU 唤醒状态针对“任务卡在 processing”的问题这里给出一个回收脚本的思路 文件路径: agent-mesh/scripts/recover_stale.py 作用: 定时重置超时未完成的任务防止任务永远卡在 processing 状态。 import sqlite3 import os import time DB_PATH os.path.join(os.path.dirname(__file__), .., db, mesh.db) STALE_MINUTES 5 while True: conn sqlite3.connect(DB_PATH, timeout10) try: cursor conn.execute( UPDATE tasks SET status pending, current_agent NULL, updated_at datetime(now) WHERE status processing AND updated_at datetime(now, ?) , (f-{STALE_MINUTES} minutes,), ) conn.commit() if cursor.rowcount 0: print(f[recover] 重置了 {cursor.rowcount} 条超时任务) finally: conn.close() time.sleep(60)这个脚本每 60 秒执行一次把updated_at超过 5 分钟还处于processing状态的任务重新放回待处理队列。配合attempts字段你还可以限制每个任务的最大重试次数超过次数直接标记为failed。针对“进程被系统杀掉”的问题Termux 提供了唤醒锁# 在 Termux 中保持 CPU 唤醒状态 termux-wake-lock # 取消唤醒锁 termux-wake-unlock运行 agent 之前先执行termux-wake-lock可以降低息屏后进程被冻结的概率。另外在部分安卓系统上你还需要在系统设置中把 Termux 的“后台运行”权限设为允许否则即使有 wake lock系统仍然可能干预进程。还有一点需要特别注意Termux 是一个用户态环境/sys/class/thermal/中的温度节点在部分设备上可能没有读取权限。如果你遇到Permission denied不要尝试 root 或绕过权限换个思路改用termux-battery-status的电池温度作为热保护依据也能达到差不多的效果。7. 性能优化与最佳实践7.1 SQLite 写入性能优化SQLite 在默认的 delete 模式刷新方式下每次事务提交都要等待数据写入磁盘性能较差。启用 WAL 模式后读操作和写操作可以并发执行写入性能提升明显。初始化脚本已经设置了PRAGMA journal_mode WAL但如果你的库已经创建可以单独执行sqlite3 db/mesh.db PRAGMA journal_modeWAL;此外建议把synchronous设置为NORMALsqlite3 db/mesh.db PRAGMA synchronousNORMAL;在 WAL 模式下NORMAL依然可以保证数据库不会损坏只是极端掉电场景下可能丢失最近的事务。对 agent mesh 这种允许任务重放的场景来说这个权衡是值得的。7.2 Agent 数量与轮询间隔的平衡在手机上跑 10 层 mesh不一定每层都要启动一个独立进程。如果你的手机只有 4 个性能核心同时跑 10 个 Python 进程会造成频繁的线程切换温度很容易失控。比较合理的做法是先启动 6 到 8 个 Agent让每个 Agent 监听一个 tier跑通流程后再根据温度数据逐步增加。轮询间隔的设置也很关键。对于tier_1这种入口层如果任务到达频率很高可以保持 1 秒轮询。对于tier_5、tier_6这种中间层如果任务量不大把轮询间隔调到 3 到 5 秒完全够用。这样可以显著降低 CPU 空转率。7.3 手机硬件层面的控温建议软件层面的降载只能解决一部分问题下面几条硬件层面的建议对实际控温非常有效不要边充电边跑任务。充电过程本身会产生大量热量叠加 CPU 负载后温度很容易失控。取下手机壳或者把手机放在金属支架上有利于被动散热。关闭后台不相关的 App特别是短视频、社交类应用它们会周期性唤醒 CPU。如果条件允许使用小风扇对着手机背面吹效果立竿见影。在系统设置中开启飞行模式关闭蜂窝网络和 Wi-Fi 热点避免网络模块持续工作发热。7.4 日志与可观测性多 Agent 系统排错时最怕“黑盒”。建议在项目中加入统一的日志格式至少包含以下信息时间戳。Agent ID。任务 ID。当前层级。耗时。温度。在agent_worker.py中可以把print替换为 logging 模块import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.FileHandler(logs/agent.log), logging.StreamHandler(), ], ) logger logging.getLogger(agent)这样每个 Agent 的日志都会写入logs/agent.log。排查问题时用grep过滤关键字grep 任务 5 logs/agent.log可以快速定位某个任务经过了哪些 Agent。7.5 数据备份与迁移SQLite 单文件备份非常简单。如果 Agent 处理的是重要数据建议用定时任务把db/mesh.db复制到外部存储或使用termux-api上传到自己的服务器。cp db/mesh.db db/mesh-$(date %Y%m%d-%H%M%S).db如果要在另一台手机上恢复只需把备份文件复制到对应的db/目录即可。注意备份时最好先执行一次PRAGMA wal_checkpoint;把 WAL 文件中的内容合并到主库文件中。7.6 长跑前的压测思路正式让 10 层 mesh 在后台长时间运行之前建议先做一个小规模压测。具体做法是用init_db.py的seed_tasks函数写入 100 条任务然后在logs/temperature.log中观察任务完成时间和温度变化。如果 100 条任务在 10 分钟内完成且温度控制在 75°C 以下再扩展到 500 条、1000 条任务。要批量写入更多演示数据可以修改seed_tasks(count500)python -c from scripts.init_db import seed_tasks; seed_tasks(500)压测过程中要重点关注两个指标任务平均流转时间和系统温度峰值。如果任务流转时间随着运行时间推移越来越长说明系统已经进入热节流状态需要减少 Agent 数量或调大休眠阈值。8. 总结与下一步方向这篇文章完整记录了一套在 Termux 中搭建 10 层 SQLite agent mesh 的方案。核心思路是用 SQLite 作为多 Agent 之间共享的任务状态存储用带退避策略的数据库轮询替代传统消息队列用温度感知模块在热节流触发前主动降载。需要明确的是这个方案并不是要取代完整的分布式任务队列它更适合的实验场景是手头只有一台旧手机想验证多层 Agent 协作的流程设计或者想把手上的安卓设备变成一台低功耗开发机。在 10 层流水线全部跑通之后你可以继续尝试的方向包括在 Agent 中接入实际的 HTTP API让不同层的 Agent 调用不同的远程服务把 payload 中的任务类型扩展为多个让同一层的不同 Agent 处理不同任务使用asyncio重写 Agent 主循环在保持单进程的前提下处理更多任务。如果你在实操中遇到温度控制效果不明显的情况先不要急着加更复杂的算法回到最基础的三件事减少 Agent 数量、拉长轮询间隔、让手机物理散热更好。这个项目本身就是一场实验不断调整参数、观察温度曲线、记录任务流转时间你会慢慢找到适合自己手机的最佳配置。如果这篇文章对你有帮助可以把 init 脚本和 agent 工作循环代码保存下来直接在 Termux 里跑一遍然后打开温度日志看看你的手机能承受多大的并发量。