自建 24 小时在线的个人 Agent

学习目标：用三种可行路径搭出一个真正常驻的个人 Agent，掌握调度循环、工具白名单、审批闸门与审计日志的最小实现，并算出量级成本
预计时间：60 分钟（含动手）
难度：⭐⭐⭐

先说结论

自建常驻 Agent 的门槛比想象中低：一台最低配云主机 + 一个 while 循环 + 一个工具白名单 + 一个审批队列 + 一个只追加日志，就能跑起来。

难的不是「让它跑」，是让它只在你允许的范围内跑。所以本文的顺序是：

先决定「买还是建」
选一条路径（云 VM / 开源运行时 / 托管 SDK）
抄一份最小骨架，跑通只读巡检
按四阶段上线，最后才碰写操作

如果你还没有 OpenClaw 基础，建议先看 OpenClaw 模块，它的 Heartbeat 调度在深度指南里有完整拆解。

一、买还是建

维度	用现成产品	自建
上手时间	分钟级	半天到两天
连接器	现成 4000-6000+	自己接，接到哪算哪
权限控制	受产品限制，只能选档位	完全可定制
数据位置	厂商云，按其政策	自己的机器
成本结构	订阅费，可预测	VM + token，随用量波动
出问题	提工单	自己修
适用	通用办公流程	特殊流程、敏感数据、想完全掌控

一条实用判断：如果你的需求能被现成产品的连接器覆盖，别自建。 自建的真正理由只有三个——数据不能出去、流程太特殊、需要产品没提供的权限粒度。

二、三条可行路径

路径 A：云端 VM + 常驻调度（最可控）

一台小 VM，一个 systemd 服务，一个调度循环。所有东西都在你手里。

text

云 VM (2 vCPU / 4 GB)
├── systemd: personal-agent.service   # 常驻进程，崩了自动重启
├── /srv/agent/
│   ├── agent.py          # 调度 + 决策循环
│   ├── policy.yaml       # 工具白名单与审批规则
│   ├── state/            # 任务状态、断点
│   ├── logs/audit.jsonl  # 只追加审计
│   └── outbox/           # 待审批动作
└── cron (可选): 每日清理、备份

优点：完全可控、成本最低、数据不出你的账号
缺点：连接器要自己写；浏览器自动化要自己维护；安全全靠自己
适合：读多写少、流程固定、对数据位置敏感的场景

调度用 systemd timer 还是进程内循环？进程内循环 + 状态持久化更简单：它能保证「一个任务没跑完不会被下一个覆盖」，而 cron 会。

路径 B：开源运行时（最快见效）

以 OpenClaw 为代表的自托管 Agent 运行时，已经把 Gateway、调度、工具执行、记忆分层、渠道适配都做好了。你要做的是配置和写 Skill，而不是造轮子。

优点：开箱即用，社区 Skill 可直接复用；有现成的 Heartbeat 调度机制
缺点：跑在你自己的机器上，安全边界要自己划；升级会带来配置漂移
适合：想快速验证「常驻 Agent 对我到底有没有用」的人

它的安全模型（沙箱、权限、审计）在这一篇有专门讨论，自建前值得读一遍。

路径 C：托管 Agent SDK / API（最省运维）

不自己管进程和 VM，把调度托管给平台，你只写「决策逻辑 + 工具」。

两种形态：

厂商托管 Agent 平台：Anthropic 的 Claude Agent SDK / Managed Agents 路线（Duet 的凭证隔离设计就沿用它）、OpenAI 的 Agent 与插件体系。省掉 VM 和凭证管理，但绑定厂商；具体包名与能力以官方文档为准。
OpenAI 兼容 API + 自建薄壳：任何兼容 /v1/chat/completions 的模型都能接，工具执行仍在你自己的进程里。灵活性最高，也最容易换模型。
优点：不用管可用性、扩容、凭证保险箱
缺点：数据在厂商侧；平台改 API 你就得改；深度定制受限
适合：已经在用某家生态、不想碰运维的团队

