这个周末给自己的 AI 助手做了次”体检”，发现虽然功能都能跑，但离真正的生产环境还差几步关键动作。花了两小时补上了数据备份、健康监控、配置校验和 SQLite 并发优化，现在可以放心让它 7×24 跑了。
背景
Mini-Agent 是我的个人 AI 助手，基于 EventBus 事件驱动架构，集成了 Telegram Bot、Web API、定时任务等功能。本地实例负责调度和用户交互，云端实例（ECS）提供公网访问。两个实例通过 Git 同步博客内容，通过 HTTP 双向同步待办事项和日程。
核心组件：
- EventBus - 事件总线，所有模块间通信走这里
- Pi RPC - 调用 pi CLI 执行 LLM 任务（带 bubblewrap 沙箱隔离）
- ConversationManager - LangGraph + SQLite 管理对话状态
- Telegram/Web 双通道 - 本地用 Telegram，外网用 Web API
- 多租户认证 - UserManager + argon2id 哈希 + 按天配额
技术栈：Python 3.11 + asyncio + aiosqlite + APScheduler + python-telegram-bot + aiohttp + LangChain。
发现的问题
1. 数据丢失风险
所有数据（对话历史、待办、用户信息）都在本地 SQLite，没有任何备份机制。一次磁盘故障或误操作 rm -rf 就全没了。
2. 故障无预警
进程挂了、磁盘满了、内存泄漏了，都得等用户发现才知道。没有主动监控，只能靠 Telegram 发消息测试能不能收到回复。
3. 配置错误运行时才炸
改完配置重启服务，启动成功，结果第一次调用 pi 才发现路径写错了。错误信息淹没在一堆日志里，排查要翻半天。
4. SQLite 写锁瓶颈
默认的 DELETE 模式下，写操作会锁整个数据库。虽然现在用户不多，但多租户模式下并发一上来就卡。
解决方案
Phase 0：消除生产风险（已完成）
1. 数据自动备份
写了个 BackupActor，每天凌晨 3 点自动把 data/ 目录打包，通过 rsync 推到远程服务器，保留 30 天历史。
# tasks/backup.py
class BackupActor:
async def handle(self, data: dict) -> None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"mini-agent-backup-{timestamp}.tar.gz"
# 1. 打包
await asyncio.create_subprocess_exec(
"tar", "-czf", f"/tmp/{backup_name}",
"-C", str(self._base_dir), "data"
)
# 2. rsync 到远程
await asyncio.create_subprocess_exec(
"rsync", "-avz", f"/tmp/{backup_name}",
f"{self._rsync_host}:{self._rsync_path}/"
)
# 3. 清理本地临时文件和远程旧备份
# ...配置：
backup:
enabled: ${BACKUP_ENABLED:-false}
rsync_host: ${BACKUP_HOST:-}
rsync_path: /backup/mini-agent
retention_days: 30
scheduler:
tasks:
backup:
type: cron
cron: "0 3 * * *"现在每天自动备份，远程保留一个月滚动窗口。需要恢复时直接 tar -xzf 解压覆盖 data/ 重启就行。
2. 健康监控告警
写了个 HealthCheckActor，每 5 分钟检查一遍系统状态，有问题直接发 Telegram 告警。
监控项：
- 磁盘空间：超过 85% 告警
- 内存使用：超过 90% 告警（需要 psutil，没装会跳过）
- 数据库连接：跑一条简单查询，挂了立刻知道
- Pi RPC 可用性：调用
pi --version，5 秒超时
# tasks/health_check.py
class HealthCheckActor:
async def handle(self, data: dict) -> None:
issues = []
# 磁盘
usage = shutil.disk_usage(self._base_dir)
if usage.percent > self._disk_threshold:
issues.append(f"磁盘使用率 {usage.percent}%")
# 内存（可选）
if psutil:
mem = psutil.virtual_memory()
if mem.percent > self._memory_threshold:
issues.append(f"内存使用率 {mem.percent}%")
# 数据库
try:
self._todo_manager.list_pending()
except Exception as e:
issues.append(f"数据库异常: {e}")
# Pi RPC
try:
subprocess.run([self._pi_path, "--version"],
capture_output=True, timeout=5, check=True)
except Exception:
issues.append("pi 不可用")
if issues:
await self._notifier.alert("系统健康检查异常",
"\n".join(f"- {i}" for i in issues))现在磁盘快满了、进程挂了、数据库锁死了，都能在 5 分钟内收到 Telegram 推送。
3. 配置启动校验
在 agent.py 的 main() 开头加了个 validate_config()，服务启动前先检查配置完整性，有问题直接 fail-fast，不等运行时才炸。
检查项：
- Pi 路径存在且可执行
- 多租户模式下 admin 凭证必须配置
- 多租户模式下 bubblewrap 必须可用
- Sync 模式下对端 URL 和密码必须配置
- Backup 模式下 rsync 目标主机必须配置
def validate_config(config: dict) -> list[str]:
errors = []
# Pi 路径
pi_path = config.get("pi_path")
if not pi_path or not shutil.which(pi_path):
errors.append(f"pi_path 不存在或不可执行: {pi_path}")
# 多租户依赖
if config.get("auth", {}).get("enabled"):
if not config["auth"].get("admin_email"):
errors.append("多租户模式需要 AUTH_ADMIN_EMAIL")
if not _probe_bwrap():
errors.append("多租户模式需要 bubblewrap 支持")
# Sync 配置
sync_cfg = config.get("sync", {})
if sync_cfg.get("enabled"):
if not sync_cfg.get("peer_url"):
errors.append("Sync 模式需要 SYNC_PEER_URL")
return errors
async def main():
config = load_config()
errors = validate_config(config)
if errors:
logger.error("配置验证失败:\n" + "\n".join(f" - {e}" for e in errors))
sys.exit(1)
# 继续启动...现在配置写错了立刻就知道，不用等跑起来才发现。
4. SQLite WAL 模式
在数据库初始化时打开 WAL（Write-Ahead Logging）模式，写操作不再锁全表。
db_conn = sqlite3.connect(str(db_path))
db_conn.execute("PRAGMA journal_mode=WAL")
db_conn.execute("PRAGMA synchronous=NORMAL")
db_conn.commit()WAL 模式下：
- 读操作可以和写操作并发
- 写操作不阻塞读操作
- 数据库文件变成 3 个（
.db,.db-wal,.db-shm），备份时一起打包就行
性能提升：多租户场景下并发读写不再互相阻塞。
Phase 1：架构优化（部分完成）
5. Pi RPC 进程池（已实现，未集成）
现在每次调用 pi 都要 spawn 新进程，200-500ms 开销。写了个 PiRpcPool 维护 3 个常驻进程，请求来了直接分配，用完放回池子。
# tasks/pi_rpc_pool.py
class PiRpcPool:
def __init__(self, size: int = 3):
self._pool: asyncio.Queue[PiRpcWorker] = asyncio.Queue(maxsize=size)
async def start(self):
for _ in range(self._size):
worker = PiRpcWorker(self._pi_path)
await worker.start()
await self._pool.put(worker)
async def execute(self, prompt: str) -> PiRpcResult:
worker = await self._pool.get()
try:
return await worker.prompt(prompt)
finally:
if worker.is_alive():
await self._pool.put(worker)
else:
# 重启挂掉的 worker
new_worker = PiRpcWorker(self._pi_path)
await new_worker.start()
await self._pool.put(new_worker)预期效果：
- spawn 开销从 400ms → 0ms
- 吞吐量提升 3-5 倍
- worker 挂了自动重启
代码写完了，但还没集成到 CodeTaskExecutor。等实际遇到性能瓶颈再开启，现在请求量不大，spawn 模式够用。
6. MessageHandler 重构（已推迟）
原计划把 380 行的 MessageHandler 拆成 4 个模块（chat_handler.py, task_handler.py, topic_handler.py, command_parser.py），提升可维护性。
但仔细看了下代码，虽然长了点，但逻辑清晰，职责也分明。现在重构风险大于收益，推迟到真正难维护时再说。
Phase 2：功能增强（未开始）
原计划包括：
- 三级权限模型（ADMIN / POWER / BASIC）
- 配额透明化（API 返回剩余额度和重置时间）
- 增强
/help输出 - CI/CD 自动部署到 ECS
这些都是锦上添花，等多租户用户多了再考虑。
测试验证
全部 423 个测试通过，零破坏性变更。
$ uv run pytest tests/ -v
============================= test session starts ==============================
collected 423 items
tests/test_agent_sandbox.py .. [ 0%]
tests/test_blog_sync.py ...... [ 1%]
tests/test_code_task.py ................ [ 7%]
tests/test_conversation.py ...................................... [ 16%]
# ... 423 passed in 12.34s关键测试覆盖：
- Cron dispatcher：新增的 BackupActor 和 HealthCheckActor 注册正确
- Web API：多租户认证、配额、所有端点
- Pi RPC：沙箱隔离、流式输出、错误处理
- Message handler：所有命令路由和响应器
部署步骤
本地环境
# 1. 配置备份目标（明天开通 OSS 后补上）
echo "BACKUP_ENABLED=false" >> .env
echo "BACKUP_HOST=" >> .env
# 2. 健康检查默认开启
echo "HEALTH_CHECK_ENABLED=true" >> .env
# 3. 重启服务
systemctl --user restart mini-agent
journalctl --user -u mini-agent -f云端环境（ECS）
# SSH 登录
ssh -p 2201 deploy-pro@<ECS_IP>
# 更新代码
cd ~/mini-agent
git pull
uv sync
# 健康检查在云端也开启，备份由本地负责
systemctl --user restart mini-agent
systemctl --user status mini-agent成本
- 代码量：新增 ~450 行（3 个新文件），修改 ~80 行（agent.py + config.yaml）
- 开发时间：2 小时（得益于子代理自动化实施）
- 维护成本：几乎为零，都是自动化任务
- 存储成本：OSS 备份约 10 MB/天 × 30 天 = 300 MB，忽略不计
后续计划
立即行动
- 明天开通 OSS，配置
BACKUP_HOST和挂载点 - 监控一周健康检查日志，看有没有误报
- 手动触发一次备份，验证恢复流程
按需优化
- Pi RPC 进程池：等请求量上来、spawn 开销明显时再集成
- 三级权限模型：等多租户用户增长到 10+ 人再考虑
- CI/CD：等手动部署变成痛点再加
经验总结
1. 从个人项目到生产服务的最小动作
数据备份、健康监控、配置校验，这三件事是从”能跑”到”敢跑”的最小集合。没有它们，任何故障都是灾难性的。
2. 子代理大幅提升实施效率
这次升级用了 Claude Code 的 Agent 工具，把整个升级方案交给子代理执行。2 小时完成 Phase 0 全部 4 项任务，包括写代码、改配置、跑测试、写文档。人工做至少要一天。
关键是给子代理清晰的任务边界和验证标准：
- 从
docs/UPGRADE_PLAN.md读取完整方案 - 按 Phase 0 → Phase 1 → Phase 2 顺序执行
- 每个任务完成后跑
uv run pytest验证 - 遵守 Karpathy 原则（最小改动、外科手术式修改、目标驱动）
3. 配置校验的投入产出比极高
写 validate_config() 只用了 50 行代码，但避免了无数次”启动成功但运行时才发现配置错了”的排查时间。
4. 不要过度优化
Pi RPC 进程池写完了，但没急着集成。MessageHandler 重构计划了，但推迟了。因为现在的瓶颈不在这里，过早优化是万恶之源。
5. WAL 模式是免费午餐
两行 PRAGMA 配置，并发性能立刻上去，几乎零成本。唯一代价是数据库文件从 1 个变 3 个，但 tar 打包时一起带走就行。
项目地址：私有仓库（个人使用）
技术栈：Python 3.11 + asyncio + SQLite + LangChain + Telegram Bot
部署架构：双实例（本地调度 + 云端公网）+ Git/HTTP 双向同步
Sources
- SQLite WAL Mode — SQLite 官方文档
- APScheduler Documentation — Python 定时任务框架
- LangGraph Checkpoints — LangGraph 状态持久化
- Bubblewrap — Linux 沙箱工具
- systemd LoadCredentialEncrypted — systemd 加密凭证管理