后台 Agent:webhook、邮件与定时任务触发,收件箱式审批

今天要解决的问题

这是专题篇的一课,建议在 Day 25 · LangGraph 进阶:checkpoint、人工审批 之后读。专题篇不按顺序学,哪篇和你手上的事有关就先看哪篇。

到 Day 25 为止,我们的 Agent 都是「有人在对面敲字」才动:你在命令行里输入一句话,它跑一圈,回你一句。真实的客服系统不是这样的。客户发了邮件、工单系统里新建了一张单、某张单已经挂了三周没人回,这些事情发生的时候,没有人坐在 Agent 对面。我们希望 Agent 自己醒过来,把能做的先做了(读工单、分类、起草回复),然后把需要人拍板的那一步放进一个收件箱,等客服有空的时候一条条处理。

LangChain 在 2025 年 1 月给这类东西起了个名字,叫 ambient agent(环境智能体,也有人翻成「后台智能体」):由环境里的事件触发,在后台持续运行,只在需要人的时候才打扰人。名字是新的,里面的工程问题却都很老:webhook 怎么验签、重复投递怎么去重、两个 worker 怎么不抢同一个任务、进程崩了怎么续上、失败三次的任务放哪。今天就是把这些老问题和 Day 25 学过的 interrupt 拼在一起。

学完今天,你要能做到:

1. 说清 webhook、邮件轮询、定时任务三种触发源各自的特点,以及为什么它们都应该先写进一张事件表,而不是收到就直接跑 Agent。
2. 写出一个带签名校验和防重放的 webhook 入口,能挡住伪造请求,能识别上游的重复投递。
3. 用 SQLite 实现「原子领任务、失败重试、超过次数进死信、卡住的任务放回队列」,并说出它和 Day 37 的任务队列是什么关系。
4. 用 LangGraph 的 interrupt 做一个收件箱:每条待办支持批准、修改后发送、提意见让 Agent 重写、忽略四种处理。
5. 让一个跑到一半崩掉的工单从失败的那个节点接着跑,而不是从头再来。

预计用时 4–5 小时。前置:Day 23 的 cloudledger.db(41 张工单、30 个客户),Day 24 的 llm.py,Day 25 的 checkpointer、thread_id、interrupt。Day 37 讲过的队列、重试、幂等如果看过会更顺,没看过也不影响。

聊天窗口之外的 Agent

先把「后台 Agent」和我们之前做的东西区分清楚。区别不在模型,也不在图怎么画,而在谁来启动一次运行、结果交给谁。

聊天式 Agent 是同步的:人发起,人等着,结果直接回给这个人。一次运行通常几秒到几十秒,中间出错了,人会看到报错然后重试。后台 Agent 是异步的:启动它的是一个事件(一条 webhook、一封邮件、一个时间点),没人在等它;它的产出也不一定直接给最终用户,更多时候是给内部的人一条待办。

LangChain 那篇介绍文章里把后台 Agent 找人的方式分成三种,我觉得这个分法很实用:

• 通知(notify):告诉人发生了一件事,但 Agent 自己不处理。比如「这张工单提到了律师函,我没动它,你看一下」。
• 提问(question):Agent 卡住了,需要人给一个它自己查不到的信息。比如「客户说的『上次那张发票』是哪张,系统里有三张候选」。
• 审核(review):Agent 准备做一件有后果的事(发邮件给客户、建退款申请),先给人看一眼,人可以批准、修改或者否掉。

今天的 CloudLedger 场景主要是第三种:新工单进来,Agent 起草一封回复,客服在收件箱里过一遍再发。第一种和第二种在练习里做。

为什么不直接让 Agent 把回复发出去?因为你还没有足够的数据证明它可靠。Day 29 到 Day 35 一直在讲,Agent 上线前要有评测集和回归门禁;即使评测分数不错,对外发出去的每一个字都代表公司。收件箱是一个过渡形态:先让人审,同时把人的每一次修改都记下来(这就是 Day 144 要讲的反馈闭环),等某一类工单连续几百条都被原样批准,再考虑让这一类自动发出。

三种触发源

webhook 是对方系统推给你:工单系统新建了一张单,它往你配置好的 URL 发一个 POST 请求,正文是一段 JSON。优点是实时,缺点是你得有一个公网能访问的地址,还得面对这几个事实:对方不知道你处理得多慢,一般几秒到几十秒没收到 2xx 就算失败,然后重试;重试意味着同一个事件可能到你这里两次、三次;任何人知道了你的 URL 都可以往里发假数据。Stripe、GitHub、飞书、企业微信的 webhook 文档,翻来覆去讲的就是这三件事:快速返回、去重、验签。

邮件 一般是你去拉:每分钟连一次客服邮箱,用 IMAP 把未读邮件取下来。好处是不需要公网地址,邮箱本身就是一个持久化的队列;坏处是有延迟,而且邮件格式五花八门(HTML、附件、转发链、编码)。如果邮件量大,也可以用 Postmark、Cloudflare Email Workers 这类服务把收到的邮件转成 webhook,这样就和第一种合并了。

定时任务 是你给自己定的闹钟:每天早上 9 点扫一遍「超过 20 天还开着的工单」,每周一生成上周的处理报告。它没有外部输入,但同样要考虑重复:定时器有时会因为机器重启补跑一次,你手动又跑了一次,结果同一张工单被提醒了两遍。

这三种来源差别很大,但我建议所有来源都先变成事件表里的一行,然后由同一个 worker 处理。理由有三个。第一,webhook 入口只做验签和落库,几毫秒就能回 202,不会因为模型慢而让上游超时重试。第二,去重、重试、失败记录只需要写一遍。第三,出了问题你只需要看一张表:哪些事件还没处理、哪些在等人、哪些失败了。这其实就是 Day 37 讲的任务队列,只是今天为了少装东西,直接用 SQLite 当队列。量上来以后换成 Redis、RabbitMQ 或者云厂商的队列服务,结构不用变。

准备目录和依赖

