AI Agent 一旦从对话框走向后台运行，问题就不再只是“模型能否回答”，而是任务如何触发、状态如何追踪、失败后如何重试，以及高风险操作由谁批准。由一群在 AWS 工作的开发者开源的 Pizza Bot，尝试用自托管的收件箱界面解决这组工程问题：Agent 在后台执行任务，完成后把结果投递到收件箱；任务可以由计划或 Webhook 触发，也可以委派给专门的 Worker，并在必要时暂停等待人工审批。
收件箱不是聊天窗口，而是任务控制面
聊天界面通常假设用户在线：发送问题、等待响应、继续追问。后台 Agent 的生命周期则可能持续几分钟甚至数小时，发起任务的人也未必一直停留在页面上。
因此，收件箱式设计更接近一个任务控制面。每条记录至少需要包含：
- 触发来源：人工提交、定时计划或外部 Webhook。
- 当前状态：排队、执行中、等待审批、完成或失败。
- 执行角色：通用 Agent，或负责摘要、报告、发布等工作的专用 Worker。
- 输入与结果：原始参数、产物、错误信息以及执行时间。
- 审批轨迹：谁批准了什么操作，审批前后的参数是否发生变化。
这种设计的价值在于把“让 Agent 做事”和“确认 Agent 做完了什么”分离开。使用者不必守着一次长连接，运维人员也能从统一入口检查积压、失败和待审批任务。
触发、委派与人工审批如何组合
Pizza Bot 摘要中提到的几项能力可以组成一条典型流水线：
- 定时器或业务系统通过 Webhook 创建任务。
- 调度层根据任务类型，把工作委派给专门的 Worker。
- Worker 执行资料整理、生成报告或调用外部工具。
- 如果下一步涉及发布、删除或发送消息，任务进入待审批状态。
- 人工确认后继续执行，最终结果回到收件箱。
这里最重要的边界是：审批应该发生在副作用之前，而不是模型生成内容之后就算完成。例如，生成一封客户邮件可以自动进行，但真正发送邮件应当作为单独动作接受审批。这样既能保留 Agent 的自动化效率，也能限制错误结果的影响范围。
委派也不应该只靠一段提示词决定。生产系统通常需要显式的任务类型、允许调用的工具列表、超时和资源配额。专门 Worker 的权限越窄，越容易审计，也越不容易因为提示注入而越权。
可以这样实践：搭一个最小后台任务收件箱
下面的示例用于演示这类系统的核心数据流，不是 Pizza Bot 的官方 API 或配置。它提供任务提交、Webhook 触发、Worker 委派、危险动作审批和收件箱查询，可以直接运行后再替换成真实 Agent、数据库与队列。
将以下内容保存为 app.py：
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from threading import Lock
from time import sleep
from uuid import uuid4
from flask import Flask, jsonify, request
app = Flask(__name__)
executor = ThreadPoolExecutor(max_workers=4)
tasks = {}
lock = Lock()
def now():
return datetime.now(timezone.utc).isoformat()
def update(task_id, **changes):
with lock:
tasks[task_id].update(changes)
tasks[task_id]['updated_at'] = now()
def run_worker(task_id):
try:
update(task_id, status='running')
with lock:
task = dict(tasks[task_id])
# Replace these branches with real LLM, tool, or service calls.
sleep(2)
if task['kind'] == 'summary':
result = f"Summary worker processed: {task['payload']}"
elif task['kind'] == 'report':
result = f"Report worker processed: {task['payload']}"
else:
result = f"General worker processed: {task['payload']}"
update(task_id, status='completed', result=result)
except Exception as exc:
update(task_id, status='failed', error=str(exc))
def create_task(data):
task_id = str(uuid4())
action = data.get('action', 'analyze')
requires_approval = action in {'publish', 'delete', 'send'}
task = {
'id': task_id,
'kind': data.get('kind', 'general'),
'action': action,
'payload': data.get('payload', {}),
'status': 'waiting_approval' if requires_approval else 'queued',
'result': None,
'created_at': now(),
'updated_at': now(),
}
with lock:
tasks[task_id] = task
if not requires_approval:
executor.submit(run_worker, task_id)
return task
@app.post('/tasks')
@app.post('/webhooks/jobs')
def submit_task():
return jsonify(create_task(request.get_json(force=True))), 202
@app.post('/tasks/<task_id>/approve')
def approve_task(task_id):
with lock:
task = tasks.get(task_id)
if not task:
return jsonify({'error': 'task not found'}), 404
if task['status'] != 'waiting_approval':
return jsonify({'error': 'task is not waiting for approval'}), 409
task['status'] = 'queued'
task['approved_by'] = request.headers.get('X-User', 'anonymous')
task['updated_at'] = now()
executor.submit(run_worker, task_id)
return jsonify(task), 202
@app.get('/inbox')
def inbox():
status = request.args.get('status')
with lock:
items = [dict(task) for task in tasks.values()]
if status:
items = [task for task in items if task['status'] == status]
items.sort(key=lambda task: task['created_at'], reverse=True)
return jsonify(items)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)
安装依赖并启动：
python -m venv .venv
. .venv/bin/activate
pip install flask
python app.py
提交一个可以自动执行的摘要任务：
curl -s http://localhost:8080/tasks \
-H 'Content-Type: application/json' \
-d '{"kind":"summary","payload":{"document":"weekly-notes.md"}}'
sleep 3
curl -s http://localhost:8080/inbox
提交一个必须审批的发布任务。先记录响应中的 id，再执行审批：
curl -s http://localhost:8080/webhooks/jobs \
-H 'Content-Type: application/json' \
-d '{"kind":"report","action":"publish","payload":{"target":"status-page"}}'
curl -s 'http://localhost:8080/inbox?status=waiting_approval'
TASK_ID='替换为任务 ID'
curl -s -X POST "http://localhost:8080/tasks/${TASK_ID}/approve" \
-H 'X-User: reviewer@example.com'
如果要模拟计划任务，可以先用 cron 定期调用 Webhook：
0 8 * * 1-5 curl -fsS http://localhost:8080/webhooks/jobs -H 'Content-Type: application/json' -d '{"kind":"report","payload":{"range":"yesterday"}}'
这个示例只适合本地理解流程。进程重启会丢失内存中的任务，线程池也不具备持久化重试能力。正式部署时，应把 tasks 换成 PostgreSQL 等持久化存储，把线程池换成可靠队列，并为任务增加幂等键、重试次数和租约超时。
自托管之后，安全责任也留在自己手里
自托管让组织能够控制任务内容、模型凭据和审批数据，但同时意味着团队要自行承担认证、升级、备份和可观测性工作。评估 Pizza Bot 或同类系统时，可以重点检查以下项目：
- Webhook 是否验证签名、时间戳，并拒绝重复事件。
- Agent 凭据是否按 Worker 隔离，而不是共享一个高权限密钥。
- 审批人能否看到最终参数、目标系统和即将产生的副作用。
- 失败任务是否支持安全重试，重复执行会不会造成重复发送或重复扣费。
- 日志是否会泄露提示词、客户数据、访问令牌或模型输出中的敏感信息。
- 队列积压、任务耗时、失败率和待审批数量是否有指标与告警。
- 对外部内容的处理是否能抵御提示注入，并限制 Agent 可调用的工具。
适合从低风险、结果可复核的任务开始，例如日报汇总、文档分类或内部研究。等状态管理、权限隔离和审计链路稳定后，再逐步开放发送邮件、修改工单或发布内容等有副作用的动作。Pizza Bot 所代表的方向并不是让 Agent 完全无人值守，而是让异步执行、专业分工和人工控制出现在同一个可追踪的工作面上。