AI智能体人工确认闸门:AWS Pizza Bot的架构与部署
2026/9/15 4:24:50 网站建设 项目流程

不知道你有没有过这种经历:在折腾 AI 智能体工作流的时候,总会在某个环节卡住——“这个权限到底要不要给智能体?”“给了怕它闯祸,不给流程又跑不通”。我最近在做一套自动化的订餐流程,让智能体帮我去比价、下单,结果头一回实测它就差点把一个备注写错的订单直接提交出去。幸好我在中间加了一道“人工确认闸门”,才没出事故。这个闸门,就是 AWS 开源的 Pizza Bot:一个面向后台 AI 智能体的自托管收件箱应用。

说白了,Pizza Bot 不是帮你做披萨的机器人,而是给智能体用的“待办确认箱”。它的思路很直白:智能体遇到需要人拍板的事情,不要自作主张去执行,而是把这件事打包成一条消息,投递到一个收件箱里。人在手机上、邮件里看到这条消息,点一下“确认”或“拒绝”,智能体收到回调之后再继续往下跑。整个过程是异步的、有留痕的、完全可控的。

这个项目对于三类人特别有价值:一是正在做 AI Agent 应用、需要处理“人工介入点”的开发者;二是做自动化运维或内部工具,想给机器人任务加一层审批机制的团队;三是单纯想研究 AWS 无服务器架构怎么组织事件驱动服务的人。下面我会从项目设计思路、架构拆解、本地部署到接入智能体,完整走一遍,最后聊几个我实测踩过的坑。

1. 为什么 AI 智能体需要一个“收件箱”

1.1 智能体自动化里的“失控焦虑”

做 AI 智能体的人应该都有类似的体感:智能体的“能力”越强,你越不敢放手。让一个 Agent 去查询天气、整理文档,这没问题;让它去下单、发邮件、调接口,你就得反复掂量——万一它理解错了用户意图怎么办?万一它的 API 密钥被滥用怎么办?万一它在一个错误的上下文里执行了危险操作怎么办?

传统做法里,最常见的“人机协作”方式是实时对话:智能体在 IM 工具里问用户“确认下单吗?”,用户回一句“确认”。这种方式的问题在于,它要求智能体和用户保持在线的、同步的通信状态。如果智能体是后台任务,跑在无人值守的服务器上,用户也没有一直盯着聊天窗口,那么这条确认消息可能迟迟没人回,智能体就只能干等,甚至超时误判。

还有一种是直接在代码里写死“自动执行”,完全不给人介入的机会。这种方式在初期很爽,一旦出问题就是大事故。我见过有人把支付接口的密钥放在 Agent 的环境变量里,Agent 在一次测试中因为 prompt 被污染,差点真扣了一笔钱。从那以后我就坚信:任何涉及真实业务动作的 Agent 工作流,中间必须有一道“人确认”的闸门。

Pizza Bot 解决的问题,就是把这个“人工确认”的过程做成一个标准化的基础设施组件。它不像聊天窗口那样要求双方同时在线,而是把确认请求变成一条持久化的消息,投递到收件箱,用户在任何时间、任何设备上看到再处理。智能体也不需要阻塞等待,它可以先去忙别的,等回调来了再继续。

介入方式实时性要求对智能体主流程的侵入可审计性场景适配
实时聊天确认高,双方必须在线高,需要挂起流程等待回复弱,消息容易淹没简单交互,单次确认
代码内自动执行无,但有风险弱,动作不可控只适合零风险操作
收件箱异步确认低,天然支持离线低,投递后即可干别的强,全程留痕支付、下单、发布等敏感操作

1.2 “收件箱模式”解决的三个关键问题

Pizza Bot 把人工确认抽离成“投递消息 -> 用户确认 -> 回调结果”,这个模式本质上解决了三个问题:

第一是权限收敛。智能体不需要真的拥有“下订单”这个高阶权限,它只需要有权限调用 Pizza Bot 的 API 来创建一条待确认消息。就算 Agent 的 API 密钥被泄露,攻击者最多就是往你的收件箱里塞几条垃圾确认请求,无法直接操作用户真实业务系统。这其实应用的是最小权限原则:给智能体的权限,永远只够它“申请执行”,而不是“直接执行”。

第二是异步解耦。Agent 投递完消息就可以立即返回,继续处理其他任务。用户确认这个动作发生在未来的某个时间点,两者之间通过消息 ID 关联,不需要建立一个长时间的同步连接。这种设计对整个系统的稳定性也很重要,用户思考几分钟也好,几小时也好,都不会占用智能体的运行资源。