今天的代码放在练习仓的 topics/ambient-agents/ 下面,和 w4/ 平级。业务库直接用 Day 23 造的那份 w4/cloudledger/cloudledger.db。

cd agent-60-lab
uv add langgraph langchain-openai langgraph-checkpoint-sqlite fastapi uvicorn
mkdir -p topics/ambient-agents
cd topics/ambient-agents

我实测用的版本是 langgraph 1.2.16、langgraph-checkpoint-sqlite 3.1.2、langchain-openai 1.7.2、fastapi 0.159.1、uvicorn 0.54.0,和 Day 25 一致。模型还是 .env 里的 DeepSeek deepseek-flash。

llm.py 直接从 Day 24 复制一份过来,内容一字不改,这样这个目录里的脚本都能按文件路径直接跑,不用处理跨目录导入:

# agent-60-lab/topics/ambient-agents/llm.py
"""和 Day 24 的 w4/lgdemo/llm.py 一模一样,复制一份过来,省得跨目录导入。"""
import json
import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()


def get_llm(temperature: float = 0) -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ["LLM_MODEL"],
        base_url=os.environ["LLM_BASE_URL"],
        api_key=os.environ["LLM_API_KEY"],
        temperature=temperature,
        extra_body=json.loads(os.environ.get("LLM_EXTRA_BODY", "{}")),
    )

第一层:事件表

先建两张表。events 是队列,每个外部事件一行;outbox 是「已经发给客户的回复」,今天不真的发邮件,写进这张表就算发出去了。

# agent-60-lab/topics/ambient-agents/store.py
"""事件表:所有触发源(webhook、邮件、定时任务)进来的东西先落到这里,再由 worker 慢慢处理。"""
import sqlite3
from datetime import datetime
from pathlib import Path

HERE = Path(__file__).parent
EVENTS_DB = HERE / "ambient.db"
BIZ_DB = HERE.parents[1] / "w4" / "cloudledger" / "cloudledger.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS events (
    event_id     TEXT PRIMARY KEY,          -- 上游给的唯一 id,去重全靠它
    source       TEXT NOT NULL,             -- webhook / email / cron
    kind         TEXT NOT NULL,             -- ticket.created / digest.daily ...
    payload      TEXT NOT NULL,             -- 原始 JSON
    status       TEXT NOT NULL DEFAULT 'pending',  -- pending / processing / waiting_human / done / ignored / dead
    attempts     INTEGER NOT NULL DEFAULT 0,
    last_error   TEXT,
    received_at  TEXT NOT NULL,
    updated_at   TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    ticket_id   TEXT NOT NULL UNIQUE,       -- 一张工单只发一次回复,UNIQUE 挡住重复发送
    body        TEXT NOT NULL,
    sent_by     TEXT NOT NULL,
    sent_at     TEXT NOT NULL
);
"""


def now() -> str:
    return datetime.now().isoformat(timespec="seconds")


def connect(path: Path = EVENTS_DB) -> sqlite3.Connection:
    conn = sqlite3.connect(path, timeout=10)
    conn.row_factory = sqlite3.Row
    return conn


def init() -> None:
    with connect() as conn:
        conn.executescript(SCHEMA)


def add_event(event_id: str, source: str, kind: str, payload: str) -> bool:
    """返回 True 表示新事件,False 表示重复投递(已经收过)。"""
    with connect() as conn:
        cur = conn.execute(
            "INSERT OR IGNORE INTO events (event_id, source, kind, payload, received_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (event_id, source, kind, payload, now(), now()),
        )
        return cur.rowcount == 1


def set_status(event_id: str, status: str, error: str | None = None) -> None:
    with connect() as conn:
        conn.execute(
            "UPDATE events SET status = ?, last_error = ?, updated_at = ? WHERE event_id = ?",
            (status, error, now(), event_id),
        )

第二层:webhook 入口

webhook 入口只做三件事:验签、落库、回 202。不调模型,不查业务库。

签名方案我照着 Stripe 的格式写:上游用双方共享的密钥,对「时间戳 + 点 + 原始请求体」算 HMAC-SHA256,放在请求头 X-CloudLedger-Signature: t=1729580000,v1=<十六进制> 里。时间戳的作用是防重放:有人截获了一个合法请求,一小时后原样再发一次,签名是对的,但时间戳太旧,会被拒绝。

业界还有一个叫 Standard Webhooks 的规范,把这件事标准化成三个请求头 webhook-id、webhook-timestamp、webhook-signature,签名内容是 id.timestamp.payload。它的官网上列出的兼容实现里有 OpenAI、Anthropic、Twilio、PagerDuty 等。如果你在设计自己公司对外的 webhook,直接按它来,省得自己发明格式。

# agent-60-lab/topics/ambient-agents/webhook.py
"""接收「新工单」webhook:验签、去重、落库,立刻回 202。真正的处理交给 worker。

