主动型 Agent 怎么自己写：定时唤醒、草案入队、审批后才动手

一句话总结：主动型 Agent 和普通对话机器人的区别只有一个，它先开口。Anthropic 的 Claude Duet 就是这个形态：跑在一台云电脑上，早上主动给你发当日简报，发送消息、付款、签字这些动作必须等你点头才做。这篇用一个 194 行的单文件实现把同一套结构搭出来，实测跑通了去重、审批、幂等三件事。

一、问题在哪

被动 Agent 很简单，你问一句它答一句，一切动作都在当轮对话的语境里，出了事你能立刻看见。

主动 Agent 不一样。它在你不在的时候醒来，自己读材料，自己攒出一个判断，然后决定要不要打扰你。这里面有三件事必须靠工程解决，模型本身给不了答案：

它凭什么醒，醒了以后生成的东西存在哪，谁有权把"待办"变成"已发送"。

我见过最常见的错误做法是让模型在一个循环里直接把事情做完。跑起来很爽，出事很惨：它会在凌晨三点给你客户发一封语气热情但内容空泛的邮件，而且连着发三封，因为循环里没有去重。

二、前置条件

Python 3.9 以上，只用标准库，SQLite3 自带。不需要装任何包。

我实测用的环境是 Windows + Python 3.14，代码在 Linux 上一样能跑。文件结构就两个：agentd.py 和一个自动生成的 agentd.db。

核心设计是一个状态机加一张表。状态只有五个：pending（草案等人看）、approved（人批了）、rejected（人否了）、executed（做完了）、expired（放太久自动作废）。

图 1：五个状态和四条转移。关键约束是任何执行动作的前置状态只能是 approved

三、第一步：定时唤醒，别用死循环

唤醒这块做个区分：什么时候检查和什么时候干活是两件事。

检查可以频繁，比如每 30 秒 tick 一次，代价很低；干活必须收敛到明确的时间窗，否则你在凌晨刷到一条晨报就是产品事故。

BRIEF_HOUR = 8
DRAFT_TTL_HOURS = 48

def tick():
    c = conn()
    n = expire_stale(c)
    if n:
        print(f"[过期] {n} 条草案超过 {DRAFT_TTL_HOURS} 小时未处理")
    if datetime.now().hour >= BRIEF_HOUR:
        ctx = collect_context()
        day = datetime.now().strftime("%Y-%m-%d")
        ok = add_draft(c, "morning_brief", "self", draft_brief(ctx), day)
        print("[生成] 晨报草案 " + ("已入队" if ok else "今天已生成过，跳过"))
    else:
        print(f"[等待] 未到 {BRIEF_HOUR} 点，本轮不生成")

这段有个细节：tick 里只做"生成草案"和"清理过期"，不做执行。生成和执行分开，是这个结构能稳住的前提。

expire_stale 做的是另一件事，它把放了超过 48 小时还没人处理的草案标成 expired。这条不写，你出差一周回来会看到七条过期的晨报排在队列里，然后一次性全批了，等于批量发废消息。

四、第二步：去重键，防的是你自己

调度器一定会重复触发。进程重启、容器漂移、cron 重叠，随便一个都会让同一天的晨报生成两次。

解决办法不是加 if 判断，是把去重键做成数据库唯一约束：

def dedup(kind, day, target):
    raw = f"{kind}|{day}|{target}".encode()
    return hashlib.sha1(raw).hexdigest()[:16]

def add_draft(c, kind, target, payload, day):
    key = dedup(kind, day, target)
    try:
        c.execute(
            "INSERT INTO drafts(kind,payload,status,dedup_key,created_at,updated_at)"
            " VALUES(?,?,'pending',?,?,?)",
            (kind, json.dumps(payload, ensure_ascii=False), key, now(), now()),
        )
        c.comMIT()
        return True
    except sqlite3.IntegrityError:
        return False   # 今天已经生成过，跳过

把幂等交给数据库而不是交给代码逻辑，好处是并发也不会漏。两个进程同时插入，一个成功一个收到 IntegrityError，逻辑上永远只有一条。

实测跑两次的结果是这样：

=== 第一次 tick ===
[生成] 晨报草案 已入队
=== 第二次 tick（验证去重）===
[生成] 晨报草案 今天已生成过，跳过

五、第三步：草案存"终稿"，不存"意图"

这一条我认为是整个设计里最容易被忽略、后果最严重的。

草案进队列的时候，必须存下最终要发出去的完整内容。不能存一句"给客户 A 发个催附件提醒"，然后等审批通过之后让模型重新生成一遍。