第三是完整留痕。每一条确认消息的创建时间、内容、确认人、确认结果、回传状态,全都可以记录在数据库里。一旦智能体执行出了问题,你可以回溯到“人到底点了什么”“在什么时间点点的”,这对审计和排查都是救命级的价值。这也是为什么我建议把 Pizza Bot 这类工具用在所有敏感动作的前置闸门上,而不是只在订餐场景里用。

2. 整体架构设计与方案选型

2.1 Pizza Bot 的完整数据流

在动手部署之前,先把 Pizza Bot 的数据流完整走一遍。我把它拆成八个环节,每一步都对应到后面的具体组件。

  1. 后台 AI 智能体调用 API,把一条待确认消息投递到 Pizza Bot 的收件箱(POST /messages)。
  2. API Gateway 做鉴权,校验 API Key 或 IAM 身份之后,转发请求给后端的 Lambda 函数。
  3. Lambda 函数解析消息内容,生成唯一的 messageId,把整条消息写入 DynamoDB,初始状态为 pending。
  4. 写入成功后,通过 SNS 主题触发用户通知——可以是邮件、Slack 消息、短信或者手机推送。
  5. 用户在邮件或 Slack 里点开确认链接,浏览器加载一个简单的确认页面,展示消息详情。
  6. 用户点击“批准”或“拒绝”,页面调用 POST /messages/{id}/response 接口。
  7. 后端更新消息状态为 approved 或 rejected,同时向消息里带的 callbackUrl 发起 HTTP 回调。
  8. Agent 收到回调后,根据结果继续或终止后续流程。

这条链路看起来简单,但每一步都有讲究。最核心的设计决策是:消息的“待确认状态”和“确认结果”都以 DynamoDB 里的数据为准,SNS 通知、回调这些都只是围绕这个核心数据模型的附属动作。这样的好处是,就算某次通知丢失了,数据还在,后续还能补发或者通过查询接口拉取。

2.2 为什么选 AWS SAM + 无服务器架构

Pizza Bot 选择 AWS SAM(Serverless Application Model)作为部署框架,是 AWS 无服务器生态里很顺理成章的选择。SAM 本质上是 CloudFormation 的扩展,把 API Gateway、Lambda、DynamoDB 这些资源用一种更简洁的语法描述出来。你写一个 template.yaml,里面声明好函数、接口、表的定义,然后 sam build 构建、sam deploy 部署,整个基础设施就上线了。

选无服务器架构而非传统服务器,核心考量在于 Pizza Bot 是一个典型的低频、事件驱动型应用。它不会像 Web 服务那样每秒处理大量请求,而是一天里可能只有几十条消息进进出出。用一台 EC2 挂着,浪费资源也浪费成本;用 Lambda 按调用计费,空闲时几乎不花钱。再加上它天生要跟 API Gateway、DynamoDB、SNS 这些 AWS 服务配合,无服务器就成了最自然的选择。

SAM 对开发体验的提升也很明显。你可以直接在本地用 sam local start-api 把接口跑起来,配合 Docker 模拟 Lambda 运行环境,调试完再推到云端。整个 CloudFormation 的栈还能一键回滚,出问题的时候把上一版配置恢复即可。这些能力对基础设施不熟的开发者来说非常友好。

2.3 自托管的价值:数据主权与可扩展性

“自托管”这个词这两年越来越热,我自己在一堆工具上都倾向于自托管,Pizza Bot 也是同样的逻辑。你用自己的 AWS 账号部署,所有消息数据都存在自己的 DynamoDB 表里,不经过任何第三方服务,天然满足数据主权和合规要求。

自托管还有一个好处是扩展方便。Pizza Bot 本身只是一个基础实现,你可以根据自己业务的需要去改。举个例子,我在本地部署的时候就把确认页面从简单的 HTML 换成了公司内部风格的 UI,还额外加了一个“附言”字段,让用户在拒绝的时候填原因。这些改动如果用的是托管 SaaS 服务,根本没机会做。你可以理解成:自托管 Runner 解决的是“代码和凭证不出内网”的问题,自托管的 Pizza Bot 解决的是“决策数据和确认流程不出自家系统”的问题,逻辑上是一脉相承的。

不过自托管也意味着你要自己负责运维。好消息是,无服务器架构已经把运维成本压得很低了,偶尔需要做的就是看看 Lambda 日志、检查 DynamoDB 的容量使用,偶尔调一下 CloudWatch 告警。总体来说,可控性带来的收益远大于额外付出的那点维护成本。