运行:uv run uvicorn webhook:app --port 8139
"""
import hashlib
import hmac
import json
import os
import time

from dotenv import load_dotenv
from fastapi import FastAPI, Header, HTTPException, Request

import store

load_dotenv()
SECRET = os.environ.get("WEBHOOK_SECRET", "dev-secret-change-me").encode()
TOLERANCE_S = 300

app = FastAPI()
store.init()


def verify(raw: bytes, header: str) -> None:
    try:
        parts = dict(p.split("=", 1) for p in header.split(","))
        ts, sig = int(parts["t"]), parts["v1"]
    except (KeyError, ValueError):
        raise HTTPException(400, "bad signature header")
    if abs(time.time() - ts) > TOLERANCE_S:
        raise HTTPException(400, "timestamp too old")
    expected = hmac.new(SECRET, f"{ts}.".encode() + raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig):
        raise HTTPException(401, "bad signature")


@app.post("/hooks/cloudledger", status_code=202)
async def receive(request: Request, x_cloudledger_signature: str = Header()):
    raw = await request.body()
    verify(raw, x_cloudledger_signature)
    event = json.loads(raw)
    if event["type"] != "ticket.created":
        return {"accepted": False, "reason": "ignored type"}
    is_new = store.add_event(event["id"], "webhook", event["type"], raw.decode())
    return {"accepted": True, "duplicate": not is_new}

再写一个模拟上游的小脚本,用来发测试请求。它故意支持「同一个事件发 N 次」,模拟上游超时重试:

# agent-60-lab/topics/ambient-agents/send_webhook.py
"""模拟上游系统发 webhook。用法:uv run python send_webhook.py T1012 [重复次数]"""
import hashlib
import hmac
import json
import os
import sys
import time

import httpx
from dotenv import load_dotenv

load_dotenv()
SECRET = os.environ.get("WEBHOOK_SECRET", "dev-secret-change-me").encode()

ticket_id = sys.argv[1]
times = int(sys.argv[2]) if len(sys.argv) > 2 else 1
body = json.dumps({"id": f"evt_{ticket_id}_created", "type": "ticket.created", "data": {"ticket_id": ticket_id}}).encode()

for _ in range(times):  # 上游超时重试时,同一个事件会原样再发一次
    ts = int(time.time())
    sig = hmac.new(SECRET, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    r = httpx.post(
        "http://127.0.0.1:8139/hooks/cloudledger",
        content=body,
        headers={"content-type": "application/json", "x-cloudledger-signature": f"t={ts},v1={sig}"},
    )
    print(r.status_code, r.json())

开两个终端。第一个起服务:

cd agent-60-lab/topics/ambient-agents
uv run uvicorn webhook:app --port 8139

第二个发四张工单,其中 T1012 连发两次,T9999 是一个业务库里不存在的工单(故意的,后面用来演示死信):

uv run python send_webhook.py T1012 2
uv run python send_webhook.py T1041
uv run python send_webhook.py T1003
uv run python send_webhook.py T9999
sqlite3 ambient.db "select event_id, status, attempts from events"

我实跑的输出:

202 {'accepted': True, 'duplicate': False}
202 {'accepted': True, 'duplicate': True}
202 {'accepted': True, 'duplicate': False}
202 {'accepted': True, 'duplicate': False}
202 {'accepted': True, 'duplicate': False}
evt_T1012_created|pending|0
evt_T1041_created|pending|0
evt_T1003_created|pending|0
evt_T9999_created|pending|0

webhook 入口对 T9999 也回了 202。它不知道也不应该知道这张工单存不存在,查业务库是 worker 的事。入口越笨越好:它是整个系统里唯一必须一直在线、必须快速响应的部分。

第三层:处理一张工单的图

图和 Day 25 的退款审批是同一个套路,换了个场景:读工单 → 模型起草回复 → 停下来等人 → 按人的决定发送、重写或者结束。

收件箱条目的格式,我参考了 LangChain 开源的 Agent Inbox 定义的 HumanInterrupt 结构:每条待办包含 action_request(Agent 想做什么动作、参数是什么)、config(允许人做哪几种处理)、description(给人看的一句话说明)。人的回复是 HumanResponse,type 取 accept、edit、response、ignore 四种之一。好处是格式通用:你今天写的命令行收件箱,将来换成网页,数据结构不用改。

需要说明一下现状:2026 年 10 月我去看的时候,langchain-ai/agent-inbox 这个仓库已经是只读的归档状态,LangChain 官方现在更推荐的是 LangChain 1.x 里的 HumanInTheLoopMiddleware,它给工具调用加审批,决定类型叫 approve、edit、reject、respond。名字不同,意思差不多。我们这里没有用 create_agent,是自己画的图,所以借用 Agent Inbox 那套结构自己实现。

# agent-60-lab/topics/ambient-agents/agent.py
"""后台处理一张工单:读工单 → 起草回复 → 停下来进收件箱等人 → 按人的处理结果发送、重写或忽略。"""
import sqlite3
from typing import Literal

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from pydantic import BaseModel, Field
from typing_extensions import TypedDict

import store
from llm import get_llm

CKPT_DB = store.HERE / "checkpoints.db"


class Draft(BaseModel):
    category: Literal["refund", "billing", "invoice", "bug", "account", "other"]  # 和 Day 28 同一套类别
    summary: str = Field(description="一句话概括客户的问题")
    reply: str = Field(description="给客户的回复草稿,中文,不超过 150 字,不承诺具体退款金额")


class TicketState(TypedDict, total=False):
    ticket_id: str
    ticket_text: str
    draft: dict
    feedback: list[str]
    outcome: str


def load(state: TicketState) -> dict:
    with store.connect(store.BIZ_DB) as conn:
        t = conn.execute("SELECT subject, body FROM tickets WHERE id = ?", (state["ticket_id"],)).fetchone()
    return {"ticket_text": f"{t['subject']}:{t['body']}", "feedback": []}


def draft(state: TicketState) -> dict:
    llm = get_llm().with_structured_output(Draft, method="function_calling")
    notes = "\n".join(f"- {f}" for f in state.get("feedback", []))
    prompt = (
        "你是云簿 CloudLedger 的客服。给下面的工单分类并起草回复。\n\n"
        f"工单 {state['ticket_id']}:{state['ticket_text']}\n"
    )
    if notes:
        prompt += f"\n客服主管对上一版草稿的意见,必须照改:\n{notes}\n"
    return {"draft": llm.invoke(prompt).model_dump()}


def review(state: TicketState) -> dict:
    d = state["draft"]
    # 收件箱条目:要做什么动作(action_request)、允许人怎么处理(config)、给人看的说明(description)
    answer = interrupt(
        {
            "action_request": {"action": "send_reply", "args": {"ticket_id": state["ticket_id"], "reply": d["reply"]}},
            "config": {"allow_accept": True, "allow_edit": True, "allow_respond": True, "allow_ignore": True},
            "description": f"[{d['category']}] {d['summary']}",
        }
    )
    kind = answer["type"]
    if kind == "accept":
        return {"outcome": "send", "draft": d}
    if kind == "edit":
        return {"outcome": "send", "draft": {**d, "reply": answer["args"]["args"]["reply"]}}
    if kind == "response":
        return {"outcome": "redraft", "feedback": state.get("feedback", []) + [answer["args"]]}
    return {"outcome": "ignored"}


def after_review(state: TicketState) -> str:
    return {"send": "send", "redraft": "draft"}.get(state["outcome"], END)


def send(state: TicketState) -> dict:
    with store.connect() as conn:
        cur = conn.execute(
            "INSERT OR IGNORE INTO outbox (ticket_id, body, sent_by, sent_at) VALUES (?, ?, 'agent+reviewer', ?)",
            (state["ticket_id"], state["draft"]["reply"], store.now()),
        )
    return {"outcome": "sent" if cur.rowcount == 1 else "already_sent"}


builder = StateGraph(TicketState)
builder.add_node("load", load)
builder.add_node("draft", draft)
builder.add_node("review", review)
builder.add_node("send", send)
builder.add_edge(START, "load")
builder.add_edge("load", "draft")
builder.add_edge("draft", "review")
builder.add_conditional_edges("review", after_review, ["send", "draft", END])
builder.add_edge("send", END)


def open_graph():
    """返回 (graph, conn)。调用方用完要 conn.close()。"""
    conn = sqlite3.connect(CKPT_DB, check_same_thread=False)
    return builder.compile(checkpointer=SqliteSaver(conn)), conn


def thread_of(ticket_id: str) -> dict:
    return {"configurable": {"thread_id": f"ticket-{ticket_id}"}}

第四层:worker

worker 是一个死循环:从事件表里领一个 pending 事件,跑图,根据结果改状态。看着简单,这里集中了今天大部分的坑。

# agent-60-lab/topics/ambient-agents/worker.py
"""从 events 表领事件、跑 Agent。用法:uv run python worker.py [--once]"""
import json
import sys
import time
from datetime import datetime, timedelta

import store
from agent import open_graph, thread_of

MAX_ATTEMPTS = 3
STALE_MINUTES = 10


def claim() -> dict | None:
    """原子地领一个 pending 事件。两个 worker 同时跑也不会领到同一个。"""
    with store.connect() as conn:
        row = conn.execute(
            "UPDATE events SET status = 'processing', attempts = attempts + 1, updated_at = ? "
            "WHERE event_id = (SELECT event_id FROM events WHERE status = 'pending' ORDER BY received_at LIMIT 1) "
            "RETURNING event_id, kind, payload, attempts",
            (store.now(),),
        ).fetchone()
    return dict(row) if row else None


def requeue_stale() -> int:
    """worker 进程中途被杀,事件会一直卡在 processing。超过 10 分钟没动静的放回队列。"""
    # 截止时间也在 Python 里算,格式和 store.now() 一致,字符串比较才成立
    cutoff = (datetime.now() - timedelta(minutes=STALE_MINUTES)).isoformat(timespec="seconds")
    with store.connect() as conn:
        cur = conn.execute(
            "UPDATE events SET status = 'pending', updated_at = ? WHERE status = 'processing' AND updated_at < ?",
            (store.now(), cutoff),
        )
    return cur.rowcount


def handle(graph, ev: dict) -> str:
    ticket_id = json.loads(ev["payload"])["data"]["ticket_id"]
    config = thread_of(ticket_id)
    snap = graph.get_state(config)
    if snap.interrupts or snap.values.get("outcome") in ("sent", "already_sent", "ignored"):
        # 这张工单已经在收件箱里,或者已经处理完了(上游换了个 event_id 又发了一次)
        return "done"
    if snap.next:
        # 上次跑到一半抛了异常,checkpoint 停在失败的节点前面:不给新输入,从那里接着跑
        result = graph.invoke(None, config)
    else:
        result = graph.invoke({"ticket_id": ticket_id}, config)
    return "waiting_human" if "__interrupt__" in result else "done"


def main() -> None:
    once = "--once" in sys.argv
    store.init()
    graph, conn = open_graph()
    try:
        while True:
            if n := requeue_stale():
                print(f"放回队列 {n} 个卡住的事件")
            ev = claim()
            if ev is None:
                if once:
                    break
                time.sleep(2)
                continue
            try:
                status = handle(graph, ev)
                store.set_status(ev["event_id"], status)
                print(ev["event_id"], "→", status)
            except Exception as e:  # noqa: BLE001
                status = "dead" if ev["attempts"] >= MAX_ATTEMPTS else "pending"
                store.set_status(ev["event_id"], status, f"{type(e).__name__}: {e}")
                print(ev["event_id"], f"第 {ev['attempts']} 次失败 → {status}:", e)
    finally:
        conn.close()


if __name__ == "__main__":
    main()

跑一次:

uv run python worker.py --once

evt_T1012_created → waiting_human
evt_T1041_created → waiting_human
evt_T1003_created → waiting_human
evt_T9999_created 第 1 次失败 → pending: 'NoneType' object is not subscriptable
evt_T9999_created 第 2 次失败 → pending: 'NoneType' object is not subscriptable
evt_T9999_created 第 3 次失败 → dead: 'NoneType' object is not subscriptable

三张正常的工单各调用一次模型,一共 8 秒左右。T9999 三次失败后进了死信,不会无限循环。

第五层:收件箱

收件箱就是「列出所有 waiting_human 的事件,对其中一条给出处理结果」。今天用命令行做,原理和网页版一样:网页版只是把 list 换成一个列表页,把四个子命令换成四个按钮。

# agent-60-lab/topics/ambient-agents/inbox.py
"""命令行收件箱。
  uv run python inbox.py list
  uv run python inbox.py accept  T1012
  uv run python inbox.py edit    T1012 "改过的回复全文"
  uv run python inbox.py respond T1012 "语气再软一点,先道歉"
  uv run python inbox.py ignore  T1012