选型口诀：要数据主权选 A，要速度选 B，要少运维选 C。 三者可以混——比如调度在 A、模型走 C、工具执行留在本地。

三、最小实现骨架

下面这份代码是完整可跑的骨架，只依赖 Python 标准库。它实现了四件事：调度循环、工具白名单、审批闸门、只追加日志。

python

#!/usr/bin/env python3
"""agent.py — 最小常驻 Agent：调度 + 白名单 + 审批 + 审计"""
import json, os, time, urllib.request, uuid
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).parent
STATE, LOG = ROOT / "state", ROOT / "logs" / "audit.jsonl"
OUTBOX = ROOT / "outbox"
for p in (STATE, LOG.parent, OUTBOX):
    p.mkdir(parents=True, exist_ok=True)

# ---------- 1. 工具白名单：名字 -> (函数, 是否只读) ----------
def read_web(url: str) -> str:
    req = urllib.request.Request(url, headers={"User-Agent": "personal-agent/1.0"})
    with urllib.request.urlopen(req, timeout=20) as r:
        return r.read(200_000).decode("utf-8", "ignore")

def write_note(path: str, text: str) -> str:
    target = (ROOT / "notes" / path).resolve()
    if ROOT.resolve() not in target.parents:          # 防路径穿越
        raise PermissionError(f"path escapes workspace: {path}")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(text, encoding="utf-8")
    return str(target)

TOOLS = {
    "read_web":   (read_web,   True),
    "write_note": (write_note, False),
}

# ---------- 2. 硬禁区 ----------
FORBIDDEN = {"transfer_funds", "trade_securities", "delete_data", "send_email"}

def gate(tool: str) -> str:
    """返回 allow / approve / deny。未知工具一律 approve，绝不默认放行。"""
    if tool in FORBIDDEN:
        return "deny"
    if tool not in TOOLS:
        return "approve"                       # 未知 = 不可信
    return "allow" if TOOLS[tool][1] else "approve"

# ---------- 3. 只追加审计 ----------
def audit(**fields) -> None:
    rec = {"ts": datetime.now(timezone.utc).isoformat(), **fields}
    with LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

# ---------- 4. 审批闸门（文件即队列） ----------
def request_approval(task_id: str, tool: str, args: dict) -> str:
    pid = uuid.uuid4().hex[:8]
    req = {"id": pid, "task_id": task_id, "tool": tool, "args": args,
           "requested_at": time.time(), "ttl": 600}
    (OUTBOX / f"{pid}.pending.json").write_text(
        json.dumps(req, ensure_ascii=False, indent=2), encoding="utf-8")
    audit(event="approval_requested", **{k: req[k] for k in ("id", "task_id", "tool")})
    return pid

def poll_approval(pid: str) -> str:
    """approved / rejected / expired / pending —— 由外部（人）写 approved/rejected 文件"""
    for verdict in ("approved", "rejected"):
        if (OUTBOX / f"{pid}.{verdict}").exists():
            return verdict
    pending = OUTBOX / f"{pid}.pending.json"
    if pending.exists():
        req = json.loads(pending.read_text(encoding="utf-8"))
        if time.time() - req["requested_at"] > req["ttl"]:
            pending.rename(OUTBOX / f"{pid}.expired.json")
            return "expired"
    return "pending"

def resume_approved() -> None:
    """每轮开头扫描 outbox：已批准的补执行，过期/拒绝的丢弃。"""
    for f in OUTBOX.glob("*.pending.json"):
        pid = f.name.split(".")[0]
        verdict = poll_approval(pid)
        if verdict != "approved":
            continue
        req = json.loads(f.read_text(encoding="utf-8"))
        tool, args = req["tool"], req["args"]
        try:
            result = TOOLS[tool][0](**args)
            audit(event="tool_ok", task_id=req["task_id"], tool=tool,
                  result=str(result)[:500], approved_by="human")
        except Exception as e:
            audit(event="tool_error", task_id=req["task_id"], tool=tool, error=repr(e))
        f.unlink()                      # 执行完即出队，避免重复