3. 部署实操:从零把 Pizza Bot 跑起来

3.1 前置条件与项目获取

在动手之前,先把环境准备好。部署 Pizza Bot 需要以下几个工具和权限:

  • 一个 AWS 账号,并且本地的 AWS CLI 已经配置好了凭证(aws configure)。
  • SAM CLI,这是部署的核心工具,大部分系统上可以用 brew install aws-sam-cli 或 pip install aws-sam-cli 安装。
  • Docker,SAM 本地调试时需要用来模拟 Lambda 运行环境。
  • Node.js 18 及以上版本,Pizza Bot 的示例代码是基于 Node.js 写的。

项目本身是开源项目,示例代码可以从 GitHub 上获取,以官方示例仓库为例:

git clone https://github.com/aws-samples/pizza-bot-sample.git cd pizza-bot-sample

拿到代码之后,先看一下目录结构。一般情况下会有一个 template.yaml 放在根目录,用于声明整个基础设施的资源;src 目录里面是各个 Lambda 函数的源码;前端如果带了确认页面,通常会有一个 static 或者 web 目录。

pizza-bot-sample/ ├── template.yaml ├── src/ │ ├── handlers/ │ │ ├── inbound.js │ │ ├── respond.js │ │ └── callback.js │ └── utils/ ├── web/ │ └── index.html └── README.md

3.2 核心 SAM 模板与资源解析

template.yaml 是整个部署的灵魂。我挑几个关键资源讲一下,你后面如果想改项目,也基本是从这里下手。

AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Resources: InboxTable: Type: AWS::DynamoDB::Table Properties: BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: messageId AttributeType: S KeySchema: - AttributeName: messageId KeyType: HASH TimeToLiveSpecification: Enabled: true AttributeName: expiresAt PizzaBotApi: Type: AWS::Serverless::Api Properties: StageName: Prod Auth: ApiKeyRequired: true InboundFunction: Type: AWS::Serverless::Function Properties: CodeUri: src/handlers/ Handler: inbound.handler Runtime: nodejs20.x Policies: - DynamoDBCrudPolicy: TableName: !Ref InboxTable - SNSPublishMessagePolicy: TopicName: !Ref NotificationTopic Events: Api: Type: Api Properties: RestApiId: !Ref PizzaBotApi Path: /messages Method: POST

这里有几个地方要重点说明。

DynamoDB 表用的是按需计费模式(PAY_PER_REQUEST),而不是预留容量。这是有意为之,因为确认消息的量通常不大,按需计费可以避免你在流量评估上花时间,价格也不会贵。

TimeToLive 字段很值得学习。确认消息一般有时效性,比如“请在 24 小时内确认”,过了这个时间智能体就应该按超时处理或自动取消。DynamoDB 的 TTL 机制可以在后台自动删除过期数据,省去你写定时清扫逻辑的麻烦。我在示例里用的是 expiresAt 字段,写入消息时根据业务需要计算过期时间。

API 上设置了 ApiKeyRequired,要求调用方必须携带 API Key 才能访问。这个 API Key 由 API Gateway 管理,后期你可以按调用方维度去发不同的 Key,甚至做限流计划。对外暴露的接口加一层 Key 保护是底线,如果你要面向更多调用方,还可以在 API Gateway 前面挂 AWS WAF 做一层防护,防一下注入类的攻击。

3.3 本地调试:sam local 的妙用

部署到云端之前,强烈建议先在本地把整条链路跑通。SAM 的本地调试能力是它对比纯 CloudFormation 最大的优势。

sam build sam local start-api --env-vars env.json

sam build 会把源码打包成 Lambda 的运行格式,sam local start-api 会在本地起一个 API Gateway 模拟器,监听 3000 端口。你本地改了代码之后,它会自动重新加载对应的函数,调试体验和写普通的 Express 服务差不多。

然后我用 curl 模拟一个智能体投递消息的动作:

curl -X POST http://127.0.0.1:3000/messages \ -H "Content-Type: application/json" \ -H "x-api-key: local-test-key" \ -d '{ "requestId": "test-001", "fromAgent": "pizza-order-agent", "type": "confirmation", "content": { "title": "确认披萨订单", "summary": "玛格丽特披萨 x 2,共 42.99 元", "details": ["大份", "加量芝士", "30 分钟内送达"] }, "callbackUrl": "https://your-agent.example.com/webhook/confirm" }'