"""
import json
import sys

from langgraph.types import Command

import store
from agent import open_graph, thread_of


def waiting() -> list[tuple[str, str]]:
    with store.connect() as conn:
        rows = conn.execute(
            "SELECT event_id, payload FROM events WHERE status = 'waiting_human' ORDER BY received_at"
        ).fetchall()
    return [(r["event_id"], json.loads(r["payload"])["data"]["ticket_id"]) for r in rows]


def main() -> None:
    cmd = sys.argv[1]
    graph, conn = open_graph()
    try:
        if cmd == "list":
            for event_id, ticket_id in waiting():
                item = graph.get_state(thread_of(ticket_id)).interrupts[0].value
                print(f"{ticket_id}  {item['description']}")
                print(f"        草稿:{item['action_request']['args']['reply']}")
            return

        ticket_id = sys.argv[2]
        text = sys.argv[3] if len(sys.argv) > 3 else ""
        event_id = next((e for e, t in waiting() if t == ticket_id), None)
        if event_id is None:
            print(ticket_id, "不在收件箱里(已经处理过,或者还没轮到 worker)")
            return
        if cmd == "accept":
            answer = {"type": "accept", "args": None}
        elif cmd == "edit":
            answer = {"type": "edit", "args": {"action": "send_reply", "args": {"ticket_id": ticket_id, "reply": text}}}
        elif cmd == "respond":
            answer = {"type": "response", "args": text}
        else:
            answer = {"type": "ignore", "args": None}

        result = graph.invoke(Command(resume=answer), thread_of(ticket_id))
        if "__interrupt__" in result:
            print("已按意见重写,新草稿:", result["__interrupt__"][0].value["action_request"]["args"]["reply"])
            return
        store.set_status(event_id, "ignored" if result["outcome"] == "ignored" else "done")
        print(ticket_id, "→", result["outcome"])
    finally:
        conn.close()


if __name__ == "__main__":
    main()

把四种处理都走一遍:

uv run python inbox.py list
uv run python inbox.py accept T1012
uv run python inbox.py respond T1041 "这张已确认是重复扣款,直接告诉客户退款申请已提交、原路退回,别写金额"
uv run python inbox.py edit T1041 "您好,已确认 10 月 8 日的 INV-2026-0123 是重复扣款,退款申请已提交,会原路退回,到账时间以支付渠道为准。给您添麻烦了。"
uv run python inbox.py ignore T1003
sqlite3 ambient.db "select event_id, status, attempts from events"

最后一条查询的实跑结果:

evt_T1012_created|done|1
evt_T1041_created|done|1
evt_T1003_created|ignored|1
evt_T9999_created|dead|3

outbox 里有两行:T1012 是模型原稿,T1041 是人改过的版本。T1041 先 respond 再 edit,说明一条待办可以来回多轮,直到人满意为止。

有一个细节值得停下来想:T1041 的模型重写版其实已经可以用了,我为什么还要 edit?因为我想在回复里写明具体是哪一张发票(INV-2026-0123)。这种修改每天都会发生,它们是最有价值的数据:人改了什么、为什么改,下一版提示词或者下一次微调就从这里来。把 outbox 里「模型原稿」和「最终发出版」都存下来,是练习 4 的内容。

邮件和定时任务也走同一条队列

webhook 跑通以后,另外两种触发源只需要做一件事:把自己的输入变成事件表里的一行。

先看邮件。下面这个脚本每次运行收一遍客服邮箱的未读邮件,能对上已知客户的,在业务库里建一张工单,再写一个事件:

# agent-60-lab/topics/ambient-agents/email_poller.py
"""每分钟收一次客服邮箱,把新邮件变成工单 + 事件。
环境变量:IMAP_HOST、IMAP_USER、IMAP_PASSWORD(邮箱的授权码,不是登录密码)
crontab:* * * * * cd ~/agent-60-lab/topics/ambient-agents && uv run python email_poller.py
"""
import email
import hashlib
import imaplib
import json
import os
from email.header import decode_header, make_header
from email.message import Message
from email.utils import parseaddr

from dotenv import load_dotenv

import store

load_dotenv()


def text_of(msg: Message) -> str:
    part = next((p for p in msg.walk() if p.get_content_type() == "text/plain"), None)
    if part is None:
        return ""
    raw = part.get_payload(decode=True) or b""
    return raw.decode(part.get_content_charset() or "utf-8", errors="replace").strip()


def parse(raw: bytes) -> dict:
    msg = email.message_from_bytes(raw)
    return {
        "message_id": msg["Message-ID"].strip(),
        "from": parseaddr(msg["From"])[1].lower(),
        "subject": str(make_header(decode_header(msg["Subject"] or ""))),
        "body": text_of(msg)[:2000],
    }


def to_event(mail: dict) -> str:
    with store.connect(store.BIZ_DB) as biz:
        c = biz.execute("SELECT id FROM customers WHERE email = ?", (mail["from"],)).fetchone()
        if c is None:
            return "skip: 不是已知客户"
        ticket_id = "M" + hashlib.sha1(mail["message_id"].encode()).hexdigest()[:6]
        biz.execute(
            "INSERT OR IGNORE INTO tickets VALUES (?, ?, ?, ?, 'open', ?)",
            (ticket_id, c["id"], mail["subject"], mail["body"], store.now()[:10]),
        )
    payload = json.dumps({"id": mail["message_id"], "type": "ticket.created", "data": {"ticket_id": ticket_id}})
    return ticket_id if store.add_event(mail["message_id"], "email", "ticket.created", payload) else "duplicate"


def main() -> None:
    store.init()
    with imaplib.IMAP4_SSL(os.environ["IMAP_HOST"]) as imap:
        imap.login(os.environ["IMAP_USER"], os.environ["IMAP_PASSWORD"])
        imap.select("INBOX")
        _, data = imap.search(None, "UNSEEN")
        for num in data[0].split():
            _, parts = imap.fetch(num, "(RFC822)")  # fetch 会顺手把邮件标成已读
            print(to_event(parse(parts[0][1])))


if __name__ == "__main__":
    main()

再看定时任务。两个子命令:scan 把超过 20 天还开着、也没回复过的工单当成事件塞进队列;report 出一份日报,有 Slack webhook 地址就顺便发过去。

# agent-60-lab/topics/ambient-agents/cron_jobs.py
"""两个定时任务,交给 crontab 每天跑:
  uv run python cron_jobs.py scan     # 把超过 N 天没回复的工单当成事件塞进队列
  uv run python cron_jobs.py report   # 收件箱和死信的日报,有 SLACK_WEBHOOK_URL 就发过去