原因是模型有随机性。你批的是 A 版本，重跑生成出的是 B 版本，中间没有人看过 B，但发出去的是 B。这个 bug 平时测不出来，出了事也复现不了。

所以 payload 里存的就直接是可执行内容：

def draft_brief(ctx):
    urgent = [m for m in ctx["unread"] if m["urgent"]]
    lines = [f"早上好，{ctx['date']}待办草案："]
    for m in urgent:
        lines.append(f"- 需回复：{m['from']} - {m['subject']}")
    for t in ctx["tickets"]:
        lines.append(f"- 待跟进：工单 {t['id']} {t['title']}（已停留 {t['age_hours']} 小时）")
    lines.append(f"共 {len(urgent)} 封紧急邮件、{len(ctx['tickets'])} 个滞留工单。")
    return {"action": "send_message", "channel": "self", "body": "\n".join(lines)}

真实项目里，把 draft_brief 换成一次 LLM 调用就行，其余结构不用动。示例用规则生成，是为了让它离线可跑、结果可复现。

六、第四步：审批和执行分开，中间隔一层状态

审批就是个状态迁移，pending → approved 或 rejected。执行只捞 approved：

def execute_approved():
    c = conn()
    rows = c.execute("SELECT * FROM drafts WHERE status='approved' ORDER BY id").fetchall()
    for r in rows:
        payload = json.loads(r["payload"])
        try:
            result = do_action(payload)
            c.execute("UPDATE drafts SET status='executed', result=?, updated_at=? WHERE id=?",
                      (result, now(), r["id"]))
            c.commit()
        except Exception as e:
            c.execute("UPDATE drafts SET status='failed', result=?, updated_at=? WHERE id=?",
                      (f"{type(e).__name__}: {e}", now(), r["id"]))
            c.commit()

状态在动作成功之后才改，顺序不能反。先改成 executed 再执行，中间挂了就会留下一条"看起来做完其实没做"的记录，排查时最难查。

实测链路：

=== list ===
#1 [morning_brief] 2026-11-01 09:22:22
    早上好，11月01日待办草案：
    - 需回复：客户 A - 合同附件缺失
    - 需回复：运维 - 昨晚备份失败一次
    - 待跟进：工单 4012 登录验证码收不到（已停留 9 小时）
    共 2 封紧急邮件、1 个滞留工单。
=== approve 1 ===
[批准] #1 成功
=== exec ===
[执行] #1 成功 -> 已写入 sent.log
=== 再 exec（验证幂等）===
(没有已批准项)

再跑一次 exec 什么都不做，因为状态已经不是 approved 了。这就是状态机比布尔字段好用的地方。

七、坑清单

按我实际踩到的概率排序，不是按理论重要性。

1. 没设去重键，一早上收到七条一样的晨报。 命中率最高。调度器重启一次就多一条，因为"今天生成过没有"这个判断如果靠内存里的变量，进程一挂就归零。

2. 生成和执行写在一个函数里，生成失败连带执行失败。 表现是某些天完全没动静，日志里只有一条异常。分开以后，生成失败不影响已批准的执行。

3. 草案存意图不存终稿，批准的和发出的不是一份。 前面讲过，这个 bug 的可怕之处是不可复现。

4. 审批队列没有超时，一周后批量批准了七条旧消息。 加 TTL 只要三行代码，不加就得靠人记性。

5. 动作没有分级，把所有操作都塞进审批。 反过来的错误。如果连"读一封邮件"都要你点确认，两天之后你就不会再点开这个队列了。我的做法是只把不可逆动作送审（发消息、付款、删除、签字），读操作直接做。

八、边界和下一步

这套东西的上限很清楚。

它不解决"草案写得好不好"这个问题，那取决于你的模型和上下文。它也不处理多用户权限，示例里只有一个人在审批，真要做团队版得加 owner 字段和权限校验。

还有一个我没解决的：审批疲劳。当天草案超过 5 条时，通过率会明显下降，人会开始无脑点批准。这一点目前只能靠控制条数，没有更好的办法。

下一步我打算加的是回执和撤回。动作发出后记录消息 ID，允许在 10 分钟内撤回，这样误批的代价能降一个量级。

站里之前写过最小 Agent 循环和预算护栏，那两篇解决的是"Agent 能动"和"Agent 不失控"，这篇解决的是"Agent 该不该动"。三块拼起来才是一个能上生产的东西。

参考资料

• Anthropic - Claude Duet 产品说明与审批机制
• Python 官方文档 - sqlite3 模块与事务行为
• SQLite 官方文档 - UNIQUE 约束与冲突处理