如果一切正常,会返回一个 201 的响应,body 里带着这条消息的 messageId 和确认页面的链接。你在浏览器里打开那个链接,就能看到确认页面,点一下“批准”,看看回调有没有触发到你的本地服务里。

这里有个小技巧:本地调试时,回调地址可以指向本地起的另一个服务,用 ngrok 这类工具做一个公网映射,就能端到端地跑通“创建消息 -> 确认 -> 回调”的完整闭环。我第一次就是这么干的,比自己盲改代码高效太多了。

3.4 云端部署和关键参数配置

本地没问题了,就可以推到 AWS 云端。部署命令很简单:

sam deploy --guided

--guided 会引导你一步步配置部署参数,其中几个比较关键:

参数说明示例值
Stack NameCloudFormation 栈名pizza-bot-prod
AWS Region部署区域ap-southeast-1
StageNameAPI 网关环境名Prod
NotifyEmail接收通知的邮箱,用于邮件确认ops@example.com
SlackWebhookUrlSlack 通知的 Webhook 地址https://hooks.slack.com/xxx
ConfirmPageTTL确认链接有效时长604800

部署完成后,SAM CLI 会把 API Gateway 的调用地址、API Key、确认页面地址等输出信息列出来。拿到 API Key 之后,需要把它配置到调用方那边,云端环境的调用方式跟本地一样,区别只是把 URL 换掉、Key 换成真实的值。

cloudformation 栈跑完之后,你可以去 AWS Console 看一下 DynamoDB 表,里面应该会出现你测试投递的消息记录。到这里,一个 Pizza Bot 的最小可用版本已经上线了。

4. 接入 AI 智能体:把收件箱变成工作流的一部分

4.1 Agent 侧的代码封装

Pizza Bot 部署好只是一个开始,真正有价值的是把它接入到你的 Agent 工作流里。我拿一个实际的“AI 商品推荐 + 下单”场景来举例(对应目前很流行的 AI 智能体购物助手场景)。

你的 Agent 本身不需要知道 Pizza Bot 的内部实现,只需要封装一个函数,把“需要确认的内容”发过去。以 Python 为例:

import os import uuid import requests PIZZA_BOT_URL = "https://your-api-id.execute-api.ap-southeast-1.amazonaws.com/Prod/messages" API_KEY = os.environ["PIZZA_BOT_API_KEY"] def request_confirmation(title, summary, details, callback_url): payload = { "requestId": str(uuid.uuid4()), "fromAgent": "shopping-agent", "type": "confirmation", "content": { "title": title, "summary": summary, "details": details }, "callbackUrl": callback_url } resp = requests.post( PIZZA_BOT_URL, json=payload, headers={"x-api-key": API_KEY}, timeout=10 ) resp.raise_for_status() return resp.json()

注意 requestId 的生成:这个字段是幂等性的关键,它同时是智能体侧订单号与 Pizza Bot 消息之间的关联键。如果 Agent 因为网络问题重试提交,Pizza Bot 可以根据 requestId 识别出这是同一条消息,避免重复创建多条待确认消息。这是我在实际开发中特别看重的一点——幂等性做不好,确认流程一定会出乱子。

Agent 侧收到回调的时候,根据回调里的状态去做后续分支处理。比如 approved 状态继续调用支付接口,rejected 状态就取消订单并通知用户原因。如果你的 Agent 是在 Coze 这类低代码平台上搭建的,也可以把 Pizza Bot 的 API 封装成一个自定义插件动作,流程上本质一样。

4.2 整个工作流的完整串联

一个面向真实业务的完整智能体工作流大概是下面这个样子:

环节执行者动作说明
1. 用户发出指令用户“帮我买一份周六晚上的双人披萨套餐”
2. 生成候选方案Agent调用商品目录和价格接口,生成选项
3. 投递待确认Agent调用 Pizza Bot,把套餐细节投递到收件箱
4. 异步通知Pizza Bot发邮件/Slack 给用户,提示“有待确认的订单”
5. 用户审批用户点开链接,选择批准或拒绝,可填备注
6. 回调 AgentPizza Bot把批准/拒绝结果 POST 回 Agent 的回调地址
7. 执行后续Agent批准则下单,拒绝则终止并询问调整意见

这套流程的价值在于:整个链条中,Agent 始终没有直接调用下单接口的权限。它有的权限只是“创建一条待确认消息”。真正触发订单的权限被 Pizza Bot 的回调链路掌控,而这条链路最终由用户的行为来决定。哪怕 Agent 的 prompt 被恶意注入,它最多也只能生成几条欺诈性的确认请求,无法直接扣款。

