讲解生产级 Agent(如定时任务)如何通过异步通知(Slack/Telegram)实现人工审批闭环,包含代码示例。
一个按计划运行、无人值守的 Agent,仍然需要一种停下来请求批准的方式——本文将介绍无人值守 AI Agent 的异步审批模式。
大多数 human-in-the-loop 示例都假设有人坐在终端前,随时可以在几秒内完成审批。无人值守 Agent 完全打破了这一假设:比如每晚通过 cron job 起草发票、定时抓取数据并发布市场摘要,或是在你熟睡时于凌晨 3 点运行的批处理流水线。整个闭环的另一端,并没有一个能够同步响应的人。
这并不意味着你可以跳过审批步骤,而是说审批必须异步进行,并且能够持久保存。Agent 推送一项提案,然后停止运行。人类可以在下次查看手机时进行审核,而不一定非要在接下来的十秒内完成。这个审批关卡必须能够跨越这段时间间隔持续有效。
它仍然采用与其他 Impri 集成相同的三次调用结构,但使用场景发生了变化:不再是有人一直守在聊天窗口前,而是由通知渠道(Telegram、Slack、email、push)充当交互界面。Agent 推送待执行的操作,然后退出(如果它是一个触发后无需持续关注的批处理任务),或者在长时间运行的进程中通过 long poll 阻塞等待。
import os, time, requests
API = "https://api.impri.dev"
HEADERS = {"Authorization": f"Bearer {os.environ['IMPRI_API_KEY']}"}
def propose_invoice_run(invoice_summary: str, total_amount: str):
resp = requests.post(f"{API}/v1/actions", headers=HEADERS, json={
"kind": "finance.invoice_batch",
"title": f"Nightly invoice run — {total_amount} across batch",
"preview": {"format": "markdown", "body": invoice_summary},
"idempotent": False,
"undo": "Void the batch via the billing dashboard before end of day",
"expires_in": 43200, # 12h — long enough to survive the night
})
resp.raise_for_status()
return resp.json()["id"]
def wait_for_decision(action_id: str, poll_every=60):
while True:
r = requests.get(f"{API}/v1/actions/{action_id}", headers=HEADERS).json()
if r["status"] != "pending":
return r
time.sleep(poll_every)
cron job 调用 propose_invoice_run 后退出,另一个独立进程——可能是下一次定时运行的任务,也可能是一个小型 daemon——稍后再调用 wait_for_decision。API 并不要求这两个阶段必须由同一个进程完成。
在这里,expires_in 比在同步流程中更加重要。如果审核要等到八小时后喝早间咖啡时才进行,那么默认的 5 分钟有效期毫无用处。应该根据操作在多长时间内仍然有效来选择窗口,而不是根据你希望审核多快完成来设置:
如果还没来得及查看,窗口就已经关闭,那么该操作会过期,并且永远不会执行——在流水线中,应当像处理被拒绝的操作一样处理过期操作;同时,可以考虑将底层任务重新加入队列,而不是悄无声息地将其丢弃。
这里有两种切实可行的模式,应当根据无人值守 Agent 的部署方式进行选择:
无状态 cron job: 推送操作后立即退出。在下一次定时运行时,先检查上一次运行所推送的所有操作,再开始执行新任务。这样可以避免让一个进程持续运行数小时。
长时间运行的 daemon: 使用 impri_await_decision(MCP),或者使用带 backoff 的轮询循环(REST)并直接阻塞等待,因为该进程本来就会常驻运行。
无论采用哪种方式,Agent 真正产生副作用的操作——发送发票、发布摘要——都必须放在 status == "approved" 检查之后,而不是之前。无人值守 Agent 正是那种不能指望“模型会记得检查”的场景;代码执行路径必须确保未获批准的分支根本无法触达实际操作。
Impri 会存储操作、向你发送通知并保存审批决定,但它不会决定无人值守流水线应当何时运行、重试或 fan out;这些仍然是调度器的职责,无论你使用的是 cron、workflow engine,还是其他触发 Agent 的机制。另外,只有当无人值守进程没有同时持有可绕过封装层、直接执行操作的长期 credential 时,它才算得上真正的审批关卡。要了解本文所基于的底层三次调用模式,请参阅如何为 AI Agent 添加人工审批;要了解这个问题针对调度器的实现方式,请参阅如何使用人工审批为 cron job 设置关卡。
下一篇:如果你的无人值守 Agent 需要的是强制停止机制,而不是针对每项操作分别设置关卡,请参阅如何为 AI Agent 添加 kill switch。
如需采取进一步措施,你可以考虑屏蔽此人和/或举报滥用行为。