# ---------- 5. 决策：让模型产出结构化动作 ----------
PROMPT = """你是常驻个人 Agent。根据目标输出一个 JSON 动作，不要输出其他内容：
{"tool": "<工具名>", "args": {...}, "reason": "<一句话理由>"}
可用工具：read_web(url), write_note(path, text)
如果无事可做，输出 {"tool": "noop", "args": {}, "reason": "..."}"""

def decide(goal: str, context: str, model: str, api_key: str,
           base_url: str = "https://api.openai.com/v1") -> dict:
    body = json.dumps({
        "model": model,
        "messages": [{"role": "system", "content": PROMPT},
                     {"role": "user", "content": f"目标：{goal}\n上下文：{context}"}],
        "response_format": {"type": "json_object"},
        "temperature": 0.2,
    }).encode()
    req = urllib.request.Request(
        f"{base_url}/chat/completions", data=body,
        headers={"Authorization": f"Bearer {api_key}",
                 "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        content = json.loads(r.read())["choices"][0]["message"]["content"]
    return json.loads(content)

# ---------- 6. 单轮执行 ----------
def run_once(goal: str, dry_run: bool = False) -> None:
    task_id = uuid.uuid4().hex[:8]
    ctx_file = STATE / "last_result.json"
    context = ctx_file.read_text(encoding="utf-8") if ctx_file.exists() else "（无）"

    if dry_run:
        action = {"tool": "read_web", "args": {"url": "https://example.com"}, "reason": "dry-run"}
    else:
        action = decide(goal, context, os.environ["MODEL"], os.environ["API_KEY"],
                        os.environ.get("BASE_URL", "https://api.openai.com/v1"))

    tool = action.get("tool", "noop")
    args = action.get("args", {}) or {}
    if tool == "noop":
        audit(event="noop", task_id=task_id, reason=action.get("reason", ""))
        return

    verdict = gate(tool)
    audit(event="gated", task_id=task_id, tool=tool, verdict=verdict,
          args={k: str(v)[:200] for k, v in args.items()})

    if verdict == "deny":
        return
    if verdict == "approve":
        pid = request_approval(task_id, tool, args)
        print(f"[待审批] {tool} id={pid}（10 分钟内有效）")
        return

    try:
        fn, _ = TOOLS[tool]
        result = fn(**args)
    except Exception as e:                                  # 失败也要留痕
        audit(event="tool_error", task_id=task_id, tool=tool, error=repr(e))
        return

    audit(event="tool_ok", task_id=task_id, tool=tool, result=str(result)[:500])
    STATE.joinpath("last_result.json").write_text(
        json.dumps({"task_id": task_id, "tool": tool, "result": str(result)[:2000]},
                   ensure_ascii=False), encoding="utf-8")

# ---------- 7. 调度循环 ----------
def main(interval: int, goal: str) -> None:
    while True:
        try:
            resume_approved()                  # 先处理上一轮攒下的审批结果
            run_once(goal)
        except Exception as e:
            audit(event="loop_error", error=repr(e))
        for _ in range(interval):              # 可被打断的睡眠
            time.sleep(1)

if __name__ == "__main__":
    goal = os.environ.get("GOAL", "巡检指定页面，有变化就写一条笔记")
    main(int(os.environ.get("INTERVAL_SEC", "3600")), goal)

这段代码故意省略了三样东西：浏览器自动化、向量记忆、多渠道通知。它们是「能力」，不是「骨架」；骨架没跑通就加能力，等于在地基没干的时候装修。

但三样东西一样都没省：白名单（TOOLS 之外的调用一律进审批）、闸门（gate 默认 approve）、日志（每个分支都 audit，包括错误）。

冷启动验证：

bash

export API_KEY=...  MODEL=gpt-...  INTERVAL_SEC=3600
python3 agent.py            # 常驻，每小时醒一次
python3 -c "import agent; agent.run_once('test', dry_run=True)"   # 单轮干跑

四、四个必须自己补的工程件

1. 幂等

同一个任务被重复执行，不能产生重复副作用。做法：给每个任务一个幂等键（日期 + 目标哈希），写完结果先落盘再执行下一步。

2. 成本熔断

python

MONTHLY_BUDGET_USD = 50.0
if spent_this_month() > MONTHLY_BUDGET_USD:
    audit(event="budget_exceeded"); raise SystemExit("budget guard")

没有熔断的常驻 Agent 等于一张没有限额的信用卡。它在凌晨三点循环调用工具时，你不会知道。

3. Kill switch

一个文件就是开关：

python

# 放在调度循环体开头
if (ROOT / "PAUSED").exists():
    time.sleep(60)
    continue          # 跳过本轮，不进决策，也不消耗 token

touch PAUSED 立刻停所有调度。比你 ssh 上去 kill 进程快得多，也安全得多。

注意它和「暂停单个任务」不是一回事：前者连模型都不调用，后者只是让任务停在审批点。真正的紧急刹车应该做到前者。

4. 状态恢复

进程重启后，未完成的审批请求要能继续被轮询。把 outbox 里的 .pending.json 在启动时扫一遍，重新挂回等待列表。

五、成本估算：给区间，不给假精确

成本由三块构成：

text

月成本 ≈ VM/月 + (每月 token 量 × 模型单价) + 存储与出网

VM 部分（量级参考，以云厂商官网价目为准）

配置	用途	月成本量级
1-2 vCPU / 2-4 GB	纯 HTTP 巡检、无浏览器	$4 - $25
4 vCPU / 8 GB	跑 Chromium、并发 2-3 个任务	$20 - $60
8 vCPU / 16 GB	多 Agent 并行、本地小模型兜底	$50 - $150

区间下限来自欧洲低价 VPS 的入门档，上限来自主流公有云的按需实例。年付/预留能压掉一大截，按需最贵。 具体数字每隔几个月就会动，别照抄。

Token 部分（这部分才是主要变量）

token 量只取决于两个参数：巡检频率 和 单次上下文大小。

场景	计算	每月输入 token 量级
每天 1 次晨报	30 × 30k	~1 M
每小时 1 次巡检	24 × 30 × 30k	~22 M
每 15 分钟 1 次	96 × 30 × 30k	~86 M
每 5 分钟 1 次 + 长上下文	288 × 30 × 80k	~690 M

把这张表的数字乘上你选的模型单价，就是 token 成本。旗舰模型和轻量模型之间可能差一个数量级，所以「一个月多少钱」这个问题没有统一答案——先算 token 量，再看价目表。

三条降本手段（按性价比排序）

分级模型：巡检用便宜模型，判断「有没有异常」；只有异常时才叫贵模型做分析。这一条通常能省 80% 以上。
减少唤醒：把 5 分钟改成 1 小时，成本直接降 12 倍。绝大多数「监控」需求根本不需要 5 分钟粒度。
压缩上下文：别把整段历史塞进去。只带「上次结果摘要 + 当前目标」，token 量能降一个量级。

反过来说：如果你的 Agent 一个月 token 花费超过一台 VM 的百倍，那说明设计错了，不是模型贵。

六、上线顺序：先只读，再写

这是全文最重要的一节。不要跳阶段。

阶段	允许的动作	观察指标	进入下一阶段的条件
0. 影子只读	只能读，结果只写本地文件，通知只发给你自己	抓取成功率、误报率、日志完整性	连续 7 天无崩溃、误报率可接受
1. 草稿	可以生成草稿并写入 outbox，但不能发出	草稿被采纳的比例、需要重写的次数	采纳率 > 50%，且你能说清不采纳的原因
2. 审批写	所有写操作走审批闸门	审批通过率、平均审批耗时、过期率	通过率稳定、过期率 < 10%
3. 限额自动写	白名单内的低风险写操作免审批，但有次数与金额上限	出错率、撤销次数、单次最大影响面	只在明确低风险的动作上开启

每个阶段都要有退出条件，而不只是进入条件。出了事故要能退回上一阶段，而不是直接关机。

上线前最后一张清单

[ ] 有 PAUSED 开关，且你测试过它真的能停
[ ] 有月度预算熔断，数值是你愿意损失的钱
[ ] 审计日志写在 Agent 无权修改的位置
[ ] 所有写操作默认走审批，白名单是显式配置的
[ ] 每个连接器都是最窄 scope，且有撤销方案
[ ] 失败路径会留日志（不要只记成功）
[ ] 有备份，且恢复流程演练过一次
[ ] 你知道这台机器上存了哪些凭证、怎么轮换

七、常见踩坑

坑	现象	解法
用 cron 跑长任务	上一轮没跑完，下一轮又起来，结果互相覆盖	进程内循环 + 锁文件
无上限重试	一个挂了的外部 API 让你整夜烧 token	每次任务最多重试 2 次，指数退避
日志写在 Agent 自己的盘上	Agent 出问题时日志一起没了	远端 append-only 存储
浏览器自动化不稳定	页面改版就失败，且失败信息全是超时	关键流程加 DOM 断言，失败时截图存档
审批用聊天消息	消息被刷走，动作一直挂着	审批走独立队列 + 过期时间
凭证写在环境变量里	一次 `env` 泄露全失	用密钥管理服务或代执行 broker

本节小结

自建常驻 Agent 的最小集是：调度循环 + 工具白名单 + 审批闸门 + 只追加日志，四件都在上面的代码里
三条路径按「数据主权 / 速度 / 运维成本」取舍，可以混用：调度在 VM、模型走 API、工具留在本地
成本的大头是 token 不是 VM；先算 token 量级，再看模型单价，别问「一个月多少钱」
降本最有效的一招是分级模型：便宜的负责发现，贵的负责分析
上线必须四阶段：影子只读 → 草稿 → 审批写 → 限额自动写，每阶段有进入和退出条件
未知工具默认审批、写操作单独授权、监控日志不可被 Agent 修改——这三条比任何模型能力都重要

← 上一节：权限边界、隐私与安全 | 返回模块首页：个人智能体（常驻 Agent）

自建 24 小时在线的个人 Agent ​

先说结论 ​

一、买还是建 ​

二、三条可行路径 ​

路径 A：云端 VM + 常驻调度（最可控） ​

路径 B：开源运行时（最快见效） ​

路径 C：托管 Agent SDK / API（最省运维） ​

三、最小实现骨架 ​

四、四个必须自己补的工程件 ​

1. 幂等 ​

2. 成本熔断 ​

3. Kill switch ​

4. 状态恢复 ​

五、成本估算：给区间，不给假精确 ​

VM 部分（量级参考，以云厂商官网价目为准） ​

Token 部分（这部分才是主要变量） ​

三条降本手段（按性价比排序） ​

六、上线顺序：先只读，再写 ​

上线前最后一张清单 ​

七、常见踩坑 ​

本节小结 ​

自建 24 小时在线的个人 Agent

先说结论

一、买还是建

二、三条可行路径

路径 A：云端 VM + 常驻调度（最可控）

路径 B：开源运行时（最快见效）

路径 C：托管 Agent SDK / API（最省运维）

三、最小实现骨架

四、四个必须自己补的工程件

1. 幂等

2. 成本熔断

3. Kill switch

4. 状态恢复

五、成本估算：给区间，不给假精确

VM 部分（量级参考，以云厂商官网价目为准）

Token 部分（这部分才是主要变量）

三条降本手段（按性价比排序）

六、上线顺序：先只读，再写

上线前最后一张清单

七、常见踩坑

本节小结