这次我们来看一个很有意思的落地场景在 Termux 里用 SQLite 搭一个 10 层 agent mesh同时想办法不让手机被热节流拖死。先说结论这个方案的可行点在于SQLite 作为多级 agent 之间的状态存储和任务队列完全够用真正的瓶颈往往不是数据库而是手机 SoC 的散热和系统调度策略。所以这篇文章不会只讲“装个 SQLite 跑个脚本”而是会分三块第一SQLite 在 agent mesh 里怎么设计表结构和任务流转第二Termux 环境下怎么启动、验证、调接口第三怎么通过监控温度、背压调度和小批量任务尽量避免 thermal throttling。文章会覆盖环境准备、数据库初始化、worker 示例、API 调用示例、批量任务脚本、热节流监控和常见问题排查。感兴趣的读者可以直接照做最小闭环跑通后再往 10 层扩展。1. 核心能力速览先给一张规格速览表方便快速判断这套体系适不适合你。能力项说明项目类型Termux 环境下的多智能体网格agent mesh SQLite 状态存储主要功能多级 agent 任务流转、SQLite 持久化、HTTP API 提交任务、批量任务入队、热节流监控推荐硬件支持 Termux 的 Android 设备内存建议 6GB 以上高负载场景建议有散热条件或空调环境显存需求无 GPU 依赖CPU 推理即可如果 agent 内部再接本地模型则需要按模型单独评估支持平台Android TermuxmacOS/Linux 的类 Unix 环境也适用需调整部分路径启动方式命令行启动可选 Python 内置 HTTP Server 作为 API 服务是否支持 API支持可以基于 http.server 或 Flask 自行封装是否支持批量任务支持SQLite 任务表天然适合批量入队和状态更新数据库要求SQLite 3.31开启 WAL 模式设置 busy_timeout降低并发写锁冲突适合场景多 agent 流程编排、任务队列、本地自动化实验、边缘设备轻量调度、教学演示从材料来看这个方案的重点不是做一个高性能分布式任务系统而是在手机这种弱设备上用 SQLite 把“10 层 agent 网格”这件事跑通。因此判断标准不是吞吐量而是稳定性和可复现性。2. 适用场景与使用边界2.1 适合谁这类方案比较适合下面几类人在手机或平板上折腾 Termux 的用户想验证 SQLite 能不能承担多 agent 任务编排。做 agent 流程实验的开发者不需要引入 Redis、RabbitMQ 这类重量级中间件想用 SQLite 快速落地一个任务队列。希望在边缘设备上跑自动化流程的爱好者比如定时采集、文本处理、数据清洗等轻任务。想学习 agent mesh 概念的人SQLite 的表结构设计本身就是很好的教学样例。2.2 能解决什么问题SQLite 在 agent mesh 里的价值很直接用一个数据库文件把 10 层 agent 的状态、任务、结果全部串起来。每一层 agent 不再需要单独维护自己的持久化文件也不需要复杂的网络通信协议直接读写同一张任务表即可。这样设计的好处有三点状态可追溯任务从 tier 1 流转到 tier 10每一步都能在 tasks 表里看到当前层、状态和更新时间。故障恢复简单worker 挂掉后任务停留在 processing 状态重启后可以把超时任务重置回 pending。批量任务友好一次插入大量 task_key然后逐个处理天然支持批处理场景。2.3 不适合什么场景这套方案不适合高并发、强一致、大流量场景。SQLite 单写多读虽然有 WAL 模式但写入并发能力有限。10 层 agent mesh 如果每层有多个 worker 同时高频率更新任务表写锁冲突会明显增加。更稳妥的做法是控制每层 worker 数量或者在任务表中加批次号让同一个批次尽量只由一个 worker 处理。2.4 合规与安全边界在 Termux 里做自动化实验时需要特别注意使用边界只处理自己拥有或有合法授权的数据不要采集、调用或爬取第三方未授权内容。如果 agent 涉及调用外部 API需要确认该 API 的调用条款和频率限制。不要利用 Termux 做任何网络攻击、未授权访问、破解、渗透或绕过系统安全机制的操作。涉及个人信息、人脸、声音、日志等敏感数据时必须做脱敏处理只在本地测试环境中使用。在公共网络或共享设备上运行时API 服务必须绑定 127.0.0.1不要暴露到公网。3. Termux 环境准备与前置条件3.1 系统与版本要求Android 设备建议 Android 10 及以上保证 Termux 的存储和系统调用权限可用。Termux 版本建议从 F-Droid 官方渠道更新避免旧版 dependency 不适配。需要安装 packagessqlite、python、python-pip、git、curl。如果后续要用 Python 的第三方库再按需通过pip安装。3.2 检查基础环境打开 Termux 后先执行pkg update pkg upgrade -y pkg install -y sqlite python python-pip git curl sqlite3 --version python --version看到 sqlite3 和 python 的版本输出说明基础环境没问题。如果需要把 Termux 的存储权限打开执行termux-setup-storage弹出系统授权窗口后允许访问存储后续建项目目录会放到~/storage或者普通家目录下。3.3 项目目录规划建议单独建目录避免脚本和数据库文件散落mkdir -p ~/agent_mesh/{db,logs,workers} cd ~/agent_mesh目录结构预留好之后数据库文件放db/日志放logs/Python worker 脚本放workers/。4. 10 层 Agent Mesh SQLite 架构设计4.1 什么是 agent meshagent mesh 不是传统的主从架构而是让多个 agent 节点按层级组织每个节点可以独立处理任务也可以把任务交给下一层继续处理。10 层意味着任务从输入层开始依次经过多个处理层最终输出结果。每一层可以有不同的职责比如tier 1任务接收、参数解析。tier 2文本分类、意图识别。tier 3工具调用、信息查询。tier 4信息汇总、上下文拼接。tier 5结果校验、格式修正。tier 6内容生成、模板填充。tier 7敏感信息过滤。tier 8质量评分。tier 9最终组装。tier 10输出归档。实际实现时不需要所有层都是复杂逻辑可以先用简单规则函数代替重点是把“层与层之间的任务流转”跑通。4.2 SQLite 表结构设计这里给出推荐的 3 张核心表agents记录每个 agent 节点的 ID、所属层级、当前状态、心跳时间。tasks记录每个任务的当前层级、目标层级、状态、payload 和重试次数。agent_logs记录每层 agent 的处理日志方便追踪问题。初始化 SQLCREATE TABLE IF NOT EXISTS agents ( id INTEGER PRIMARY KEY AUTOINCREMENT, tier INTEGER NOT NULL, node_id TEXT NOT NULL UNIQUE, status TEXT NOT NULL DEFAULT idle, last_heartbeat INTEGER, created_at INTEGER ); CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_key TEXT NOT NULL UNIQUE, current_tier INTEGER NOT NULL DEFAULT 1, target_tier INTEGER NOT NULL DEFAULT 10, status TEXT NOT NULL DEFAULT pending, payload TEXT, attempts INTEGER NOT NULL DEFAULT 0, created_at INTEGER, updated_at INTEGER ); CREATE INDEX IF NOT EXISTS idx_tasks_queue ON tasks(status, current_tier); CREATE TABLE IF NOT EXISTS agent_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, tier INTEGER NOT NULL, node_id TEXT NOT NULL, log_message TEXT, created_at INTEGER );把这段 SQL 保存为db/init.sql然后执行sqlite3 db/agent_mesh.db db/init.sql执行完后可以验证sqlite3 db/agent_mesh.db .tables输出结果里应该包含agents、tasks、agent_logs三张表。4.3 任务状态流转任务状态建议用以下几类状态含义pending等待当前层 worker 处理processing当前层 worker 正在处理done已到达目标层级任务完成failed任务处理失败等待重试或人工处理每次 worker 取任务时把 pending 改成 processing处理完成后如果还有下一层就把 current_tier 1状态改回 pending如果已经是 target_tier直接改成 done。这个状态机的核心逻辑很轻但它是整个 agent mesh 的骨架。只要有这一套状态流转10 层还是 20 层差别只在于循环次数。5. 安装部署与启动方式5.1 数据库初始化把上面db/init.sql写好之后执行一次初始化。如果数据库已经存在SQLite 会跳过已存在的表所以重复执行是安全的。5.2 Python worker 示例在workers/目录下创建worker.py实现任务获取、处理和流转import sqlite3 import time import json DB_PATH db/agent_mesh.db TIER_ID 1 def get_connection(): conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.execute(PRAGMA synchronousNORMAL;) return conn def fetch_next_task(tier): conn get_connection() try: cur conn.execute( SELECT id, task_key, payload FROM tasks WHERE statuspending AND current_tier? ORDER BY created_at ASC LIMIT 1, (tier,) ) row cur.fetchone() if row: conn.execute( UPDATE tasks SET statusprocessing, updated_at? WHERE id?, (int(time.time()), row[0]) ) conn.commit() return row finally: conn.close() def complete_task(task_id, result_payload): conn get_connection() try: cur conn.execute( SELECT current_tier, target_tier FROM tasks WHERE id?, (task_id,) ) row cur.fetchone() if not row: return current_tier, target_tier row next_tier current_tier 1 if next_tier target_tier: new_status done else: new_status pending conn.execute( UPDATE tasks SET status?, current_tier?, payload?, updated_at? WHERE id?, (new_status, next_tier, json.dumps(result_payload), int(time.time()), task_id) ) conn.commit() finally: conn.close() def process_payload(payload): # 这里是每一层 agent 的实际业务逻辑按需替换 data payload if isinstance(payload, dict) else {} data[processed_by_tier] TIER_ID return data def run_loop(): while True: task fetch_next_task(TIER_ID) if not task: time.sleep(2) continue task_id, task_key, payload_str task try: payload json.loads(payload_str) if payload_str else {} result process_payload(payload) complete_task(task_id, result) print(f[{time.strftime(%H:%M:%S)}] task {task_key} done at tier {TIER_ID}) except Exception as exc: print(f[ERROR] task {task_key} failed: {exc}) conn get_connection() conn.execute( UPDATE tasks SET attemptsattempts1, statusfailed, updated_at? WHERE id?, (int(time.time()), task_id) ) conn.commit() conn.close() time.sleep(1) if __name__ __main__: run_loop()启动 workercd ~/agent_mesh python workers/worker.py5.3 插入测试任务另开一个终端插入一条测试数据sqlite3 db/agent_mesh.db INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(task_001, 1, 10, pending, {\text\:\hello\}, strftime(%s,now), strftime(%s,now));worker 终端里如果打印了task task_001 done at tier 1说明任务已经被拉起来。之后再插入几条任务worker 会逐条处理直到 current_tier 超过 target_tier状态自动变为 done。6. 功能测试与效果验证6.1 测试目标验证 5 件事任务能够被 worker 获取并处理。任务能逐层递增到 target_tier。任务到达目标层级后状态变成 done。多个连续任务能被依次处理。异常任务会被标记为 failed不影响后续任务。6.2 批量插入测试数据执行以下命令插入 20 条任务for i in $(seq 1 20); do sqlite3 db/agent_mesh.db INSERT OR IGNORE INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(batch_$i, 1, 5, pending, {\index\: $i}, strftime(%s,now), strftime(%s,now)); done然后观察 worker 输出正常情况下应该逐条处理。6.3 查询任务状态处理过程中可以查询任务状态sqlite3 -header -column db/agent_mesh.db SELECT task_key, current_tier, target_tier, status, attempts FROM tasks ORDER BY id DESC LIMIT 10;判断是否成功的标准statusdone的任务数量符合预期。current_tier等于target_tier。没有大量failed任务。处理过程中没有出现database is locked错误。6.4 模拟失败场景把一条任务的payload改成无法解析的内容例如直接插入空字符串或者非法 JSON然后观察 worker 是否会把任务标记为 failedsqlite3 db/agent_mesh.db INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(bad_task, 1, 3, pending, {bad json, strftime(%s,now), strftime(%s,now));预期结果worker 输出 ERROR 日志任务记录 attempts 1status 变为 failed。7. 接口 API 与批量任务7.1 使用 Python 内置 HTTP Server 暴露 APITermux 环境里不一定需要装 Flask。Python 标准库的http.server足够支撑轻量接口测试。在workers/目录下创建api_server.pyfrom http.server import BaseHTTPRequestHandler, HTTPServer import json import sqlite3 import time DB_PATH db/agent_mesh.db def init_db(): conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.close() class Handler(BaseHTTPRequestHandler): def do_POST(self): if self.path ! /api/task: self.send_response(404) self.end_headers() return length int(self.headers.get(Content-Length, 0)) body json.loads(self.rfile.read(length)) task_key body.get(task_key) target_tier body.get(target_tier, 10) payload body.get(payload, {}) if not task_key: self.send_response(400) self.end_headers() self.wfile.write(btask_key is required) return conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) cur conn.execute( INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(?,?,?,?,?,?,?), (task_key, 1, target_tier, pending, json.dumps(payload, ensure_asciiFalse), int(time.time()), int(time.time())) ) conn.commit() conn.close() self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(json.dumps({task_id: cur.lastrowid}).encode()) def do_GET(self): if self.path /health: self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(b{\status\:\ok\}) return self.send_response(404) self.end_headers() def log_message(self, format, *args): pass if __name__ __main__: init_db() server HTTPServer((127.0.0.1, 8745), Handler) print(API server running on http://127.0.0.1:8745) server.serve_forever()启动 API 服务cd ~/agent_mesh python workers/api_server.py7.2 用 curl 提交任务curl -s -X POST http://127.0.0.1:8745/api/task \ -H Content-Type: application/json \ -d {task_key:api_001,target_tier:10,payload:{text:hello agent}}预期返回{task_id: 21}7.3 用 Python requests 提交批量任务如果安装了 requests可以这样批量提交import requests import time url http://127.0.0.1:8745/api/task for i in range(50): resp requests.post(url, json{ task_key: fpy_batch_{i}, target_tier: 5, payload: {index: i} }, timeout10) print(i, resp.json()) time.sleep(0.5)没有安装 requests 时可以用标准库urllib代替这里先给出 requests 示例方便理解。7.4 批量任务的调度策略批量任务最怕两件事数据库写锁集中和 CPU 持续高负载。建议在提交端做限速比如每 0.5 秒或 1 秒提交一条在 worker 端控制处理速率处理完一条任务后 sleep 0.5 到 1 秒。这样既降低 SQLite 锁冲突概率也减少瞬时发热。8. 避免热节流温度监控与背压调度8.1 热节流发生的原理移动设备没有桌面级散热条件CPU 或 GPU 温度超过阈值后系统会主动降频这就是 thermal throttling。表现是任务处理变慢、卡顿、后台进程被系统回收。如果只是简单跑一个 SQLite 写入脚本没那么容易触发但 10 层 agent mesh 如果开了多个 worker 持续跑CPU 高负载会明显拉升温度。8.2 读取 CPU 温度在多数 Android 设备上可以尝试读取热区温度接口cat /sys/class/thermal/thermal_zone0/temp输出通常是毫摄氏度比如45000表示 45 摄氏度。不同设备 thermal_zone 编号不同有的叫thermal_zone1或thermal_zone10。建议多试几个路径读取到有效值后记录下来。8.3 Python 侧温度监控与冷却等待在 worker 里增加温度判断超过阈值就暂停任务处理import os import time import glob def read_cpu_temp(): # 遍历常见 thermal zone 路径读取第一个有效温度 for path in glob.glob(/sys/class/thermal/thermal_zone*/temp): try: with open(path, r) as f: raw int(f.read().strip()) return raw / 1000.0 except Exception: continue return 0.0 def wait_for_cooldown(max_temp55.0): while True: temp read_cpu_temp() if temp max_temp: return print(f[thermal] temp {temp:.1f}C, waiting for cooldown...) time.sleep(5)然后在 run_loop 的 while 循环开头调用def run_loop(): while True: wait_for_cooldown(max_temp55.0) task fetch_next_task(TIER_ID) # 后续逻辑不变这里的max_temp阈值需要根据设备调整。身边有空调或者开了风扇时可以把阈值调到 60无散热环境建议调低到 50避免前面触发系统级热节流。8.4 进程优先级与系统限制如果设备 root 可用可以通过nice调整进程优先级降低对前台系统进程的影响。没有 root 时Termux 可以尝试用termux-wake-lock保持 CPU 唤醒termux-wake-lock处理完任务后释放termux-wake-unlock注意保持唤醒状态会增加功耗长时间运行时建议外接电源并放在散热良好的平面上。8.5 控制并发 Worker 数量如果数据库连接数过多SQLite 写锁冲突会更明显。更稳妥的做法是每个 tier 最多一个 worker 进程全局最多 10 到 12 个 worker 进程。每次取任务前增加随机退避时间避免多个 worker 同时查到同一条 pending 任务。示例import random time.sleep(random.uniform(0.2, 1.0))8.6 优化 SQLite 写入参数把数据库连接配置固定下来conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.execute(PRAGMA synchronousNORMAL;) conn.execute(PRAGMA cache_size-16000;)注意synchronousNORMAL在 WAL 模式下的崩溃恢复能力仍然可以接受但比 FULL 模式更能降低写入等待。如果对数据安全性要求更高可以保持synchronousFULL。9. 资源占用与性能观察9.1 观察 CPU 和内存在 Termux 里可以通过top查看进程资源占用top -n 1重点看 Python 进程的 CPU 百分比和 RES 内存。9.2 观察数据库连接数SQLite 不提供直接的系统连接数查询但可以通过lsof查看打开 db 文件的进程数lsof 2/dev/null | grep agent_mesh.db如果大量进程同时打开同一个文件说明并发控制需要加强。9.3 任务处理速率通过查询 tasks 表里 done 状态的任务数量变化估算速率sqlite3 db/agent_mesh.db SELECT status, COUNT(*) FROM tasks GROUP BY status;观察分钟级变化如果速率突然下降先看温度再看数据库锁日志。9.4 如何降低高负载调小批量提交速率。增加 worker 内部 sleep 时间。关闭不必要的动画和后台 App。使用充电器供电降低系统功耗限制对性能的影响。如果只是做接口验证把 target_tier 改成 3 或 5减少重复层数处理。10. 常见问题与排查方法问题现象可能原因排查方式解决方案sqlite3 命令不存在未安装 sqlite执行which sqlite3pkg install sqlitePython 脚本找不到模块依赖未安装或路径错误检查python -c import sqlite3确认在项目根目录运行或设置PYTHONPATH启动 API 后 127.0.0.1:8745 无法访问端口被占用或服务未启动执行netstat -tlnp查看端口换端口或杀掉占用进程worker 报 database is locked写入并发过高查看日志中的锁错误频率增加 busy_timeout开启 WAL减少 worker 并发任务长时间停在 processingworker 崩溃或任务卡住查询 tasks 表该任务的状态写一个重启脚本把超时 processing 任务重置为 pending设备明显发烫任务变慢thermal throttling 触发读取温度观察 CPU 频率下降增加冷却等待降低任务密度插入任务后 worker 没有反应任务不在当前 tier检查 current_tier 和 worker 的 TIER_ID修改插入任务的 current_tier 或 worker 层级API 返回 400 缺少参数请求体格式不对检查 curl 或 requests 参数确认 task_key 和 payload 字段名数据库文件突然变大日志表和任务表积累历史数据查询各表行数定期清理已完成的 tasks 和 agent_logsTermux 后台被杀系统内存回收使用 termux-wake-lock保持前台/锁屏唤醒或外接电源持续运行额外提醒如果在 API 服务里返回了 Windows 风格的换行符或者代码从本地复制到 Termux 出现格式问题可以先在编辑器里统一转换为 LF 换行避免 Python 出现\r解释错误。11. 最佳实践与使用建议11.1 先跑最小闭环第一次测试不要直接上 10 层。先把 target_tier 改成 2插入一条任务确认 worker 能把它从 tier 1 推到 tier 2 并变成 done。这个闭环跑通之后再逐步放大到 5 层、10 层。11.2 目录管理建议推荐这样的目录结构~/agent_mesh/ ├── db/ │ ├── init.sql │ └── agent_mesh.db ├── logs/ │ └── worker.log ├── workers/ │ ├── worker.py │ └── api_server.py └── scripts/ └── submit_batch.py数据库文件和代码分开日志单独存放批量脚本单独管理后续排查问题时能快速定位。11.3 批量任务要加日志不要让 worker 只是静默处理。至少在每个任务开始和结束时打印一行日志。如果批量任务数量大建议把 stdout 重定向到日志文件python workers/worker.py logs/worker.log 21 11.4 失败重试策略任务失败后直接标记 failed 是最简单的做法但更稳妥的做法是如果 attempts 小于 3把任务重置回 pending 并让后续 worker 再试一次。示例UPDATE tasks SET statuspending, attemptsattempts1, updated_atstrftime(%s,now) WHERE id? AND attempts 3;11.5 安全说明再次强调如果未来要把这个 agent mesh 接到外部服务务必确认每个调用都有授权不要在这次实验里导入或处理非授权数据API 服务绑定在 127.0.0.1这是 Termux 环境里比较稳妥的做法。到这里整个方案的骨架、实现和常见问题都已经覆盖完整。最容易踩的坑是同时开太多 worker导致 SQLite 写锁和发热一起出现。建议先把最少可用版本跑通再加入温度监控和冷却等待。先去验证单条任务从 tier 1 走到 tier 10再考虑扩大批量和处理速度这是手机端 agent mesh 实验最稳的路径。 Termux 中用 SQLite 搭建 10 层 Agent Mesh 与热节流应对实践
这次我们来看一个很有意思的落地场景在 Termux 里用 SQLite 搭一个 10 层 agent mesh同时想办法不让手机被热节流拖死。先说结论这个方案的可行点在于SQLite 作为多级 agent 之间的状态存储和任务队列完全够用真正的瓶颈往往不是数据库而是手机 SoC 的散热和系统调度策略。所以这篇文章不会只讲“装个 SQLite 跑个脚本”而是会分三块第一SQLite 在 agent mesh 里怎么设计表结构和任务流转第二Termux 环境下怎么启动、验证、调接口第三怎么通过监控温度、背压调度和小批量任务尽量避免 thermal throttling。文章会覆盖环境准备、数据库初始化、worker 示例、API 调用示例、批量任务脚本、热节流监控和常见问题排查。感兴趣的读者可以直接照做最小闭环跑通后再往 10 层扩展。1. 核心能力速览先给一张规格速览表方便快速判断这套体系适不适合你。能力项说明项目类型Termux 环境下的多智能体网格agent mesh SQLite 状态存储主要功能多级 agent 任务流转、SQLite 持久化、HTTP API 提交任务、批量任务入队、热节流监控推荐硬件支持 Termux 的 Android 设备内存建议 6GB 以上高负载场景建议有散热条件或空调环境显存需求无 GPU 依赖CPU 推理即可如果 agent 内部再接本地模型则需要按模型单独评估支持平台Android TermuxmacOS/Linux 的类 Unix 环境也适用需调整部分路径启动方式命令行启动可选 Python 内置 HTTP Server 作为 API 服务是否支持 API支持可以基于 http.server 或 Flask 自行封装是否支持批量任务支持SQLite 任务表天然适合批量入队和状态更新数据库要求SQLite 3.31开启 WAL 模式设置 busy_timeout降低并发写锁冲突适合场景多 agent 流程编排、任务队列、本地自动化实验、边缘设备轻量调度、教学演示从材料来看这个方案的重点不是做一个高性能分布式任务系统而是在手机这种弱设备上用 SQLite 把“10 层 agent 网格”这件事跑通。因此判断标准不是吞吐量而是稳定性和可复现性。2. 适用场景与使用边界2.1 适合谁这类方案比较适合下面几类人在手机或平板上折腾 Termux 的用户想验证 SQLite 能不能承担多 agent 任务编排。做 agent 流程实验的开发者不需要引入 Redis、RabbitMQ 这类重量级中间件想用 SQLite 快速落地一个任务队列。希望在边缘设备上跑自动化流程的爱好者比如定时采集、文本处理、数据清洗等轻任务。想学习 agent mesh 概念的人SQLite 的表结构设计本身就是很好的教学样例。2.2 能解决什么问题SQLite 在 agent mesh 里的价值很直接用一个数据库文件把 10 层 agent 的状态、任务、结果全部串起来。每一层 agent 不再需要单独维护自己的持久化文件也不需要复杂的网络通信协议直接读写同一张任务表即可。这样设计的好处有三点状态可追溯任务从 tier 1 流转到 tier 10每一步都能在 tasks 表里看到当前层、状态和更新时间。故障恢复简单worker 挂掉后任务停留在 processing 状态重启后可以把超时任务重置回 pending。批量任务友好一次插入大量 task_key然后逐个处理天然支持批处理场景。2.3 不适合什么场景这套方案不适合高并发、强一致、大流量场景。SQLite 单写多读虽然有 WAL 模式但写入并发能力有限。10 层 agent mesh 如果每层有多个 worker 同时高频率更新任务表写锁冲突会明显增加。更稳妥的做法是控制每层 worker 数量或者在任务表中加批次号让同一个批次尽量只由一个 worker 处理。2.4 合规与安全边界在 Termux 里做自动化实验时需要特别注意使用边界只处理自己拥有或有合法授权的数据不要采集、调用或爬取第三方未授权内容。如果 agent 涉及调用外部 API需要确认该 API 的调用条款和频率限制。不要利用 Termux 做任何网络攻击、未授权访问、破解、渗透或绕过系统安全机制的操作。涉及个人信息、人脸、声音、日志等敏感数据时必须做脱敏处理只在本地测试环境中使用。在公共网络或共享设备上运行时API 服务必须绑定 127.0.0.1不要暴露到公网。3. Termux 环境准备与前置条件3.1 系统与版本要求Android 设备建议 Android 10 及以上保证 Termux 的存储和系统调用权限可用。Termux 版本建议从 F-Droid 官方渠道更新避免旧版 dependency 不适配。需要安装 packagessqlite、python、python-pip、git、curl。如果后续要用 Python 的第三方库再按需通过pip安装。3.2 检查基础环境打开 Termux 后先执行pkg update pkg upgrade -y pkg install -y sqlite python python-pip git curl sqlite3 --version python --version看到 sqlite3 和 python 的版本输出说明基础环境没问题。如果需要把 Termux 的存储权限打开执行termux-setup-storage弹出系统授权窗口后允许访问存储后续建项目目录会放到~/storage或者普通家目录下。3.3 项目目录规划建议单独建目录避免脚本和数据库文件散落mkdir -p ~/agent_mesh/{db,logs,workers} cd ~/agent_mesh目录结构预留好之后数据库文件放db/日志放logs/Python worker 脚本放workers/。4. 10 层 Agent Mesh SQLite 架构设计4.1 什么是 agent meshagent mesh 不是传统的主从架构而是让多个 agent 节点按层级组织每个节点可以独立处理任务也可以把任务交给下一层继续处理。10 层意味着任务从输入层开始依次经过多个处理层最终输出结果。每一层可以有不同的职责比如tier 1任务接收、参数解析。tier 2文本分类、意图识别。tier 3工具调用、信息查询。tier 4信息汇总、上下文拼接。tier 5结果校验、格式修正。tier 6内容生成、模板填充。tier 7敏感信息过滤。tier 8质量评分。tier 9最终组装。tier 10输出归档。实际实现时不需要所有层都是复杂逻辑可以先用简单规则函数代替重点是把“层与层之间的任务流转”跑通。4.2 SQLite 表结构设计这里给出推荐的 3 张核心表agents记录每个 agent 节点的 ID、所属层级、当前状态、心跳时间。tasks记录每个任务的当前层级、目标层级、状态、payload 和重试次数。agent_logs记录每层 agent 的处理日志方便追踪问题。初始化 SQLCREATE TABLE IF NOT EXISTS agents ( id INTEGER PRIMARY KEY AUTOINCREMENT, tier INTEGER NOT NULL, node_id TEXT NOT NULL UNIQUE, status TEXT NOT NULL DEFAULT idle, last_heartbeat INTEGER, created_at INTEGER ); CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_key TEXT NOT NULL UNIQUE, current_tier INTEGER NOT NULL DEFAULT 1, target_tier INTEGER NOT NULL DEFAULT 10, status TEXT NOT NULL DEFAULT pending, payload TEXT, attempts INTEGER NOT NULL DEFAULT 0, created_at INTEGER, updated_at INTEGER ); CREATE INDEX IF NOT EXISTS idx_tasks_queue ON tasks(status, current_tier); CREATE TABLE IF NOT EXISTS agent_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, tier INTEGER NOT NULL, node_id TEXT NOT NULL, log_message TEXT, created_at INTEGER );把这段 SQL 保存为db/init.sql然后执行sqlite3 db/agent_mesh.db db/init.sql执行完后可以验证sqlite3 db/agent_mesh.db .tables输出结果里应该包含agents、tasks、agent_logs三张表。4.3 任务状态流转任务状态建议用以下几类状态含义pending等待当前层 worker 处理processing当前层 worker 正在处理done已到达目标层级任务完成failed任务处理失败等待重试或人工处理每次 worker 取任务时把 pending 改成 processing处理完成后如果还有下一层就把 current_tier 1状态改回 pending如果已经是 target_tier直接改成 done。这个状态机的核心逻辑很轻但它是整个 agent mesh 的骨架。只要有这一套状态流转10 层还是 20 层差别只在于循环次数。5. 安装部署与启动方式5.1 数据库初始化把上面db/init.sql写好之后执行一次初始化。如果数据库已经存在SQLite 会跳过已存在的表所以重复执行是安全的。5.2 Python worker 示例在workers/目录下创建worker.py实现任务获取、处理和流转import sqlite3 import time import json DB_PATH db/agent_mesh.db TIER_ID 1 def get_connection(): conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.execute(PRAGMA synchronousNORMAL;) return conn def fetch_next_task(tier): conn get_connection() try: cur conn.execute( SELECT id, task_key, payload FROM tasks WHERE statuspending AND current_tier? ORDER BY created_at ASC LIMIT 1, (tier,) ) row cur.fetchone() if row: conn.execute( UPDATE tasks SET statusprocessing, updated_at? WHERE id?, (int(time.time()), row[0]) ) conn.commit() return row finally: conn.close() def complete_task(task_id, result_payload): conn get_connection() try: cur conn.execute( SELECT current_tier, target_tier FROM tasks WHERE id?, (task_id,) ) row cur.fetchone() if not row: return current_tier, target_tier row next_tier current_tier 1 if next_tier target_tier: new_status done else: new_status pending conn.execute( UPDATE tasks SET status?, current_tier?, payload?, updated_at? WHERE id?, (new_status, next_tier, json.dumps(result_payload), int(time.time()), task_id) ) conn.commit() finally: conn.close() def process_payload(payload): # 这里是每一层 agent 的实际业务逻辑按需替换 data payload if isinstance(payload, dict) else {} data[processed_by_tier] TIER_ID return data def run_loop(): while True: task fetch_next_task(TIER_ID) if not task: time.sleep(2) continue task_id, task_key, payload_str task try: payload json.loads(payload_str) if payload_str else {} result process_payload(payload) complete_task(task_id, result) print(f[{time.strftime(%H:%M:%S)}] task {task_key} done at tier {TIER_ID}) except Exception as exc: print(f[ERROR] task {task_key} failed: {exc}) conn get_connection() conn.execute( UPDATE tasks SET attemptsattempts1, statusfailed, updated_at? WHERE id?, (int(time.time()), task_id) ) conn.commit() conn.close() time.sleep(1) if __name__ __main__: run_loop()启动 workercd ~/agent_mesh python workers/worker.py5.3 插入测试任务另开一个终端插入一条测试数据sqlite3 db/agent_mesh.db INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(task_001, 1, 10, pending, {\text\:\hello\}, strftime(%s,now), strftime(%s,now));worker 终端里如果打印了task task_001 done at tier 1说明任务已经被拉起来。之后再插入几条任务worker 会逐条处理直到 current_tier 超过 target_tier状态自动变为 done。6. 功能测试与效果验证6.1 测试目标验证 5 件事任务能够被 worker 获取并处理。任务能逐层递增到 target_tier。任务到达目标层级后状态变成 done。多个连续任务能被依次处理。异常任务会被标记为 failed不影响后续任务。6.2 批量插入测试数据执行以下命令插入 20 条任务for i in $(seq 1 20); do sqlite3 db/agent_mesh.db INSERT OR IGNORE INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(batch_$i, 1, 5, pending, {\index\: $i}, strftime(%s,now), strftime(%s,now)); done然后观察 worker 输出正常情况下应该逐条处理。6.3 查询任务状态处理过程中可以查询任务状态sqlite3 -header -column db/agent_mesh.db SELECT task_key, current_tier, target_tier, status, attempts FROM tasks ORDER BY id DESC LIMIT 10;判断是否成功的标准statusdone的任务数量符合预期。current_tier等于target_tier。没有大量failed任务。处理过程中没有出现database is locked错误。6.4 模拟失败场景把一条任务的payload改成无法解析的内容例如直接插入空字符串或者非法 JSON然后观察 worker 是否会把任务标记为 failedsqlite3 db/agent_mesh.db INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(bad_task, 1, 3, pending, {bad json, strftime(%s,now), strftime(%s,now));预期结果worker 输出 ERROR 日志任务记录 attempts 1status 变为 failed。7. 接口 API 与批量任务7.1 使用 Python 内置 HTTP Server 暴露 APITermux 环境里不一定需要装 Flask。Python 标准库的http.server足够支撑轻量接口测试。在workers/目录下创建api_server.pyfrom http.server import BaseHTTPRequestHandler, HTTPServer import json import sqlite3 import time DB_PATH db/agent_mesh.db def init_db(): conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.close() class Handler(BaseHTTPRequestHandler): def do_POST(self): if self.path ! /api/task: self.send_response(404) self.end_headers() return length int(self.headers.get(Content-Length, 0)) body json.loads(self.rfile.read(length)) task_key body.get(task_key) target_tier body.get(target_tier, 10) payload body.get(payload, {}) if not task_key: self.send_response(400) self.end_headers() self.wfile.write(btask_key is required) return conn sqlite3.connect(DB_PATH, timeout30) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) cur conn.execute( INSERT INTO tasks(task_key, current_tier, target_tier, status, payload, created_at, updated_at) VALUES(?,?,?,?,?,?,?), (task_key, 1, target_tier, pending, json.dumps(payload, ensure_asciiFalse), int(time.time()), int(time.time())) ) conn.commit() conn.close() self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(json.dumps({task_id: cur.lastrowid}).encode()) def do_GET(self): if self.path /health: self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(b{\status\:\ok\}) return self.send_response(404) self.end_headers() def log_message(self, format, *args): pass if __name__ __main__: init_db() server HTTPServer((127.0.0.1, 8745), Handler) print(API server running on http://127.0.0.1:8745) server.serve_forever()启动 API 服务cd ~/agent_mesh python workers/api_server.py7.2 用 curl 提交任务curl -s -X POST http://127.0.0.1:8745/api/task \ -H Content-Type: application/json \ -d {task_key:api_001,target_tier:10,payload:{text:hello agent}}预期返回{task_id: 21}7.3 用 Python requests 提交批量任务如果安装了 requests可以这样批量提交import requests import time url http://127.0.0.1:8745/api/task for i in range(50): resp requests.post(url, json{ task_key: fpy_batch_{i}, target_tier: 5, payload: {index: i} }, timeout10) print(i, resp.json()) time.sleep(0.5)没有安装 requests 时可以用标准库urllib代替这里先给出 requests 示例方便理解。7.4 批量任务的调度策略批量任务最怕两件事数据库写锁集中和 CPU 持续高负载。建议在提交端做限速比如每 0.5 秒或 1 秒提交一条在 worker 端控制处理速率处理完一条任务后 sleep 0.5 到 1 秒。这样既降低 SQLite 锁冲突概率也减少瞬时发热。8. 避免热节流温度监控与背压调度8.1 热节流发生的原理移动设备没有桌面级散热条件CPU 或 GPU 温度超过阈值后系统会主动降频这就是 thermal throttling。表现是任务处理变慢、卡顿、后台进程被系统回收。如果只是简单跑一个 SQLite 写入脚本没那么容易触发但 10 层 agent mesh 如果开了多个 worker 持续跑CPU 高负载会明显拉升温度。8.2 读取 CPU 温度在多数 Android 设备上可以尝试读取热区温度接口cat /sys/class/thermal/thermal_zone0/temp输出通常是毫摄氏度比如45000表示 45 摄氏度。不同设备 thermal_zone 编号不同有的叫thermal_zone1或thermal_zone10。建议多试几个路径读取到有效值后记录下来。8.3 Python 侧温度监控与冷却等待在 worker 里增加温度判断超过阈值就暂停任务处理import os import time import glob def read_cpu_temp(): # 遍历常见 thermal zone 路径读取第一个有效温度 for path in glob.glob(/sys/class/thermal/thermal_zone*/temp): try: with open(path, r) as f: raw int(f.read().strip()) return raw / 1000.0 except Exception: continue return 0.0 def wait_for_cooldown(max_temp55.0): while True: temp read_cpu_temp() if temp max_temp: return print(f[thermal] temp {temp:.1f}C, waiting for cooldown...) time.sleep(5)然后在 run_loop 的 while 循环开头调用def run_loop(): while True: wait_for_cooldown(max_temp55.0) task fetch_next_task(TIER_ID) # 后续逻辑不变这里的max_temp阈值需要根据设备调整。身边有空调或者开了风扇时可以把阈值调到 60无散热环境建议调低到 50避免前面触发系统级热节流。8.4 进程优先级与系统限制如果设备 root 可用可以通过nice调整进程优先级降低对前台系统进程的影响。没有 root 时Termux 可以尝试用termux-wake-lock保持 CPU 唤醒termux-wake-lock处理完任务后释放termux-wake-unlock注意保持唤醒状态会增加功耗长时间运行时建议外接电源并放在散热良好的平面上。8.5 控制并发 Worker 数量如果数据库连接数过多SQLite 写锁冲突会更明显。更稳妥的做法是每个 tier 最多一个 worker 进程全局最多 10 到 12 个 worker 进程。每次取任务前增加随机退避时间避免多个 worker 同时查到同一条 pending 任务。示例import random time.sleep(random.uniform(0.2, 1.0))8.6 优化 SQLite 写入参数把数据库连接配置固定下来conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout30000;) conn.execute(PRAGMA synchronousNORMAL;) conn.execute(PRAGMA cache_size-16000;)注意synchronousNORMAL在 WAL 模式下的崩溃恢复能力仍然可以接受但比 FULL 模式更能降低写入等待。如果对数据安全性要求更高可以保持synchronousFULL。9. 资源占用与性能观察9.1 观察 CPU 和内存在 Termux 里可以通过top查看进程资源占用top -n 1重点看 Python 进程的 CPU 百分比和 RES 内存。9.2 观察数据库连接数SQLite 不提供直接的系统连接数查询但可以通过lsof查看打开 db 文件的进程数lsof 2/dev/null | grep agent_mesh.db如果大量进程同时打开同一个文件说明并发控制需要加强。9.3 任务处理速率通过查询 tasks 表里 done 状态的任务数量变化估算速率sqlite3 db/agent_mesh.db SELECT status, COUNT(*) FROM tasks GROUP BY status;观察分钟级变化如果速率突然下降先看温度再看数据库锁日志。9.4 如何降低高负载调小批量提交速率。增加 worker 内部 sleep 时间。关闭不必要的动画和后台 App。使用充电器供电降低系统功耗限制对性能的影响。如果只是做接口验证把 target_tier 改成 3 或 5减少重复层数处理。10. 常见问题与排查方法问题现象可能原因排查方式解决方案sqlite3 命令不存在未安装 sqlite执行which sqlite3pkg install sqlitePython 脚本找不到模块依赖未安装或路径错误检查python -c import sqlite3确认在项目根目录运行或设置PYTHONPATH启动 API 后 127.0.0.1:8745 无法访问端口被占用或服务未启动执行netstat -tlnp查看端口换端口或杀掉占用进程worker 报 database is locked写入并发过高查看日志中的锁错误频率增加 busy_timeout开启 WAL减少 worker 并发任务长时间停在 processingworker 崩溃或任务卡住查询 tasks 表该任务的状态写一个重启脚本把超时 processing 任务重置为 pending设备明显发烫任务变慢thermal throttling 触发读取温度观察 CPU 频率下降增加冷却等待降低任务密度插入任务后 worker 没有反应任务不在当前 tier检查 current_tier 和 worker 的 TIER_ID修改插入任务的 current_tier 或 worker 层级API 返回 400 缺少参数请求体格式不对检查 curl 或 requests 参数确认 task_key 和 payload 字段名数据库文件突然变大日志表和任务表积累历史数据查询各表行数定期清理已完成的 tasks 和 agent_logsTermux 后台被杀系统内存回收使用 termux-wake-lock保持前台/锁屏唤醒或外接电源持续运行额外提醒如果在 API 服务里返回了 Windows 风格的换行符或者代码从本地复制到 Termux 出现格式问题可以先在编辑器里统一转换为 LF 换行避免 Python 出现\r解释错误。11. 最佳实践与使用建议11.1 先跑最小闭环第一次测试不要直接上 10 层。先把 target_tier 改成 2插入一条任务确认 worker 能把它从 tier 1 推到 tier 2 并变成 done。这个闭环跑通之后再逐步放大到 5 层、10 层。11.2 目录管理建议推荐这样的目录结构~/agent_mesh/ ├── db/ │ ├── init.sql │ └── agent_mesh.db ├── logs/ │ └── worker.log ├── workers/ │ ├── worker.py │ └── api_server.py └── scripts/ └── submit_batch.py数据库文件和代码分开日志单独存放批量脚本单独管理后续排查问题时能快速定位。11.3 批量任务要加日志不要让 worker 只是静默处理。至少在每个任务开始和结束时打印一行日志。如果批量任务数量大建议把 stdout 重定向到日志文件python workers/worker.py logs/worker.log 21 11.4 失败重试策略任务失败后直接标记 failed 是最简单的做法但更稳妥的做法是如果 attempts 小于 3把任务重置回 pending 并让后续 worker 再试一次。示例UPDATE tasks SET statuspending, attemptsattempts1, updated_atstrftime(%s,now) WHERE id? AND attempts 3;11.5 安全说明再次强调如果未来要把这个 agent mesh 接到外部服务务必确认每个调用都有授权不要在这次实验里导入或处理非授权数据API 服务绑定在 127.0.0.1这是 Termux 环境里比较稳妥的做法。到这里整个方案的骨架、实现和常见问题都已经覆盖完整。最容易踩的坑是同时开太多 worker导致 SQLite 写锁和发热一起出现。建议先把最少可用版本跑通再加入温度监控和冷却等待。先去验证单条任务从 tier 1 走到 tier 10再考虑扩大批量和处理速度这是手机端 agent mesh 实验最稳的路径。