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