在实际实现里,回调接口本身也要做安全性加固。至少要做到两点:第一,只接受 HTTPS 回调地址,防止明文传输;第二,回调地址建议做白名单校验,避免 Pizza Bot 被诱导向任意外部地址发起请求。如果业务量大了,回调的通知投递可以再换成 SQS 做可靠投递,保证不丢消息。

5. 常见问题与排障实录

5.1 高频问题的速查表

我在部署和接入 Pizza Bot 的过程中,前后踩了不少坑。整理成一个速查表,方便你遇到了直接对照。

现象可能原因排查与解决
调用 API 返回 403API Key 没传或传错检查请求头是否带 x-api-key,去 API Gateway 控制台确认 Key 存在
消息写入成功,但没收到通知邮件SNS 邮件订阅需要确认第一次订阅邮件时,需要去邮箱点击“Confirm subscription”
确认页面打开 404消息已过期(TTL 触发删除)检查创建消息时设置的 TTL,或直接查 DynamoDB 确认数据是否存在
Agent 收到了重复回调回调没有做幂等处理在 Agent 侧用 requestId 或 messageId 去重,只处理第一次结果
中文内容显示乱码创建消息时没有指定 UTF-8请求头加 Content-Type: application/json; charset=utf-8
本地调试时回调失败Docker 网络访问限制用公网可访问的回调地址,或把回调地址指向本机的局域网 IP
Lambda 日志看不到内容函数里没有打印上下文在 handler 入口加 console.log(event) 辅助排查

5.2 我印象比较深的几个排障过程

第一个印象比较深的坑是“回调丢失”。有一次我在测试整个链路,DynamoDB 里状态已经更新成 approved,但 Agent 侧就是没有继续执行。查了半天才发现,问题出在 callbackUrl 指向的是一个内网地址,Lambda 根本访问不到。这个其实很好排查,只要去 CloudWatch 看 Lambda 的日志,就能看到 ECONNREFUSED 的连接错误。后来我在入参校验里严格限制了回调地址的格式,必须是 HTTPS 公网可达的 URL,内网地址一律拒绝。

第二个坑是“确认页面在手机上排版乱掉”。Pizza Bot 自带的确认页面是极简风格的 HTML,PC 浏览器打开没问题,到了手机上按钮会被挤到屏幕外面。我在本地改成了响应式布局,用 viewport 适配,并且把确认按钮做成了大按钮,方便手指点击。这个改动对真实使用体验的提升很明显——因为大多数用户收到通知时,是在手机上处理的。

第三个坑是“冷启动延迟”。无服务器架构的 Lambda 在长时间没有调用之后,第一次请求会有几秒钟的冷启动延迟。Pizza Bot 这种低频应用特别容易触发这个问题。如果你对首次点击确认页面的响应速度敏感,可以在 SAM 模板里给确认相关的 Lambda 配置 Provisioned Concurrency,预置一个并发实例,代价是会产生一点常驻费用。根据我个人的经验,对 Pizza Bot 这种非核心链路来说,冷启动带来的那 1-2 秒延迟一般可以接受,不去折腾配置也问题不大。

5.3 对日志、监控和后期维护的建议

部署完成之后,建议第一时间把 CloudWatch 告警配好。至少监控这几个指标:Lambda 的错误率、DynamoDB 表的读写错误、SNS 投递失败数。具体的告警规则可以设置得宽松一点,比如错误率连续 5 分钟超过 5% 就触发通知,重点是让你在出了问题的时候能第一时间感知,而不是事后被用户反馈提醒。

日志方面,推荐在 Lambda 的 handler 入口统一打一条包含 messageId 和动作的日志,比如“message created”,后续排查问题时可以用 messageId 把整个生命周期串起来。这个习惯非常有用,尤其当你开始接入多个 Agent、多种消息类型之后,日志的检索效率直接决定排障速度。

Pizza Bot 这类应用本质上是一个很小的、边界清晰的基础设施组件。它不会替你解决所有 Agent 的权限问题,但能帮你把最关键的“人工确认点”落地得很干净。关于权限和确认这件事,我的体会是:不要指望智能体永远正确,好的系统设计应该默认它可能会出错,然后在关键路径上设置人工兜底。Pizza Bot 就是这样一个兜底的角色。它后续还可以扩展的方向不少,比如把回调替换成 SQS 做异步可靠投递、接入统一的审计报表,或者把确认页面换成企业内部的单点登录,让整个流程既有自动化效率,又有让人放心的人工干预点。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询