流自救：用 SQLite 搭建 10 層 Agent Mesh 實(shí)戰(zhàn))
之前在 Termux 里跑并發(fā)任務(wù)時(shí)經(jīng)常遇到一個(gè)奇怪現(xiàn)象任務(wù)一多手機(jī)背面就開始發(fā)燙緊接著 CPU 頻率像被一只手按住所有進(jìn)程一起變慢。這個(gè)現(xiàn)象就是 thermal throttling熱節(jié)流。它不會直接報(bào)錯(cuò)但會讓你的多進(jìn)程系統(tǒng)從“流暢”變成“寸步難行”。本文將圍繞這個(gè)問題記錄一套在 Termux 中通過 SQLite 搭建 10 層 agent mesh 的完整方案包含可運(yùn)行的 Python 代碼、數(shù)據(jù)庫表設(shè)計(jì)、溫度監(jiān)控方法和進(jìn)程調(diào)度策略。無論你是想在舊手機(jī)上搭建一個(gè)實(shí)驗(yàn)性的多 Agent 任務(wù)系統(tǒng)還是準(zhǔn)備把 Termux 當(dāng)作低功耗開發(fā)環(huán)境這篇文章都值得收藏。1. 背景為什么要在手機(jī)上跑 10 層 Agent Mesh1.1 這個(gè)實(shí)驗(yàn)解決什么問題日常開發(fā)中當(dāng)我們提到“多 Agent 協(xié)作”第一反應(yīng)往往是 K8s、Docker Compose、Redis 消息隊(duì)列這些重型基礎(chǔ)設(shè)施。但如果你手里的設(shè)備只是一臺安卓手機(jī)沒有 root 權(quán)限也沒有完整的 Linux 內(nèi)核管理能力這些方案基本都跑不起來。Termux 提供了一個(gè)輕量級的 Linux 環(huán)境但它對進(jìn)程管理、系統(tǒng)服務(wù)、網(wǎng)絡(luò)端口綁定都有一定的限制。本文要探討的是如何不依賴復(fù)雜基礎(chǔ)設(shè)施僅用 Termux SQLite Python 就搭建出一個(gè) 10 層任務(wù)處理網(wǎng)格并且保證長時(shí)間運(yùn)行不觸發(fā)熱節(jié)流。一個(gè)典型的 agent mesh 場景是這樣的系統(tǒng)里有多個(gè) worker 節(jié)點(diǎn)每個(gè)節(jié)點(diǎn)處理特定類型的任務(wù)。任務(wù)從第一層進(jìn)入經(jīng)過抽取、清洗、轉(zhuǎn)換、計(jì)算、聚合等多個(gè)階段最終輸出結(jié)果。每一層都可能有多個(gè) agent 并發(fā)處理任務(wù)層與層之間通過共享的存儲介質(zhì)傳遞數(shù)據(jù)。在傳統(tǒng)架構(gòu)中這個(gè)存儲介質(zhì)可能是 Redis、RabbitMQ 或 Kafka但在 Termux 這種輕量環(huán)境里SQLite 反而成了最合適的選擇。1.2 SQLite 為什么適合做 Agent 總線很多人對 SQLite 的印象還停留在“單機(jī)小數(shù)據(jù)庫”覺得它撐不起并發(fā)場景。實(shí)際上SQLite 在 WALWrite-Ahead Logging模式下多進(jìn)程并發(fā)讀、單進(jìn)程寫的能力足以支撐中小規(guī)模的 Agent 任務(wù)調(diào)度。關(guān)鍵在于你怎么設(shè)計(jì)表結(jié)構(gòu)和事務(wù)粒度。SQLite 相比消息隊(duì)列有幾個(gè)不可替代的優(yōu)勢零配置Termux 里直接安裝就能用。數(shù)據(jù)庫即文件備份、遷移、查看都非常方便。支持事務(wù)、索引、唯一約束可以避免任務(wù)被重復(fù)消費(fèi)。一個(gè)文件就能保存全部任務(wù)狀態(tài)排查問題時(shí)用 db browser 打開就能看。缺點(diǎn)當(dāng)然也有寫入并發(fā)能力有限單條記錄大小不宜過大。但對于 agent mesh 這種以任務(wù)狀態(tài)流轉(zhuǎn)為主、消息體通常只有幾 KB 的場景SQLite 的瓶頸完全夠用。1.3 熱節(jié)流是最大的敵人在手機(jī)上跑任務(wù)最怕的不是 CPU 不夠快而是“發(fā)揮不出來”。手機(jī) SoC 有嚴(yán)格的溫度控制策略當(dāng)芯片溫度超過閾值時(shí)系統(tǒng)會主動(dòng)降低 CPU 頻率、限制核心調(diào)度甚至關(guān)閉高性能核心。這個(gè)機(jī)制在 x86 設(shè)備上通常叫 PROCHOTProcessor Hot信號在 ARM 設(shè)備上則是通過 thermal_zone 溫度節(jié)點(diǎn)與 governor 策略共同實(shí)現(xiàn)。熱節(jié)流帶來的問題非常隱蔽你的進(jìn)程并沒有崩潰但運(yùn)行速度可能會降到原來的十分之一。如果 agent 之間有超時(shí)機(jī)制還會引發(fā)連鎖的任務(wù)堆積和重復(fù)處理。所以在 Termux 中搭建多 Agent 系統(tǒng)第一優(yōu)先級不是“跑滿 CPU”而是“控制溫度”。2. 環(huán)境準(zhǔn)備Termux 里的最小開發(fā)棧2.1 安裝 Termux 并更新軟件源Termux 的安裝渠道比較特殊Google Play 上的版本已經(jīng)停止維護(hù)推薦從 F-Droid 或 GitHub Releases 下載最新 APK。安裝完成后首先執(zhí)行pkg update pkg upgrade -y這一步會更新軟件源和已安裝包。需要注意如果你的 Termux 是剛安裝的pkg命令可能需要先初始化存儲目錄執(zhí)行termux-setup-storage授權(quán)存儲權(quán)限。2.2 安裝 SQLite、Python 與監(jiān)控工具本文的實(shí)戰(zhàn)部分以 Python 為例需要安裝以下包pkg install python sqlite clang -y pkg install termux-api -y這里說明一下每個(gè)包的用途python用來編寫 Agent 工作循環(huán)。sqlite提供 SQLite 命令行工具方便調(diào)試和查詢。clangPython 的某些依賴需要編譯提前安裝可以避免報(bào)錯(cuò)。termux-api用來讀取電池溫度、充電狀態(tài)等系統(tǒng)信息是熱節(jié)流保護(hù)模塊的基礎(chǔ)。如果你希望 Agent 輸出更規(guī)范的日志還可以安裝pandas、pydantic這類包但本文為了保持依賴最小化只用 Python 標(biāo)準(zhǔn)庫。pip install --upgrade pip國內(nèi)網(wǎng)絡(luò)環(huán)境下pip 源如果不穩(wěn)定可以臨時(shí)使用清華鏡像pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple版本方面需要提醒一下Termux 的軟件源是滾動(dòng)更新的不同時(shí)間安裝的 Python 和 SQLite 版本可能不同。本文的代碼只使用 SQLite 的基礎(chǔ)語法和 Python 標(biāo)準(zhǔn)庫只要你的 SQLite 版本在 3.8 以上支持 WAL 模式基本不存在兼容性問題。2.3 規(guī)劃項(xiàng)目目錄建議按下面的結(jié)構(gòu)組織項(xiàng)目agent-mesh/ ├── db/ │ └── mesh.db ├── logs/ │ └── agent.log ├── scripts/ │ ├── init_db.py │ ├── agent_worker.py │ ├── thermal_guard.py │ └── dispatch.py └── README.md在 Termux 中執(zhí)行mkdir -p agent-mesh/db agent-mesh/logs agent-mesh/scripts cd agent-mesh這樣目錄結(jié)構(gòu)就準(zhǔn)備好了。3. 理解熱節(jié)流手機(jī) SoC 的自我保護(hù)和監(jiān)控方法3.1 Thermal Throttling 的硬件邏輯熱節(jié)流不是一個(gè)單純的操作系統(tǒng)概念它本質(zhì)上是硬件層面的保護(hù)機(jī)制。手機(jī) SoC 內(nèi)集成了多個(gè)溫度傳感器分布在 CPU 集群、GPU、充電管理芯片等位置。當(dāng)某個(gè)傳感器的溫度超過設(shè)定閾值時(shí)SoC 的電源管理單元會通過 DVFS動(dòng)態(tài)電壓頻率調(diào)整降低核心電壓和頻率。所以你在top里看到的 CPU 占用率 100%實(shí)際運(yùn)行速度可能已經(jīng)低到不可用。更麻煩的是熱節(jié)流的觸發(fā)條件不止 CPU 負(fù)載還包括環(huán)境溫度、屏幕亮度、充電狀態(tài)。如果你的手機(jī)一邊充電一邊跑任務(wù)溫度會迅速上升。3.2 在 Termux 中監(jiān)控溫度與頻率在 Termux 中最直接的監(jiān)控方式是讀取系統(tǒng)溫度節(jié)點(diǎn)。Android 系統(tǒng)通常會把溫度傳感器信息暴露在/sys/class/thermal/目錄下。執(zhí)行for zone in /sys/class/thermal/thermal_zone*/; do echo -n $zone: cat $zone/temp done溫度值的單位通常是毫攝氏度也就是 65000 表示 65 攝氏度。為了方便我們可以用 Python 封裝一個(gè)溫度采集函數(shù) 文件路徑: agent-mesh/scripts/thermal_guard.py 作用: 采集電池溫度和 CPU 溫度作為熱節(jié)流保護(hù)模塊的基礎(chǔ)。 import os import subprocess import time def read_cpu_temp(): 讀取第一個(gè)可用的 thermal_zone 溫度單位轉(zhuǎn)換為攝氏度。 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/中的傳感器節(jié)點(diǎn)在不同手機(jī)上有差異有的設(shè)備是thermal_zone0對應(yīng) CPU有的設(shè)備是thermal_zone1對應(yīng)電池。這個(gè)腳本會遍歷前 10 個(gè) zone取第一個(gè)有效值實(shí)際使用時(shí)建議你先手動(dòng)執(zhí)行上面的 shell 命令確定哪個(gè)節(jié)點(diǎn)對應(yīng) CPU。3.3 目標(biāo)把溫度壓在閾值以下經(jīng)驗(yàn)數(shù)據(jù)表明手機(jī) CPU 在 80°C 以下時(shí)通?？梢员３州^高的運(yùn)行頻率超過 85°C 后熱節(jié)流會明顯加強(qiáng)。為了保證 agent mesh 的吞吐穩(wěn)定我們設(shè)定的目標(biāo)溫度建議不高于 75°C。具體的控制策略后文會詳細(xì)展開。4. 10-Tier Agent Mesh 架構(gòu)設(shè)計(jì)4.1 什么是 Agent MeshAgent mesh 指的是一種多節(jié)點(diǎn)協(xié)作模式每個(gè)節(jié)點(diǎn)是一個(gè)輕量級 Agent負(fù)責(zé)從隊(duì)列中取出任務(wù)執(zhí)行特定操作然后把結(jié)果寫入下一個(gè)隊(duì)列。這些 Agent 之間不直接通信而是通過共享數(shù)據(jù)庫協(xié)調(diào)。這樣做的好處是解耦每個(gè) Agent 只知道自己的輸入和輸出不需要了解其他 Agent 的內(nèi)部狀態(tài)。“10-tier”并不是說必須有 10 臺設(shè)備而是把任務(wù)處理流程縱向切分為 10 層每一層代表一種不同的處理階段。本文的例子中各層職責(zé)如下層級職責(zé)示例操作tier_1原始任務(wù)接收寫入原始數(shù)據(jù)tier_2數(shù)據(jù)清洗去空格、去重tier_3字段標(biāo)準(zhǔn)化格式化時(shí)間、枚舉值tier_4關(guān)鍵字提取簡單文本掃描tier_5規(guī)則過濾按規(guī)則丟棄任務(wù)tier_6計(jì)算處理數(shù)值計(jì)算或聚合tier_7結(jié)果分組按目錄字段歸組tier_8匯總統(tǒng)計(jì)生成統(tǒng)計(jì)摘要tier_9格式轉(zhuǎn)換轉(zhuǎn) JSON、CSV 等tier_10輸出歸檔寫入結(jié)果表每一層可以運(yùn)行一個(gè)或多個(gè) Agent層數(shù)越多每個(gè) Agent 的職責(zé)越單一代碼越容易維護(hù)。4.2 數(shù)據(jù)模型10 層隊(duì)列表怎么設(shè)計(jì)在 SQLite 中設(shè)計(jì) 10 層任務(wù)表時(shí)有兩種思路一種是物理建 10 張表另一種是只建一張任務(wù)表通過tier字段區(qū)分。本文推薦第二種因?yàn)槿蝿?wù)在各層之間流轉(zhuǎn)時(shí)本質(zhì)上只是修改tier字段的值不需要跨表復(fù)制性能更好代碼也更簡潔。核心表結(jié)構(gòu)如下-- 文件路徑: 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當(dāng)前任務(wù)所在層級1 到 10。status狀態(tài)pending表示等待處理processing表示正在被某個(gè) Agent 處理done表示完成failed表示失敗。payload任務(wù)數(shù)據(jù)建議存儲 JSON 字符串。attempts嘗試次數(shù)用于防止任務(wù)無限重試。updated_at記錄任務(wù)最近一次狀態(tài)變更時(shí)間是任務(wù)超時(shí)回收的重要依據(jù)。agent_heartbeats表用于記錄每個(gè) Agent 的心跳監(jiān)控 Agent 是否存活。生產(chǎn)環(huán)境中你可以進(jìn)一步細(xì)化表結(jié)構(gòu)比如增加task_type字段支持不同類型的任務(wù)流但本文的簡化設(shè)計(jì)足夠跑通整個(gè)流程。4.3 為什么選擇“數(shù)據(jù)庫輪詢”而不是消息隊(duì)列在電腦上搭 agent mesh你大概率會用 Celery、RQ 或 RabbitMQ。但在 Termux 環(huán)境里這些方案都有各自的痛點(diǎn)Celery 的 Broker 依賴 Redis 或 RabbitMQTermux 里編譯安裝較麻煩。RQ 相對輕量但同樣需要 Redis。直接使用消息隊(duì)列還需要額外的端口監(jiān)聽、持久化保障對手機(jī)來說負(fù)載偏高。數(shù)據(jù)庫輪詢的優(yōu)點(diǎn)是實(shí)現(xiàn)簡單、無額外依賴、進(jìn)程重啟后任務(wù)狀態(tài)不丟。缺點(diǎn)是需要自己控制輪詢間隔避免空轉(zhuǎn)消耗 CPU。本文的 Agent 會采用“IDLE 退避”策略沒有任務(wù)時(shí)逐步拉長輪詢間隔從 1 秒逐步增加到 10 秒。5. 完整實(shí)戰(zhàn)用 SQLite Python 搭建 Agent Mesh5.1 初始化數(shù)據(jù)庫與 10 層任務(wù)表編寫初始化腳本確保數(shù)據(jù)庫和表結(jié)構(gòu)正確創(chuàng)建 文件路徑: agent-mesh/scripts/init_db.py 作用: 初始化數(shù)據(jù)庫表結(jié)構(gòu)并寫入幾條演示任務(wù)。 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] 數(shù)據(jù)庫初始化完成: {DB_PATH}) def seed_tasks(count20): 寫入 count 條演示任務(wù)全部從 tier_1 開始。 conn sqlite3.connect(DB_PATH) tasks [ ( 1, pending, json.dumps({content: f任務(wù) {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} 條演示任務(wù)) if __name__ __main__: init_db() seed_tasks(20)執(zhí)行python scripts/init_db.py如果你本機(jī)沒有安裝sqlite3命令行工具也可以用 Python 的sqlite3模塊直接查看數(shù)據(jù)python -c import sqlite3; print(sqlite3.connect(db/mesh.db).execute(select count(*) from tasks).fetchone())預(yù)期輸出為(20,)表示有 20 條任務(wù)進(jìn)入隊(duì)列。5.2 Agent 工作循環(huán)每個(gè) Agent 都是一個(gè)獨(dú)立的 Python 進(jìn)程它周期性地執(zhí)行以下操作從指定 tier 取出一條 pending 任務(wù)并把狀態(tài)改為 processing。處理任務(wù)更新 payload。把任務(wù)寫入下一層tier 1狀態(tài)改為 pending。如果已經(jīng)是第 10 層則把狀態(tài)改為 done。更新心跳。為了避免多個(gè) Agent 同時(shí)取到同一條任務(wù)這里使用 SQLite 的事務(wù)和條件更新來完成“原子領(lǐng)取”。代碼如下 文件路徑: agent-mesh/scripts/agent_worker.py 作用: 單個(gè) Agent 工作循環(huán)從指定 tier 領(lǐng)取任務(wù)并處理后進(jìn)入下一層。 用法: 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): 原子地領(lǐng)取一條 pending 任務(wù)返回任務(wù)行或 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): 核心處理函數(shù)這里用簡單的時(shí)間戳和字段追加來模擬真實(shí)計(jì)算。 data json.loads(payload) # 模擬計(jì)算耗時(shí)避免 CPU 空轉(zhuǎn) 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): 把任務(wù)寫入下一層或者標(biāo)記為完成。 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 心跳讓運(yùn)維方知道進(jìn)程還活著。 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): 主循環(huán)持續(xù)領(lǐng)取任務(wù)并處理。 idle_sleep 1.0 while True: conn db_connect() try: row claim_task(conn, tier, agent_id) if row is None: # 沒有任務(wù)時(shí)退避降低 CPU 空轉(zhuǎn) 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}] 任務(wù) {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 監(jiān)聽的任務(wù)層級1-10) parser.add_argument(--agent, typestr, requiredTrue, helpAgent 唯一標(biāo)識) args parser.parse_args() run_agent(args.tier, args.agent)代碼中有幾個(gè)關(guān)鍵點(diǎn)需要展開解釋。claim_task函數(shù)使用了UPDATE ... WHERE id (SELECT ... LIMIT 1)的原子更新模式。在 SQLite 中這個(gè)子查詢加更新的組合在同一事務(wù)里是原子性的可以有效避免兩個(gè) Agent 同時(shí)領(lǐng)取同一條任務(wù)。如果你的 SQLite 版本較老不支持RETURNING子句SQLite 3.35 以上才支持可以先用SELECT查出任務(wù) id再通過帶條件的UPDATE把狀態(tài)從pending改為processing然后檢查影響行數(shù)是否為 1# 兼容舊版本 SQLite 的領(lǐng)取方式 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() # 返回任務(wù)行process_payload函數(shù)是 Agent 的真正業(yè)務(wù)邏輯所在。本文用一個(gè) 0.2 秒的 sleep 和數(shù)值乘 2 來模擬真實(shí)計(jì)算你在實(shí)際項(xiàng)目中可以替換成文本解析、網(wǎng)絡(luò)請求、圖像壓縮等操作。write_next_tier函數(shù)把當(dāng)前層處理完的任務(wù)送入下一層。當(dāng)層級增加到第 11 層時(shí)說明整個(gè)流水線已經(jīng)執(zhí)行完畢任務(wù)狀態(tài)被置為done。heartbeat函數(shù)每隔一段時(shí)間把 Agent 的信息寫入心跳表。主循環(huán)中的idle_sleep實(shí)現(xiàn)了退避策略任務(wù)持續(xù)到達(dá)時(shí)保持 1 秒輪詢一次沒有任務(wù)時(shí)逐步拉長到 10 秒。這個(gè)設(shè)計(jì)對控溫非常重要。5.3 啟動(dòng)多節(jié)點(diǎn)并驗(yàn)證分層流轉(zhuǎn)啟動(dòng) 10 個(gè) Agent分別監(jiān)聽 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執(zhí)行后用ps查看進(jìn)程ps aux | grep agent_worker每個(gè) tier 都有一個(gè)對應(yīng)的 Agent 進(jìn)程。然后查看任務(wù)流轉(zhuǎn)情況sqlite3 db/mesh.db select tier, status, count(*) from tasks group by tier, status;預(yù)期輸出大致如下1|pending|0 2|pending|4 3|pending|6 ... 10|pending|2由于各層處理速度不同任務(wù)會逐漸從 tier_1 流向 tier_10。最終所有任務(wù)都應(yīng)該變成10|done|20也就是說20 條任務(wù)全部走完了 10 層流水線。5.4 熱節(jié)流保護(hù)模塊在 5.2 節(jié)的 Agent 循環(huán)里我們其實(shí)還沒有加入真正的溫度控制邏輯。為了長時(shí)間運(yùn)行不觸發(fā)熱節(jié)流需要引入一個(gè)保護(hù)模塊在每次循環(huán)開始前檢查溫度如果超過閾值就主動(dòng)“躺平”一段時(shí)間讓 SoC 降溫。下面是在agent_worker.py中加入溫度控制的改造思路# 在 agent_worker.py 中 import thermal_guard from thermal_guard import read_cpu_temp, read_battery_temp def get_current_temperature(): 返回當(dāng)前系統(tǒng)溫度優(yōu)先使用電池溫度因?yàn)樽钯N近真實(shí)手感。 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 進(jìn)入冷卻休眠。 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然后在主循環(huán)開頭調(diào)用def run_agent_with_thermal_guard(tier, agent_id): idle_sleep 1.0 while True: conn db_connect() try: # 熱節(jié)流保護(hù)溫度過高時(shí)暫停處理 if wait_if_hot(agent_id): continue row claim_task(conn, tier, agent_id) ... finally: conn.close()這個(gè)模塊的工作邏輯是溫度超過 72°C 時(shí)所有 Agent 會進(jìn)入 30 秒的冷卻休眠不領(lǐng)取新任務(wù)讓系統(tǒng)負(fù)載降下來。這個(gè)策略雖然簡單但在手機(jī)這種被動(dòng)散熱設(shè)備上非常有效。更精細(xì)的做法是動(dòng)態(tài)調(diào)整休眠時(shí)長比如溫度越高休眠越長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休眠時(shí)間增加 2 秒。假設(shè)溫度 76°C休眠時(shí)間就是 15 8 23 秒溫度 80°C休眠時(shí)間就是 31 秒。這種梯度休眠策略能讓溫度回落更平緩?fù)瑫r(shí)不至于完全停止所有 Agent。需要注意的是wait_if_hot只是軟件層面的自我保護(hù)。如果你的手機(jī)本身散熱能力很差或者你正在邊充電邊跑任務(wù)軟件的降載效果會很有限。更穩(wěn)妥的做法是結(jié)合硬件策略比如關(guān)閉充電、降低屏幕亮度這些內(nèi)容在最佳實(shí)踐部分會繼續(xù)展開。運(yùn)行熱保護(hù)版 Agentcd agent-mesh python scripts/agent_worker.py --tier 1 --agent agent-hot-1觀察日志當(dāng)溫度接近閾值時(shí)你會看到類似輸出[agent-hot-1] 溫度 73.2°C, 冷卻休眠 20s5.5 驗(yàn)證熱節(jié)流是否被有效避免啟動(dòng)帶熱保護(hù)的 Agent 集群后持續(xù)運(yùn)行一段時(shí)間然后用下面的命令記錄溫度曲線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運(yùn)行 10 分鐘后打開logs/temperature.log如果溫度始終在 72°C 附近波動(dòng)而不是持續(xù)攀升說明熱保護(hù)模塊生效了。如果溫度仍然一路走高到 85°C 以上說明你的業(yè)務(wù)邏輯計(jì)算量太大或者 Agent 數(shù)量過多需要減少并發(fā) Agent 數(shù)量或者降低每層 Agent 的輪詢頻率。6. 常見問題與排查思路在 Termux 中搭建和運(yùn)行 SQLite agent mesh最常遇到的問題有下面這些。問題現(xiàn)象常見原因解決思路數(shù)據(jù)庫顯示database is locked多個(gè)進(jìn)程同時(shí)嘗試寫入事務(wù)沖突設(shè)置timeout10并確保每個(gè)連接及時(shí)關(guān)閉改用 WAL 模式Agent 領(lǐng)取任務(wù)后任務(wù)丟失進(jìn)程被系統(tǒng)殺掉事務(wù)未提交確保領(lǐng)取任務(wù)和更新狀態(tài)在同一個(gè)事務(wù)內(nèi)使用attemps字段做超時(shí)回收溫度迅速飆升到 85°C 以上Agent 數(shù)量太多或輪詢間隔太短減少并發(fā) Agent調(diào)大idle_sleep在溫度過高時(shí)休眠任務(wù)卡在processing不繼續(xù)流轉(zhuǎn)Agent 進(jìn)程崩潰或手機(jī)息屏后進(jìn)程被凍結(jié)添加超時(shí)回收機(jī)制定期把updated_at超過 5 分鐘的處理中任務(wù)重置為pendingtermux-api命令返回空未安裝 Termux:API 應(yīng)用或未授權(quán)需要在 F-Droid 額外安裝 Termux:API APK并在系統(tǒng)設(shè)置中授權(quán)Python 進(jìn)程在后臺運(yùn)行幾分鐘后被殺死Android 后臺進(jìn)程管理策略使用termux-wake-lock保持 CPU 喚醒狀態(tài)針對“任務(wù)卡在 processing”的問題這里給出一個(gè)回收腳本的思路 文件路徑: agent-mesh/scripts/recover_stale.py 作用: 定時(shí)重置超時(shí)未完成的任務(wù)防止任務(wù)永遠(yuǎn)卡在 processing 狀態(tài)。 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} 條超時(shí)任務(wù)) finally: conn.close() time.sleep(60)這個(gè)腳本每 60 秒執(zhí)行一次把updated_at超過 5 分鐘還處于processing狀態(tài)的任務(wù)重新放回待處理隊(duì)列。配合attempts字段你還可以限制每個(gè)任務(wù)的最大重試次數(shù)超過次數(shù)直接標(biāo)記為failed。針對“進(jìn)程被系統(tǒng)殺掉”的問題Termux 提供了喚醒鎖# 在 Termux 中保持 CPU 喚醒狀態(tài) termux-wake-lock # 取消喚醒鎖 termux-wake-unlock運(yùn)行 agent 之前先執(zhí)行termux-wake-lock可以降低息屏后進(jìn)程被凍結(jié)的概率。另外在部分安卓系統(tǒng)上你還需要在系統(tǒng)設(shè)置中把 Termux 的“后臺運(yùn)行”權(quán)限設(shè)為允許否則即使有 wake lock系統(tǒng)仍然可能干預(yù)進(jìn)程。還有一點(diǎn)需要特別注意Termux 是一個(gè)用戶態(tài)環(huán)境/sys/class/thermal/中的溫度節(jié)點(diǎn)在部分設(shè)備上可能沒有讀取權(quán)限。如果你遇到Permission denied不要嘗試 root 或繞過權(quán)限換個(gè)思路改用termux-battery-status的電池溫度作為熱保護(hù)依據(jù)也能達(dá)到差不多的效果。7. 性能優(yōu)化與最佳實(shí)踐7.1 SQLite 寫入性能優(yōu)化SQLite 在默認(rèn)的 delete 模式刷新方式下每次事務(wù)提交都要等待數(shù)據(jù)寫入磁盤性能較差。啟用 WAL 模式后讀操作和寫操作可以并發(fā)執(zhí)行寫入性能提升明顯。初始化腳本已經(jīng)設(shè)置了PRAGMA journal_mode WAL但如果你的庫已經(jīng)創(chuàng)建可以單獨(dú)執(zhí)行sqlite3 db/mesh.db PRAGMA journal_modeWAL;此外建議把synchronous設(shè)置為NORMALsqlite3 db/mesh.db PRAGMA synchronousNORMAL;在 WAL 模式下NORMAL依然可以保證數(shù)據(jù)庫不會損壞只是極端掉電場景下可能丟失最近的事務(wù)。對 agent mesh 這種允許任務(wù)重放的場景來說這個(gè)權(quán)衡是值得的。7.2 Agent 數(shù)量與輪詢間隔的平衡在手機(jī)上跑 10 層 mesh不一定每層都要啟動(dòng)一個(gè)獨(dú)立進(jìn)程。如果你的手機(jī)只有 4 個(gè)性能核心同時(shí)跑 10 個(gè) Python 進(jìn)程會造成頻繁的線程切換溫度很容易失控。比較合理的做法是先啟動(dòng) 6 到 8 個(gè) Agent讓每個(gè) Agent 監(jiān)聽一個(gè) tier跑通流程后再根據(jù)溫度數(shù)據(jù)逐步增加。輪詢間隔的設(shè)置也很關(guān)鍵。對于tier_1這種入口層如果任務(wù)到達(dá)頻率很高可以保持 1 秒輪詢。對于tier_5、tier_6這種中間層如果任務(wù)量不大把輪詢間隔調(diào)到 3 到 5 秒完全夠用。這樣可以顯著降低 CPU 空轉(zhuǎn)率。7.3 手機(jī)硬件層面的控溫建議軟件層面的降載只能解決一部分問題下面幾條硬件層面的建議對實(shí)際控溫非常有效不要邊充電邊跑任務(wù)。充電過程本身會產(chǎn)生大量熱量疊加 CPU 負(fù)載后溫度很容易失控。取下手機(jī)殼或者把手機(jī)放在金屬支架上有利于被動(dòng)散熱。關(guān)閉后臺不相關(guān)的 App特別是短視頻、社交類應(yīng)用它們會周期性喚醒 CPU。如果條件允許使用小風(fēng)扇對著手機(jī)背面吹效果立竿見影。在系統(tǒng)設(shè)置中開啟飛行模式關(guān)閉蜂窩網(wǎng)絡(luò)和 Wi-Fi 熱點(diǎn)避免網(wǎng)絡(luò)模塊持續(xù)工作發(fā)熱。7.4 日志與可觀測性多 Agent 系統(tǒng)排錯(cuò)時(shí)最怕“黑盒”。建議在項(xiàng)目中加入統(tǒng)一的日志格式至少包含以下信息時(shí)間戳。Agent ID。任務(wù) ID。當(dāng)前層級。耗時(shí)。溫度。在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)這樣每個(gè) Agent 的日志都會寫入logs/agent.log。排查問題時(shí)用grep過濾關(guān)鍵字grep 任務(wù) 5 logs/agent.log可以快速定位某個(gè)任務(wù)經(jīng)過了哪些 Agent。7.5 數(shù)據(jù)備份與遷移SQLite 單文件備份非常簡單。如果 Agent 處理的是重要數(shù)據(jù)建議用定時(shí)任務(wù)把db/mesh.db復(fù)制到外部存儲或使用termux-api上傳到自己的服務(wù)器。cp db/mesh.db db/mesh-$(date %Y%m%d-%H%M%S).db如果要在另一臺手機(jī)上恢復(fù)只需把備份文件復(fù)制到對應(yīng)的db/目錄即可。注意備份時(shí)最好先執(zhí)行一次PRAGMA wal_checkpoint;把 WAL 文件中的內(nèi)容合并到主庫文件中。7.6 長跑前的壓測思路正式讓 10 層 mesh 在后臺長時(shí)間運(yùn)行之前建議先做一個(gè)小規(guī)模壓測。具體做法是用init_db.py的seed_tasks函數(shù)寫入 100 條任務(wù)然后在logs/temperature.log中觀察任務(wù)完成時(shí)間和溫度變化。如果 100 條任務(wù)在 10 分鐘內(nèi)完成且溫度控制在 75°C 以下再擴(kuò)展到 500 條、1000 條任務(wù)。要批量寫入更多演示數(shù)據(jù)可以修改seed_tasks(count500)python -c from scripts.init_db import seed_tasks; seed_tasks(500)壓測過程中要重點(diǎn)關(guān)注兩個(gè)指標(biāo)任務(wù)平均流轉(zhuǎn)時(shí)間和系統(tǒng)溫度峰值。如果任務(wù)流轉(zhuǎn)時(shí)間隨著運(yùn)行時(shí)間推移越來越長說明系統(tǒng)已經(jīng)進(jìn)入熱節(jié)流狀態(tài)需要減少 Agent 數(shù)量或調(diào)大休眠閾值。8. 總結(jié)與下一步方向這篇文章完整記錄了一套在 Termux 中搭建 10 層 SQLite agent mesh 的方案。核心思路是用 SQLite 作為多 Agent 之間共享的任務(wù)狀態(tài)存儲用帶退避策略的數(shù)據(jù)庫輪詢替代傳統(tǒng)消息隊(duì)列用溫度感知模塊在熱節(jié)流觸發(fā)前主動(dòng)降載。需要明確的是這個(gè)方案并不是要取代完整的分布式任務(wù)隊(duì)列它更適合的實(shí)驗(yàn)場景是手頭只有一臺舊手機(jī)想驗(yàn)證多層 Agent 協(xié)作的流程設(shè)計(jì)或者想把手上的安卓設(shè)備變成一臺低功耗開發(fā)機(jī)。在 10 層流水線全部跑通之后你可以繼續(xù)嘗試的方向包括在 Agent 中接入實(shí)際的 HTTP API讓不同層的 Agent 調(diào)用不同的遠(yuǎn)程服務(wù)把 payload 中的任務(wù)類型擴(kuò)展為多個(gè)讓同一層的不同 Agent 處理不同任務(wù)使用asyncio重寫 Agent 主循環(huán)在保持單進(jìn)程的前提下處理更多任務(wù)。如果你在實(shí)操中遇到溫度控制效果不明顯的情況先不要急著加更復(fù)雜的算法回到最基礎(chǔ)的三件事減少 Agent 數(shù)量、拉長輪詢間隔、讓手機(jī)物理散熱更好。這個(gè)項(xiàng)目本身就是一場實(shí)驗(yàn)不斷調(diào)整參數(shù)、觀察溫度曲線、記錄任務(wù)流轉(zhuǎn)時(shí)間你會慢慢找到適合自己手機(jī)的最佳配置。如果這篇文章對你有幫助可以把 init 腳本和 agent 工作循環(huán)代碼保存下來直接在 Termux 里跑一遍然后打開溫度日志看看你的手機(jī)能承受多大的并發(fā)量。 Termux 熱節(jié)流自救：用 SQLite 搭建 10 層 Agent Mesh 實(shí)戰(zhàn)
之前在 Termux 里跑并發(fā)任務(wù)時(shí)經(jīng)常遇到一個(gè)奇怪現(xiàn)象任務(wù)一多手機(jī)背面就開始發(fā)燙緊接著 CPU 頻率像被一只手按住所有進(jìn)程一起變慢。這個(gè)現(xiàn)象就是 thermal throttling熱節(jié)流。它不會直接報(bào)錯(cuò)但會讓你的多進(jìn)程系統(tǒng)從“流暢”變成“寸步難行”。本文將圍繞這個(gè)問題記錄一套在 Termux 中通過 SQLite 搭建 10 層 agent mesh 的完整方案包含可運(yùn)行的 Python 代碼、數(shù)據(jù)庫表設(shè)計(jì)、溫度監(jiān)控方法和進(jìn)程調(diào)度策略。無論你是想在舊手機(jī)上搭建一個(gè)實(shí)驗(yàn)性的多 Agent 任務(wù)系統(tǒng)還是準(zhǔn)備把 Termux 當(dāng)作低功耗開發(fā)環(huán)境這篇文章都值得收藏。1. 背景為什么要在手機(jī)上跑 10 層 Agent Mesh1.1 這個(gè)實(shí)驗(yàn)解決什么問題日常開發(fā)中當(dāng)我們提到“多 Agent 協(xié)作”第一反應(yīng)往往是 K8s、Docker Compose、Redis 消息隊(duì)列這些重型基礎(chǔ)設(shè)施。但如果你手里的設(shè)備只是一臺安卓手機(jī)沒有 root 權(quán)限也沒有完整的 Linux 內(nèi)核管理能力這些方案基本都跑不起來。Termux 提供了一個(gè)輕量級的 Linux 環(huán)境但它對進(jìn)程管理、系統(tǒng)服務(wù)、網(wǎng)絡(luò)端口綁定都有一定的限制。本文要探討的是如何不依賴復(fù)雜基礎(chǔ)設(shè)施僅用 Termux SQLite Python 就搭建出一個(gè) 10 層任務(wù)處理網(wǎng)格并且保證長時(shí)間運(yùn)行不觸發(fā)熱節(jié)流。一個(gè)典型的 agent mesh 場景是這樣的系統(tǒng)里有多個(gè) worker 節(jié)點(diǎn)每個(gè)節(jié)點(diǎn)處理特定類型的任務(wù)。任務(wù)從第一層進(jìn)入經(jīng)過抽取、清洗、轉(zhuǎn)換、計(jì)算、聚合等多個(gè)階段最終輸出結(jié)果。每一層都可能有多個(gè) agent 并發(fā)處理任務(wù)層與層之間通過共享的存儲介質(zhì)傳遞數(shù)據(jù)。在傳統(tǒng)架構(gòu)中這個(gè)存儲介質(zhì)可能是 Redis、RabbitMQ 或 Kafka但在 Termux 這種輕量環(huán)境里SQLite 反而成了最合適的選擇。1.2 SQLite 為什么適合做 Agent 總線很多人對 SQLite 的印象還停留在“單機(jī)小數(shù)據(jù)庫”覺得它撐不起并發(fā)場景。實(shí)際上SQLite 在 WALWrite-Ahead Logging模式下多進(jìn)程并發(fā)讀、單進(jìn)程寫的能力足以支撐中小規(guī)模的 Agent 任務(wù)調(diào)度。關(guān)鍵在于你怎么設(shè)計(jì)表結(jié)構(gòu)和事務(wù)粒度。SQLite 相比消息隊(duì)列有幾個(gè)不可替代的優(yōu)勢零配置Termux 里直接安裝就能用。數(shù)據(jù)庫即文件備份、遷移、查看都非常方便。支持事務(wù)、索引、唯一約束可以避免任務(wù)被重復(fù)消費(fèi)。一個(gè)文件就能保存全部任務(wù)狀態(tài)排查問題時(shí)用 db browser 打開就能看。缺點(diǎn)當(dāng)然也有寫入并發(fā)能力有限單條記錄大小不宜過大。但對于 agent mesh 這種以任務(wù)狀態(tài)流轉(zhuǎn)為主、消息體通常只有幾 KB 的場景SQLite 的瓶頸完全夠用。1.3 熱節(jié)流是最大的敵人在手機(jī)上跑任務(wù)最怕的不是 CPU 不夠快而是“發(fā)揮不出來”。手機(jī) SoC 有嚴(yán)格的溫度控制策略當(dāng)芯片溫度超過閾值時(shí)系統(tǒng)會主動(dòng)降低 CPU 頻率、限制核心調(diào)度甚至關(guān)閉高性能核心。這個(gè)機(jī)制在 x86 設(shè)備上通常叫 PROCHOTProcessor Hot信號在 ARM 設(shè)備上則是通過 thermal_zone 溫度節(jié)點(diǎn)與 governor 策略共同實(shí)現(xiàn)。熱節(jié)流帶來的問題非常隱蔽你的進(jìn)程并沒有崩潰但運(yùn)行速度可能會降到原來的十分之一。如果 agent 之間有超時(shí)機(jī)制還會引發(fā)連鎖的任務(wù)堆積和重復(fù)處理。所以在 Termux 中搭建多 Agent 系統(tǒng)第一優(yōu)先級不是“跑滿 CPU”而是“控制溫度”。2. 環(huán)境準(zhǔn)備Termux 里的最小開發(fā)棧2.1 安裝 Termux 并更新軟件源Termux 的安裝渠道比較特殊Google Play 上的版本已經(jīng)停止維護(hù)推薦從 F-Droid 或 GitHub Releases 下載最新 APK。安裝完成后首先執(zhí)行pkg update pkg upgrade -y這一步會更新軟件源和已安裝包。需要注意如果你的 Termux 是剛安裝的pkg命令可能需要先初始化存儲目錄執(zhí)行termux-setup-storage授權(quán)存儲權(quán)限。2.2 安裝 SQLite、Python 與監(jiān)控工具本文的實(shí)戰(zhàn)部分以 Python 為例需要安裝以下包pkg install python sqlite clang -y pkg install termux-api -y這里說明一下每個(gè)包的用途python用來編寫 Agent 工作循環(huán)。sqlite提供 SQLite 命令行工具方便調(diào)試和查詢。clangPython 的某些依賴需要編譯提前安裝可以避免報(bào)錯(cuò)。termux-api用來讀取電池溫度、充電狀態(tài)等系統(tǒng)信息是熱節(jié)流保護(hù)模塊的基礎(chǔ)。如果你希望 Agent 輸出更規(guī)范的日志還可以安裝pandas、pydantic這類包但本文為了保持依賴最小化只用 Python 標(biāo)準(zhǔn)庫。pip install --upgrade pip國內(nèi)網(wǎng)絡(luò)環(huán)境下pip 源如果不穩(wěn)定可以臨時(shí)使用清華鏡像pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple版本方面需要提醒一下Termux 的軟件源是滾動(dòng)更新的不同時(shí)間安裝的 Python 和 SQLite 版本可能不同。本文的代碼只使用 SQLite 的基礎(chǔ)語法和 Python 標(biāo)準(zhǔn)庫只要你的 SQLite 版本在 3.8 以上支持 WAL 模式基本不存在兼容性問題。2.3 規(guī)劃項(xiàng)目目錄建議按下面的結(jié)構(gòu)組織項(xiàng)目agent-mesh/ ├── db/ │ └── mesh.db ├── logs/ │ └── agent.log ├── scripts/ │ ├── init_db.py │ ├── agent_worker.py │ ├── thermal_guard.py │ └── dispatch.py └── README.md在 Termux 中執(zhí)行mkdir -p agent-mesh/db agent-mesh/logs agent-mesh/scripts cd agent-mesh這樣目錄結(jié)構(gòu)就準(zhǔn)備好了。3. 理解熱節(jié)流手機(jī) SoC 的自我保護(hù)和監(jiān)控方法3.1 Thermal Throttling 的硬件邏輯熱節(jié)流不是一個(gè)單純的操作系統(tǒng)概念它本質(zhì)上是硬件層面的保護(hù)機(jī)制。手機(jī) SoC 內(nèi)集成了多個(gè)溫度傳感器分布在 CPU 集群、GPU、充電管理芯片等位置。當(dāng)某個(gè)傳感器的溫度超過設(shè)定閾值時(shí)SoC 的電源管理單元會通過 DVFS動(dòng)態(tài)電壓頻率調(diào)整降低核心電壓和頻率。所以你在top里看到的 CPU 占用率 100%實(shí)際運(yùn)行速度可能已經(jīng)低到不可用。更麻煩的是熱節(jié)流的觸發(fā)條件不止 CPU 負(fù)載還包括環(huán)境溫度、屏幕亮度、充電狀態(tài)。如果你的手機(jī)一邊充電一邊跑任務(wù)溫度會迅速上升。3.2 在 Termux 中監(jiān)控溫度與頻率在 Termux 中最直接的監(jiān)控方式是讀取系統(tǒng)溫度節(jié)點(diǎn)。Android 系統(tǒng)通常會把溫度傳感器信息暴露在/sys/class/thermal/目錄下。執(zhí)行for zone in /sys/class/thermal/thermal_zone*/; do echo -n $zone: cat $zone/temp done溫度值的單位通常是毫攝氏度也就是 65000 表示 65 攝氏度。為了方便我們可以用 Python 封裝一個(gè)溫度采集函數(shù) 文件路徑: agent-mesh/scripts/thermal_guard.py 作用: 采集電池溫度和 CPU 溫度作為熱節(jié)流保護(hù)模塊的基礎(chǔ)。 import os import subprocess import time def read_cpu_temp(): 讀取第一個(gè)可用的 thermal_zone 溫度單位轉(zhuǎn)換為攝氏度。 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/中的傳感器節(jié)點(diǎn)在不同手機(jī)上有差異有的設(shè)備是thermal_zone0對應(yīng) CPU有的設(shè)備是thermal_zone1對應(yīng)電池。這個(gè)腳本會遍歷前 10 個(gè) zone取第一個(gè)有效值實(shí)際使用時(shí)建議你先手動(dòng)執(zhí)行上面的 shell 命令確定哪個(gè)節(jié)點(diǎn)對應(yīng) CPU。3.3 目標(biāo)把溫度壓在閾值以下經(jīng)驗(yàn)數(shù)據(jù)表明手機(jī) CPU 在 80°C 以下時(shí)通?？梢员３州^高的運(yùn)行頻率超過 85°C 后熱節(jié)流會明顯加強(qiáng)。為了保證 agent mesh 的吞吐穩(wěn)定我們設(shè)定的目標(biāo)溫度建議不高于 75°C。具體的控制策略后文會詳細(xì)展開。4. 10-Tier Agent Mesh 架構(gòu)設(shè)計(jì)4.1 什么是 Agent MeshAgent mesh 指的是一種多節(jié)點(diǎn)協(xié)作模式每個(gè)節(jié)點(diǎn)是一個(gè)輕量級 Agent負(fù)責(zé)從隊(duì)列中取出任務(wù)執(zhí)行特定操作然后把結(jié)果寫入下一個(gè)隊(duì)列。這些 Agent 之間不直接通信而是通過共享數(shù)據(jù)庫協(xié)調(diào)。這樣做的好處是解耦每個(gè) Agent 只知道自己的輸入和輸出不需要了解其他 Agent 的內(nèi)部狀態(tài)。“10-tier”并不是說必須有 10 臺設(shè)備而是把任務(wù)處理流程縱向切分為 10 層每一層代表一種不同的處理階段。本文的例子中各層職責(zé)如下層級職責(zé)示例操作tier_1原始任務(wù)接收寫入原始數(shù)據(jù)tier_2數(shù)據(jù)清洗去空格、去重tier_3字段標(biāo)準(zhǔn)化格式化時(shí)間、枚舉值tier_4關(guān)鍵字提取簡單文本掃描tier_5規(guī)則過濾按規(guī)則丟棄任務(wù)tier_6計(jì)算處理數(shù)值計(jì)算或聚合tier_7結(jié)果分組按目錄字段歸組tier_8匯總統(tǒng)計(jì)生成統(tǒng)計(jì)摘要tier_9格式轉(zhuǎn)換轉(zhuǎn) JSON、CSV 等tier_10輸出歸檔寫入結(jié)果表每一層可以運(yùn)行一個(gè)或多個(gè) Agent層數(shù)越多每個(gè) Agent 的職責(zé)越單一代碼越容易維護(hù)。4.2 數(shù)據(jù)模型10 層隊(duì)列表怎么設(shè)計(jì)在 SQLite 中設(shè)計(jì) 10 層任務(wù)表時(shí)有兩種思路一種是物理建 10 張表另一種是只建一張任務(wù)表通過tier字段區(qū)分。本文推薦第二種因?yàn)槿蝿?wù)在各層之間流轉(zhuǎn)時(shí)本質(zhì)上只是修改tier字段的值不需要跨表復(fù)制性能更好代碼也更簡潔。核心表結(jié)構(gòu)如下-- 文件路徑: 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當(dāng)前任務(wù)所在層級1 到 10。status狀態(tài)pending表示等待處理processing表示正在被某個(gè) Agent 處理done表示完成failed表示失敗。payload任務(wù)數(shù)據(jù)建議存儲 JSON 字符串。attempts嘗試次數(shù)用于防止任務(wù)無限重試。updated_at記錄任務(wù)最近一次狀態(tài)變更時(shí)間是任務(wù)超時(shí)回收的重要依據(jù)。agent_heartbeats表用于記錄每個(gè) Agent 的心跳監(jiān)控 Agent 是否存活。生產(chǎn)環(huán)境中你可以進(jìn)一步細(xì)化表結(jié)構(gòu)比如增加task_type字段支持不同類型的任務(wù)流但本文的簡化設(shè)計(jì)足夠跑通整個(gè)流程。4.3 為什么選擇“數(shù)據(jù)庫輪詢”而不是消息隊(duì)列在電腦上搭 agent mesh你大概率會用 Celery、RQ 或 RabbitMQ。但在 Termux 環(huán)境里這些方案都有各自的痛點(diǎn)Celery 的 Broker 依賴 Redis 或 RabbitMQTermux 里編譯安裝較麻煩。RQ 相對輕量但同樣需要 Redis。直接使用消息隊(duì)列還需要額外的端口監(jiān)聽、持久化保障對手機(jī)來說負(fù)載偏高。數(shù)據(jù)庫輪詢的優(yōu)點(diǎn)是實(shí)現(xiàn)簡單、無額外依賴、進(jìn)程重啟后任務(wù)狀態(tài)不丟。缺點(diǎn)是需要自己控制輪詢間隔避免空轉(zhuǎn)消耗 CPU。本文的 Agent 會采用“IDLE 退避”策略沒有任務(wù)時(shí)逐步拉長輪詢間隔從 1 秒逐步增加到 10 秒。5. 完整實(shí)戰(zhàn)用 SQLite Python 搭建 Agent Mesh5.1 初始化數(shù)據(jù)庫與 10 層任務(wù)表編寫初始化腳本確保數(shù)據(jù)庫和表結(jié)構(gòu)正確創(chuàng)建 文件路徑: agent-mesh/scripts/init_db.py 作用: 初始化數(shù)據(jù)庫表結(jié)構(gòu)并寫入幾條演示任務(wù)。 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] 數(shù)據(jù)庫初始化完成: {DB_PATH}) def seed_tasks(count20): 寫入 count 條演示任務(wù)全部從 tier_1 開始。 conn sqlite3.connect(DB_PATH) tasks [ ( 1, pending, json.dumps({content: f任務(wù) {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} 條演示任務(wù)) if __name__ __main__: init_db() seed_tasks(20)執(zhí)行python scripts/init_db.py如果你本機(jī)沒有安裝sqlite3命令行工具也可以用 Python 的sqlite3模塊直接查看數(shù)據(jù)python -c import sqlite3; print(sqlite3.connect(db/mesh.db).execute(select count(*) from tasks).fetchone())預(yù)期輸出為(20,)表示有 20 條任務(wù)進(jìn)入隊(duì)列。5.2 Agent 工作循環(huán)每個(gè) Agent 都是一個(gè)獨(dú)立的 Python 進(jìn)程它周期性地執(zhí)行以下操作從指定 tier 取出一條 pending 任務(wù)并把狀態(tài)改為 processing。處理任務(wù)更新 payload。把任務(wù)寫入下一層tier 1狀態(tài)改為 pending。如果已經(jīng)是第 10 層則把狀態(tài)改為 done。更新心跳。為了避免多個(gè) Agent 同時(shí)取到同一條任務(wù)這里使用 SQLite 的事務(wù)和條件更新來完成“原子領(lǐng)取”。代碼如下 文件路徑: agent-mesh/scripts/agent_worker.py 作用: 單個(gè) Agent 工作循環(huán)從指定 tier 領(lǐng)取任務(wù)并處理后進(jìn)入下一層。 用法: 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): 原子地領(lǐng)取一條 pending 任務(wù)返回任務(wù)行或 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): 核心處理函數(shù)這里用簡單的時(shí)間戳和字段追加來模擬真實(shí)計(jì)算。 data json.loads(payload) # 模擬計(jì)算耗時(shí)避免 CPU 空轉(zhuǎn) 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): 把任務(wù)寫入下一層或者標(biāo)記為完成。 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 心跳讓運(yùn)維方知道進(jìn)程還活著。 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): 主循環(huán)持續(xù)領(lǐng)取任務(wù)并處理。 idle_sleep 1.0 while True: conn db_connect() try: row claim_task(conn, tier, agent_id) if row is None: # 沒有任務(wù)時(shí)退避降低 CPU 空轉(zhuǎn) 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}] 任務(wù) {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 監(jiān)聽的任務(wù)層級1-10) parser.add_argument(--agent, typestr, requiredTrue, helpAgent 唯一標(biāo)識) args parser.parse_args() run_agent(args.tier, args.agent)代碼中有幾個(gè)關(guān)鍵點(diǎn)需要展開解釋。claim_task函數(shù)使用了UPDATE ... WHERE id (SELECT ... LIMIT 1)的原子更新模式。在 SQLite 中這個(gè)子查詢加更新的組合在同一事務(wù)里是原子性的可以有效避免兩個(gè) Agent 同時(shí)領(lǐng)取同一條任務(wù)。如果你的 SQLite 版本較老不支持RETURNING子句SQLite 3.35 以上才支持可以先用SELECT查出任務(wù) id再通過帶條件的UPDATE把狀態(tài)從pending改為processing然后檢查影響行數(shù)是否為 1# 兼容舊版本 SQLite 的領(lǐng)取方式 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() # 返回任務(wù)行process_payload函數(shù)是 Agent 的真正業(yè)務(wù)邏輯所在。本文用一個(gè) 0.2 秒的 sleep 和數(shù)值乘 2 來模擬真實(shí)計(jì)算你在實(shí)際項(xiàng)目中可以替換成文本解析、網(wǎng)絡(luò)請求、圖像壓縮等操作。write_next_tier函數(shù)把當(dāng)前層處理完的任務(wù)送入下一層。當(dāng)層級增加到第 11 層時(shí)說明整個(gè)流水線已經(jīng)執(zhí)行完畢任務(wù)狀態(tài)被置為done。heartbeat函數(shù)每隔一段時(shí)間把 Agent 的信息寫入心跳表。主循環(huán)中的idle_sleep實(shí)現(xiàn)了退避策略任務(wù)持續(xù)到達(dá)時(shí)保持 1 秒輪詢一次沒有任務(wù)時(shí)逐步拉長到 10 秒。這個(gè)設(shè)計(jì)對控溫非常重要。5.3 啟動(dòng)多節(jié)點(diǎn)并驗(yàn)證分層流轉(zhuǎn)啟動(dòng) 10 個(gè) Agent分別監(jiān)聽 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執(zhí)行后用ps查看進(jìn)程ps aux | grep agent_worker每個(gè) tier 都有一個(gè)對應(yīng)的 Agent 進(jìn)程。然后查看任務(wù)流轉(zhuǎn)情況sqlite3 db/mesh.db select tier, status, count(*) from tasks group by tier, status;預(yù)期輸出大致如下1|pending|0 2|pending|4 3|pending|6 ... 10|pending|2由于各層處理速度不同任務(wù)會逐漸從 tier_1 流向 tier_10。最終所有任務(wù)都應(yīng)該變成10|done|20也就是說20 條任務(wù)全部走完了 10 層流水線。5.4 熱節(jié)流保護(hù)模塊在 5.2 節(jié)的 Agent 循環(huán)里我們其實(shí)還沒有加入真正的溫度控制邏輯。為了長時(shí)間運(yùn)行不觸發(fā)熱節(jié)流需要引入一個(gè)保護(hù)模塊在每次循環(huán)開始前檢查溫度如果超過閾值就主動(dòng)“躺平”一段時(shí)間讓 SoC 降溫。下面是在agent_worker.py中加入溫度控制的改造思路# 在 agent_worker.py 中 import thermal_guard from thermal_guard import read_cpu_temp, read_battery_temp def get_current_temperature(): 返回當(dāng)前系統(tǒng)溫度優(yōu)先使用電池溫度因?yàn)樽钯N近真實(shí)手感。 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 進(jìn)入冷卻休眠。 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然后在主循環(huán)開頭調(diào)用def run_agent_with_thermal_guard(tier, agent_id): idle_sleep 1.0 while True: conn db_connect() try: # 熱節(jié)流保護(hù)溫度過高時(shí)暫停處理 if wait_if_hot(agent_id): continue row claim_task(conn, tier, agent_id) ... finally: conn.close()這個(gè)模塊的工作邏輯是溫度超過 72°C 時(shí)所有 Agent 會進(jìn)入 30 秒的冷卻休眠不領(lǐng)取新任務(wù)讓系統(tǒng)負(fù)載降下來。這個(gè)策略雖然簡單但在手機(jī)這種被動(dòng)散熱設(shè)備上非常有效。更精細(xì)的做法是動(dòng)態(tài)調(diào)整休眠時(shí)長比如溫度越高休眠越長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休眠時(shí)間增加 2 秒。假設(shè)溫度 76°C休眠時(shí)間就是 15 8 23 秒溫度 80°C休眠時(shí)間就是 31 秒。這種梯度休眠策略能讓溫度回落更平緩?fù)瑫r(shí)不至于完全停止所有 Agent。需要注意的是wait_if_hot只是軟件層面的自我保護(hù)。如果你的手機(jī)本身散熱能力很差或者你正在邊充電邊跑任務(wù)軟件的降載效果會很有限。更穩(wěn)妥的做法是結(jié)合硬件策略比如關(guān)閉充電、降低屏幕亮度這些內(nèi)容在最佳實(shí)踐部分會繼續(xù)展開。運(yùn)行熱保護(hù)版 Agentcd agent-mesh python scripts/agent_worker.py --tier 1 --agent agent-hot-1觀察日志當(dāng)溫度接近閾值時(shí)你會看到類似輸出[agent-hot-1] 溫度 73.2°C, 冷卻休眠 20s5.5 驗(yàn)證熱節(jié)流是否被有效避免啟動(dòng)帶熱保護(hù)的 Agent 集群后持續(xù)運(yùn)行一段時(shí)間然后用下面的命令記錄溫度曲線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運(yùn)行 10 分鐘后打開logs/temperature.log如果溫度始終在 72°C 附近波動(dòng)而不是持續(xù)攀升說明熱保護(hù)模塊生效了。如果溫度仍然一路走高到 85°C 以上說明你的業(yè)務(wù)邏輯計(jì)算量太大或者 Agent 數(shù)量過多需要減少并發(fā) Agent 數(shù)量或者降低每層 Agent 的輪詢頻率。6. 常見問題與排查思路在 Termux 中搭建和運(yùn)行 SQLite agent mesh最常遇到的問題有下面這些。問題現(xiàn)象常見原因解決思路數(shù)據(jù)庫顯示database is locked多個(gè)進(jìn)程同時(shí)嘗試寫入事務(wù)沖突設(shè)置timeout10并確保每個(gè)連接及時(shí)關(guān)閉改用 WAL 模式Agent 領(lǐng)取任務(wù)后任務(wù)丟失進(jìn)程被系統(tǒng)殺掉事務(wù)未提交確保領(lǐng)取任務(wù)和更新狀態(tài)在同一個(gè)事務(wù)內(nèi)使用attemps字段做超時(shí)回收溫度迅速飆升到 85°C 以上Agent 數(shù)量太多或輪詢間隔太短減少并發(fā) Agent調(diào)大idle_sleep在溫度過高時(shí)休眠任務(wù)卡在processing不繼續(xù)流轉(zhuǎn)Agent 進(jìn)程崩潰或手機(jī)息屏后進(jìn)程被凍結(jié)添加超時(shí)回收機(jī)制定期把updated_at超過 5 分鐘的處理中任務(wù)重置為pendingtermux-api命令返回空未安裝 Termux:API 應(yīng)用或未授權(quán)需要在 F-Droid 額外安裝 Termux:API APK并在系統(tǒng)設(shè)置中授權(quán)Python 進(jìn)程在后臺運(yùn)行幾分鐘后被殺死Android 后臺進(jìn)程管理策略使用termux-wake-lock保持 CPU 喚醒狀態(tài)針對“任務(wù)卡在 processing”的問題這里給出一個(gè)回收腳本的思路 文件路徑: agent-mesh/scripts/recover_stale.py 作用: 定時(shí)重置超時(shí)未完成的任務(wù)防止任務(wù)永遠(yuǎn)卡在 processing 狀態(tài)。 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} 條超時(shí)任務(wù)) finally: conn.close() time.sleep(60)這個(gè)腳本每 60 秒執(zhí)行一次把updated_at超過 5 分鐘還處于processing狀態(tài)的任務(wù)重新放回待處理隊(duì)列。配合attempts字段你還可以限制每個(gè)任務(wù)的最大重試次數(shù)超過次數(shù)直接標(biāo)記為failed。針對“進(jìn)程被系統(tǒng)殺掉”的問題Termux 提供了喚醒鎖# 在 Termux 中保持 CPU 喚醒狀態(tài) termux-wake-lock # 取消喚醒鎖 termux-wake-unlock運(yùn)行 agent 之前先執(zhí)行termux-wake-lock可以降低息屏后進(jìn)程被凍結(jié)的概率。另外在部分安卓系統(tǒng)上你還需要在系統(tǒng)設(shè)置中把 Termux 的“后臺運(yùn)行”權(quán)限設(shè)為允許否則即使有 wake lock系統(tǒng)仍然可能干預(yù)進(jìn)程。還有一點(diǎn)需要特別注意Termux 是一個(gè)用戶態(tài)環(huán)境/sys/class/thermal/中的溫度節(jié)點(diǎn)在部分設(shè)備上可能沒有讀取權(quán)限。如果你遇到Permission denied不要嘗試 root 或繞過權(quán)限換個(gè)思路改用termux-battery-status的電池溫度作為熱保護(hù)依據(jù)也能達(dá)到差不多的效果。7. 性能優(yōu)化與最佳實(shí)踐7.1 SQLite 寫入性能優(yōu)化SQLite 在默認(rèn)的 delete 模式刷新方式下每次事務(wù)提交都要等待數(shù)據(jù)寫入磁盤性能較差。啟用 WAL 模式后讀操作和寫操作可以并發(fā)執(zhí)行寫入性能提升明顯。初始化腳本已經(jīng)設(shè)置了PRAGMA journal_mode WAL但如果你的庫已經(jīng)創(chuàng)建可以單獨(dú)執(zhí)行sqlite3 db/mesh.db PRAGMA journal_modeWAL;此外建議把synchronous設(shè)置為NORMALsqlite3 db/mesh.db PRAGMA synchronousNORMAL;在 WAL 模式下NORMAL依然可以保證數(shù)據(jù)庫不會損壞只是極端掉電場景下可能丟失最近的事務(wù)。對 agent mesh 這種允許任務(wù)重放的場景來說這個(gè)權(quán)衡是值得的。7.2 Agent 數(shù)量與輪詢間隔的平衡在手機(jī)上跑 10 層 mesh不一定每層都要啟動(dòng)一個(gè)獨(dú)立進(jìn)程。如果你的手機(jī)只有 4 個(gè)性能核心同時(shí)跑 10 個(gè) Python 進(jìn)程會造成頻繁的線程切換溫度很容易失控。比較合理的做法是先啟動(dòng) 6 到 8 個(gè) Agent讓每個(gè) Agent 監(jiān)聽一個(gè) tier跑通流程后再根據(jù)溫度數(shù)據(jù)逐步增加。輪詢間隔的設(shè)置也很關(guān)鍵。對于tier_1這種入口層如果任務(wù)到達(dá)頻率很高可以保持 1 秒輪詢。對于tier_5、tier_6這種中間層如果任務(wù)量不大把輪詢間隔調(diào)到 3 到 5 秒完全夠用。這樣可以顯著降低 CPU 空轉(zhuǎn)率。7.3 手機(jī)硬件層面的控溫建議軟件層面的降載只能解決一部分問題下面幾條硬件層面的建議對實(shí)際控溫非常有效不要邊充電邊跑任務(wù)。充電過程本身會產(chǎn)生大量熱量疊加 CPU 負(fù)載后溫度很容易失控。取下手機(jī)殼或者把手機(jī)放在金屬支架上有利于被動(dòng)散熱。關(guān)閉后臺不相關(guān)的 App特別是短視頻、社交類應(yīng)用它們會周期性喚醒 CPU。如果條件允許使用小風(fēng)扇對著手機(jī)背面吹效果立竿見影。在系統(tǒng)設(shè)置中開啟飛行模式關(guān)閉蜂窩網(wǎng)絡(luò)和 Wi-Fi 熱點(diǎn)避免網(wǎng)絡(luò)模塊持續(xù)工作發(fā)熱。7.4 日志與可觀測性多 Agent 系統(tǒng)排錯(cuò)時(shí)最怕“黑盒”。建議在項(xiàng)目中加入統(tǒng)一的日志格式至少包含以下信息時(shí)間戳。Agent ID。任務(wù) ID。當(dāng)前層級。耗時(shí)。溫度。在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)這樣每個(gè) Agent 的日志都會寫入logs/agent.log。排查問題時(shí)用grep過濾關(guān)鍵字grep 任務(wù) 5 logs/agent.log可以快速定位某個(gè)任務(wù)經(jīng)過了哪些 Agent。7.5 數(shù)據(jù)備份與遷移SQLite 單文件備份非常簡單。如果 Agent 處理的是重要數(shù)據(jù)建議用定時(shí)任務(wù)把db/mesh.db復(fù)制到外部存儲或使用termux-api上傳到自己的服務(wù)器。cp db/mesh.db db/mesh-$(date %Y%m%d-%H%M%S).db如果要在另一臺手機(jī)上恢復(fù)只需把備份文件復(fù)制到對應(yīng)的db/目錄即可。注意備份時(shí)最好先執(zhí)行一次PRAGMA wal_checkpoint;把 WAL 文件中的內(nèi)容合并到主庫文件中。7.6 長跑前的壓測思路正式讓 10 層 mesh 在后臺長時(shí)間運(yùn)行之前建議先做一個(gè)小規(guī)模壓測。具體做法是用init_db.py的seed_tasks函數(shù)寫入 100 條任務(wù)然后在logs/temperature.log中觀察任務(wù)完成時(shí)間和溫度變化。如果 100 條任務(wù)在 10 分鐘內(nèi)完成且溫度控制在 75°C 以下再擴(kuò)展到 500 條、1000 條任務(wù)。要批量寫入更多演示數(shù)據(jù)可以修改seed_tasks(count500)python -c from scripts.init_db import seed_tasks; seed_tasks(500)壓測過程中要重點(diǎn)關(guān)注兩個(gè)指標(biāo)任務(wù)平均流轉(zhuǎn)時(shí)間和系統(tǒng)溫度峰值。如果任務(wù)流轉(zhuǎn)時(shí)間隨著運(yùn)行時(shí)間推移越來越長說明系統(tǒng)已經(jīng)進(jìn)入熱節(jié)流狀態(tài)需要減少 Agent 數(shù)量或調(diào)大休眠閾值。8. 總結(jié)與下一步方向這篇文章完整記錄了一套在 Termux 中搭建 10 層 SQLite agent mesh 的方案。核心思路是用 SQLite 作為多 Agent 之間共享的任務(wù)狀態(tài)存儲用帶退避策略的數(shù)據(jù)庫輪詢替代傳統(tǒng)消息隊(duì)列用溫度感知模塊在熱節(jié)流觸發(fā)前主動(dòng)降載。需要明確的是這個(gè)方案并不是要取代完整的分布式任務(wù)隊(duì)列它更適合的實(shí)驗(yàn)場景是手頭只有一臺舊手機(jī)想驗(yàn)證多層 Agent 協(xié)作的流程設(shè)計(jì)或者想把手上的安卓設(shè)備變成一臺低功耗開發(fā)機(jī)。在 10 層流水線全部跑通之后你可以繼續(xù)嘗試的方向包括在 Agent 中接入實(shí)際的 HTTP API讓不同層的 Agent 調(diào)用不同的遠(yuǎn)程服務(wù)把 payload 中的任務(wù)類型擴(kuò)展為多個(gè)讓同一層的不同 Agent 處理不同任務(wù)使用asyncio重寫 Agent 主循環(huán)在保持單進(jìn)程的前提下處理更多任務(wù)。如果你在實(shí)操中遇到溫度控制效果不明顯的情況先不要急著加更復(fù)雜的算法回到最基礎(chǔ)的三件事減少 Agent 數(shù)量、拉長輪詢間隔、讓手機(jī)物理散熱更好。這個(gè)項(xiàng)目本身就是一場實(shí)驗(yàn)不斷調(diào)整參數(shù)、觀察溫度曲線、記錄任務(wù)流轉(zhuǎn)時(shí)間你會慢慢找到適合自己手機(jī)的最佳配置。如果這篇文章對你有幫助可以把 init 腳本和 agent 工作循環(huán)代碼保存下來直接在 Termux 里跑一遍然后打開溫度日志看看你的手機(jī)能承受多大的并發(fā)量。