"""
import json
import os
import sys
from datetime import date, datetime, timedelta

import httpx
from dotenv import load_dotenv

import store

load_dotenv()
STALE_DAYS = 20


def scan(limit: int = 5) -> None:
    cutoff = (date.today() - timedelta(days=STALE_DAYS)).isoformat()
    with store.connect(store.BIZ_DB) as biz:
        rows = biz.execute(
            "SELECT id FROM tickets WHERE status = 'open' AND created_at <= ? ORDER BY created_at, id",
            (cutoff,),
        ).fetchall()
    with store.connect() as conn:
        sent = {r["ticket_id"] for r in conn.execute("SELECT ticket_id FROM outbox")}
    todo = [r["id"] for r in rows if r["id"] not in sent]
    new = 0
    for ticket_id in todo[:limit]:
        # event_id 由「任务名 + 工单 + 日期」拼出来:同一天重复跑 scan,不会重复入队
        event_id = f"cron:stale:{ticket_id}:{date.today().isoformat()}"
        payload = json.dumps({"id": event_id, "type": "ticket.stale", "data": {"ticket_id": ticket_id}})
        new += store.add_event(event_id, "cron", "ticket.stale", payload)
    print(f"超过 {STALE_DAYS} 天还开着的工单 {len(rows)} 张,其中没回复过的 {len(todo)} 张,本次入队 {new} 个")


def report() -> None:
    with store.connect() as conn:
        counts = dict(conn.execute("SELECT status, COUNT(*) FROM events GROUP BY status").fetchall())
        oldest = conn.execute(
            "SELECT event_id, updated_at FROM events WHERE status = 'waiting_human' ORDER BY updated_at LIMIT 1"
        ).fetchone()
        dead = conn.execute("SELECT event_id, last_error FROM events WHERE status = 'dead'").fetchall()
    lines = [f"事件状态:{counts}"]
    if oldest:
        hours = (datetime.now() - datetime.fromisoformat(oldest["updated_at"])).total_seconds() / 3600
        lines.append(f"收件箱里等得最久的:{oldest['event_id']},已等 {hours:.1f} 小时")
    for d in dead:
        lines.append(f"死信:{d['event_id']}  {d['last_error'][:80]}")
    text = "\n".join(lines)
    print(text)
    if url := os.environ.get("SLACK_WEBHOOK_URL"):
        httpx.post(url, json={"text": text}, timeout=10)


if __name__ == "__main__":
    {"scan": scan, "report": report}[sys.argv[1]]()

crontab 这样配(crontab -e 打开编辑,路径换成你自己的):

# 每分钟收一次邮件、处理一批事件
* * * * * cd ~/agent-60-lab/topics/ambient-agents && ~/.local/bin/uv run python email_poller.py >> cron.log 2>&1
* * * * * cd ~/agent-60-lab/topics/ambient-agents && ~/.local/bin/uv run python worker.py --once >> cron.log 2>&1
# 每天 9:00 扫描久未回复的工单,9:05 出日报
0 9 * * * cd ~/agent-60-lab/topics/ambient-agents && ~/.local/bin/uv run python cron_jobs.py scan >> cron.log 2>&1
5 9 * * * cd ~/agent-60-lab/topics/ambient-agents && ~/.local/bin/uv run python cron_jobs.py report >> cron.log 2>&1

crontab 里的环境变量和你的终端不一样,uv 很可能不在 PATH 里,所以写绝对路径(which uv 查)。每条都把输出追加到日志文件,否则出错了你什么都看不到。worker.py --once 每分钟跑一次时,如果上一次还没跑完,下一次又启动了,两个 worker 会同时工作;好在 claim() 是原子的,它们不会领到同一个事件。

用 crontab.guru 可以检查表达式。如果你用的是 LangGraph 的托管部署(现在叫 LangSmith Deployment),它自带 cron 功能,client.crons.create(...) 一行就能建,但官方文档特别提醒所有时间都按 UTC 解释,北京时间早上 9 点要写成 0 1 * * *。

上线以后盯什么

后台 Agent 最大的风险是安静地坏掉:没有人在对面等结果,所以也没有人会立刻发现它不工作了。我会盯这几个数:

• 队列深度:pending 的事件数。持续上涨说明 worker 处理不过来,或者已经挂了。
• 收件箱最长等待时间:上面日报里那一行。超过一个工作日就该提醒值班的人。
• 死信数:任何一条死信都值得看。今天的 T9999 是工单不存在,真实环境里更常见的是模型服务欠费、API key 过期、上游改了 payload 格式。
• 每类工单的处理结果分布:accept、edit、response、ignore 各占多少。某一类的 accept 率很高,说明可以考虑让它自动发;某一类 edit 率突然升高,说明模型或提示词出了问题,或者业务规则变了。
• 单事件成本:按上面的算法,一次起草约 0.0015 元。response 会多调一次模型,如果某类工单平均要 respond 两三轮,成本和人的耐心都会出问题。

这些数从 events 表、outbox 表和 Day 34 的 Langfuse trace 里都能算出来。Day 34 讲 tracing 时 trace 是按「一次对话」组织的,后台 Agent 要改成按「一个事件」组织,trace 的 session id 用工单号,这样一张工单来回几轮的所有调用能串在一起看。

动手练习

1. 把全流程跑通。 按正文顺序起 webhook 服务,发 T1012(两次)、T1041、T1003、T9999,跑 worker,用收件箱把四种处理都走一遍。验收:events 表最终状态和正文的四行结果一致(你的模型草稿内容会不同,状态应该一样);outbox 里恰好两行。

2. 把「工单不存在」改成正确的失败方式。 现在 T9999 要重试三次才进死信,但「工单不存在」重试一百次也不会变好。修改 load 节点:查不到时抛一个自定义的 PermanentError;修改 worker:捕获到 PermanentError 直接标 dead,不重试。验收:T9999 只失败一次就进死信,attempts 是 1,last_error 里能看出是「工单不存在」。

3. 加退避。 给 events 表加一列 not_before TEXT,失败时设成「现在 + 2 的 attempts 次方 分钟」,claim() 只领 not_before 为空或已经过去的事件。用 LLM_API_KEY=sk-wrong 制造失败来测。验收:同一个失败事件的三次尝试之间,间隔分别约 2 分钟和 4 分钟,而不是一口气连试三次。

4. 把人的修改存下来。 在 send 节点里,除了写 outbox,再往一张新表 review_log 里写一行:工单号、类别、模型原稿、最终发出的文本、处理类型(accept 还是 edit)、所有 feedback。验收:跑完正文流程,review_log 里 T1012 的处理类型是 accept、原稿和终稿相同;T1041 是 edit、原稿和终稿不同、feedback 里有一条意见。这张表是 Day 144 讲反馈闭环时的原料。

5. 加一种「通知」类待办。 修改 draft 节点:如果工单正文里出现「律师」「起诉」「投诉到」这类词,不起草回复,而是产生一条 action 为 notify_supervisor 的收件箱条目,config 里只允许 ignore(看过即可)。用邮件离线测试的方法造一封含「我要投诉到消费者协会」的邮件。验收:这封邮件进收件箱后显示的说明里标明了是通知,accept 和 edit 会被拒绝。

6. (选做)换成 Postgres。 把事件表搬到 Day 17 的 Postgres 里,claim() 改用 FOR UPDATE SKIP LOCKED,checkpointer 换成 langgraph-checkpoint-postgres。同时开三个 worker 处理 20 个事件。验收:每个事件只被处理一次,三个 worker 的输出里没有重复的 event_id。

名词解释

名词	意思
Ambient agent(后台 Agent)	由外部事件(webhook、邮件、定时任务)触发、在后台运行、只在需要时找人的 Agent。LangChain 2025 年 1 月提出的叫法
Agent Inbox	后台 Agent 找人的集中入口。LangChain 开源过一个同名的界面,定义了 HumanInterrupt 和 HumanResponse 两个数据结构,仓库现已归档
Webhook	对方系统在某件事发生时,主动向你预先登记的 URL 发一个 HTTP 请求
HMAC 签名	用双方共享的密钥对请求内容算出的摘要。接收方重算一遍对得上,才说明请求确实来自持有密钥的一方且内容没被改
重放攻击	截获一个合法请求后原样再发。签名挡不住,要靠时间戳容忍窗口和事件 id 去重
至少一次投递(at-least-once)	大多数 webhook 和消息队列的保证:不会丢,但可能重复。接收方必须自己去重
幂等	同一个操作做一次和做多次结果一样。今天靠事件 id 主键、thread_id 用业务主键、outbox 的 UNIQUE 三处实现
死信(dead letter)	重试够次数仍然失败、不再自动处理、等人排查的任务
原子领取	「挑一个任务」和「标记为我在处理」在一个不可分割的操作里完成,防止两个 worker 领到同一个
轮询(polling)	定期主动去问「有没有新东西」,今天的邮件收取和 worker 都是
crontab	Unix 系统自带的定时任务表,五个字段依次是分、时、日、月、星期

常见误区

在 webhook 处理函数里直接跑 Agent。 模型调用动辄几秒到几十秒,上游等不到 2xx 就会重试,你会同时跑好几份同一个任务。入口只做验签和落库,处理交给后台。

拿解析后的 JSON 去验签。 签名算的是原始字节。json.loads 再 json.dumps 之后,空格、键顺序、中文转义都可能变,签名就对不上了,而且这种 bug 只在特定 payload 上出现,很难查。

只靠事件 id 去重。 同一张工单可能被不同的上游、以不同的事件 id 通知好几次。还要在业务层面去重:thread_id 用工单号,发送动作本身要幂等。

把「中断」和「崩溃」当成一回事。 两者在 checkpoint 里都表现为 next 不为空。等人的 thread 要跳过,崩溃的 thread 要续跑,区分它们要看 interrupts。

用格式不同的时间字符串比大小。 SQLite 没有真正的时间类型,时间都是字符串。2026-10-22T10:00:00 和 2026-10-22 10:00:00 比较的结果和直觉不一样。全库统一一种格式,或者统一存 Unix 时间戳。

所有失败都重试。 网络抖动、限流值得重试;工单不存在、参数格式错这类永久性错误,重试只会浪费钱和时间。先分清是哪一种。

收件箱只看条数。 十条刚进来的不要紧,一条等了三天的才是问题。告警要盯最长等待时间。

学习资料

文档与文章

• LangChain · Introducing ambient agents:提出这个概念的原文,通知、提问、审核三种找人方式就出自这里,15 分钟。
• langchain-ai/agent-inbox:README 里的 HumanInterrupt 和 HumanResponse 定义是今天收件箱格式的来源。仓库已归档,只看数据结构那一节,10 分钟。
• LangChain · Human-in-the-loop:HumanInTheLoopMiddleware 的官方文档,如果你用 create_agent 而不是自己画图,审批走这一套,20 分钟。
• LangGraph · Interrupts:interrupt 的完整语义,特别是「恢复时节点重新执行」那一段,20 分钟。
• LangSmith · Cron jobs:托管部署上的定时任务,注意 UTC,10 分钟。
• Stripe · Webhooks 与 签名校验:业界写得最好的 webhook 接收指南,快速返回、去重、验签、防重放全都讲了,30 分钟。
• GitHub · Validating webhook deliveries:另一种签名头格式,对比着看,10 分钟。
• Standard Webhooks 与 规范原文:如果你要设计自己的 webhook,照着它来,20 分钟。
• Brandur · Implementing Stripe-like Idempotency Keys in Postgres:幂等键的经典长文,做练习 6 之前读,40 分钟。
• SQLite · RETURNING:claim() 用到的语法,5 分钟。
• Python · imaplib:邮件收取的标准库文档,查 search 和 fetch 的参数。
• Cloudflare · Email Workers 和 Postmark · Inbound webhook:把收到的邮件转成 webhook 的两种现成服务,邮件量大时考虑。

B 站视频

• 是的iamrio《03.让Agent可审批、可恢复、可重放,LangGraph进阶》(B 站 BV1Yt376TEUC,12 分钟):2026 年 7 月的视频,审批、恢复、重放三件事放在一起讲,和今天的 worker 续跑那一节对应。

其他经典视频

• ByteByteGo《Top 3 Things You Should Know About Webhooks!》(YouTube,按标题搜索):几分钟的动画,讲 webhook 的重试、幂等和安全,和今天的第二层对应。
• LangChain Academy《Project: Ambient Agents with LangGraph》(课程页):免费,28 节约 2.5 小时,做一个邮件助手,包含评测、人工审核、记忆和部署。想看完整版就学这门。

书籍

• Martin Kleppmann、Chris Riccomini《Designing Data-Intensive Applications》第二版(O’Reilly, 2026)第 13 章「A Philosophy of Streaming Systems」:事件日志、至少一次投递、幂等消费这些概念的源头。第一版(2017,中译本《数据密集型应用系统设计》)对应的是第 11 章「流处理」。
• Gregor Hohpe、Bobby Woolf《Enterprise Integration Patterns》(Addison-Wesley, 2003):老书,但「Idempotent Receiver」「Dead Letter Channel」「Polling Consumer」这几个模式的名字就出自这里,今天的 worker 几乎就是它们的组合。翻目录挑这三节看。
• Chip Huyen《AI Engineering》(O’Reilly, 2025)第 6 章「Agents」:讲 Agent 的失败模式和评测,补充「为什么要让人先审」的理由。

开源项目与代码

• langchain-ai/agents-from-scratch:LangChain Academy 那门课的配套代码,一个完整的邮件助手 ambient agent,看它的 HITL 那一步怎么写。
• langchain-ai/agent-inbox:收件箱前端的参考实现(已归档,只读)。
• standard-webhooks/standard-webhooks:各种语言的签名和校验库,自己写验签之前先看看有没有现成的。

自测

1. 为什么 webhook 入口要先把事件写进数据库再返回 202,而不是收到就直接调 Agent 处理?至少说出两个理由。

答案

第一,模型调用慢,上游等不到 2xx 会超时重试,结果同一个任务被跑好几次。第二,入口进程挂了或者重启时,正在处理的任务会丢;先落库就有了持久化的记录,worker 重启后接着处理。第三,去重、重试、死信、监控都只需要在 worker 一处实现,三种触发源共用。

2. 签名校验已经能证明请求来自持有密钥的一方,为什么还要检查时间戳?

答案

签名证明的是「这个请求是合法方发的、内容没被改」,但挡不住重放:攻击者截获一个合法请求后原样再发,签名依然正确。检查时间戳在容忍窗口(比如 5 分钟)内,可以让旧请求失效;窗口之内的重放,再靠事件 id 去重挡掉。

3. worker 里 if snap.next: 那一段解决了什么问题?如果只判断 snap.next 不为空就跳过,会发生什么?

答案

它让跑到一半崩掉的工单从失败的节点接着跑,而不是被当成「已经在处理中」跳过。等人的 thread 和崩溃的 thread 在 checkpoint 里 next 都不为空,区别是前者 interrupts 里有待处理的中断。只看 next 的话,第一次失败的事件在重试时会被直接标成 done,这张工单就再也不会被处理了,正文里 T9999 就是这样被我第一版代码漏掉的。

4. 定时任务的事件 id 为什么用 cron:stale:T1023:2026-10-22 这种拼出来的值,而不是随机生成?

答案

拼出来的 id 是确定的:同一天无论 scan 跑几次(补跑、手动跑、两台机器都配了 crontab),同一张工单得到的都是同一个 id,主键冲突自动去重。随机 id 每次都不同,重复跑就会重复入队。带上日期是有意的:明天还没处理的工单应该再被提醒一次。

5. 面试题:你要给一家 SaaS 公司设计一个后台客服 Agent,新工单通过 webhook 进来,Agent 起草回复后由人审核再发出。请说出你会在哪几个地方处理「重复」,每个地方防的是什么。

答案

至少三处。一是事件层:事件 id 做主键,INSERT OR IGNORE,防上游重试造成的重复投递。二是业务层:执行状态按工单号组织(thread_id 用工单号),同一张工单被不同来源、不同事件 id 触发时,发现已经在收件箱或已处理就跳过,防多个上游重复通知。三是动作层:发送回复本身幂等(outbox 的唯一约束,或调用邮件服务时带幂等键),防 worker 在「已发送」和「已标记完成」之间崩溃后重跑造成重复发送。加分项:领任务要原子(UPDATE…RETURNING 或 FOR UPDATE SKIP LOCKED),防两个 worker 处理同一个事件;定时任务的事件 id 用确定性的拼接值。

6. 收件箱里有四种处理:accept、edit、response、ignore。哪一种会让 Agent 再调一次模型?如果某一类工单的 edit 比例连续一周从 10% 涨到 40%,你会先查什么?

答案

只有 response 会再调模型(带着意见重写)。edit 比例突然上涨,先看最近有没有改动:提示词、模型版本、.env 里的模型名、知识库内容;再看业务有没有变化:价格、退款政策、新功能上线后客户问的问题变了。然后抽二三十条 edit 记录,对比原稿和终稿,看人到底在改什么,这就是 Day 30 错误分析的做法。

明天预告

专题篇可以跳着读。下一篇是 Day 140 · 低代码 Agent 平台实操:Dify、扣子、百炼、n8n 现在各是什么状态,同一个工单分流流程放到平台上和写成代码各是什么样,平台搭的应用怎么跑你自己的评测集,以及什么时候该从平台改成写代码。