基于 Amazon Bedrock AgentCore 构建事件驱动与人机协同的无感 Agent
- 作者
- 姓名
- Nino
- 职业
- Senior Tech Editor
当前大多数生成式 AI 应用均建立在对话式交互界面上，依赖用户在聊天框中手动输入提示词来触发 Agent 执行。然而在企业级应用场景中，越来越多的需求指向了无感 Agent（Ambient Agents）——这是一种在后台运行的自主智能层，它无需人类实时对话，而是被系统事件、定时任务、数据库变更或监控告警主动触发。
构建企业级无感 Agent 需要将异步消息传递机制、长周期运行的 LLM 执行状态以及确定性的人类干预机制无缝结合。借助 Amazon Bedrock AgentCore 以及 AWS 的无服务器生态系统（如 Amazon SQS、AWS Lambda 和 Amazon DynamoDB），开发者能够构建出框架无关的无感 Agent，在高效处理复杂事件信号的同时，无缝集成人机协同（HITL, Human-in-the-Loop）审核工作流。
本文将深入剖析无感 Agent 的架构设计，详细演示如何搭建事件驱动的任务接收引擎、设计包含 ask_human 工具的状态机，并分享企业生产环境落地时的最佳实践。
系统架构：异步无感 Agent 模式
传统的对话式 Agent 运行在同步的 HTTP 请求-响应循环中。与此不同，无感 Agent 消费后台事件流，跨长周期任务轨迹维持执行状态，并且仅在置信度低于预设阈值或即将执行高风险操作时触发人类干预。
以下是端到端的事件驱动无感 Agent 架构图：
+-------------------+ +-------------------+ +-------------------+
| 事件生成源 | ---> | Amazon SQS | ---> | AWS Lambda |
| (S3, CloudWatch, | | (任务队列) | | (Agent 分发器) |
| Webhooks, Cron) | +-------------------+ +---------+---------+
+-------------------+ |
v
+-------------------+
| Bedrock AgentCore |
| 推理决策循环 |
+---------+---------+
|
+-------------------+-------------------+
| |
v v
+---------------------+ +---------------------+
| 执行工具库 | | ask_human 工具 |
| (API, DB, 业务操作) | | (状态: 挂起) |
+---------------------+ +----------+----------+
|
v
+---------------------+
| Amazon DynamoDB |
| (待审核任务表) |
+----------+----------+
|
v
+---------------------+
| 人工审核管理界面 |
| (Jobs 管理页) |
+---------------------+
核心组件职责解析
- 事件接收层：Amazon S3 文件上传、CloudWatch 告警或定时 Cron 触发器将消息推送到 Amazon SQS，完成任务的异步解耦与削峰填谷。
- 调度分发器：AWS Lambda 从 SQS 拉取消息，初始化上下文会话，并启动 Amazon Bedrock AgentCore 的推理循环。
- 推理与工具决策引擎：Amazon Bedrock AgentCore 根据模型输出迭代调用工具。在多模型部署或跨云模型调度场景中，开发者亦可搭配 n1n.ai 等 API 聚合网关，实现高可用模型资源的统一调度。
- 持久化与 HITL 状态库：Amazon DynamoDB 用于记录上下文状态、挂起的审核任务、中间推理日志以及会话 Token。
- 人机协同审核网关：后台管理界面（Jobs Page）供人工审阅 Agent 生成的拟执行方案，完成批准、拒绝或修正后恢复任务执行。
端到端代码实现
步骤 1：定义事件载荷规范
当系统触发事件（例如上传新的发票 PDF 至 Amazon S3 桶）时，SQS 队列将接收到如下格式的 JSON 载荷：
{
"eventId": "evt_9876543210",
"eventType": "DOCUMENT_PROCESSING_REQUEST",
"source": "s3://enterprise-bucket/invoices/2025/INV-4092.pdf",
"timestamp": "2025-02-18T10:30:00Z",
"metadata": {
"priority": "HIGH",
"maxConfidenceThreshold": 0.85
}
}
步骤 2：构建 AWS Lambda 调度处理函数
Lambda 函数负责提取 SQS 消息体，组装 prompt 上下文，并调用 Bedrock AgentCore 执行引擎：
import json
import os
import boto3
from botocore.exceptions import ClientError
bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')
dynamodb = boto3.resource('dynamodb')
JOB_TABLE_NAME = os.environ.get('JOB_TABLE_NAME', 'AgentPendingJobs')
def lambda_handler(event, context):
for record in event['Records']:
payload = json.loads(record['body'])
event_id = payload['eventId']
event_type = payload['eventType']
source_uri = payload['source']
prompt = f