基于 Amazon Bedrock AgentCore 构建无感智能代理与人机协作工作流
- 作者

- 姓名
- Nino
- 职业
- Senior Tech Editor
传统的对话式 AI 代理依赖于“问答响应”范式:用户打开聊天窗口,输入提示词(Prompt),然后等待模型生成结果。这种模式虽然适用于搜索、创作与交互式问答,但在企业级自动化场景中存在局限。无感智能代理(Ambient Agents) 改变了这种交互逻辑,它不再等待人类的主动对话,而是以后台异步方式运行,直接由系统事件触发——例如 Amazon S3 上的文件上传、定时 Cron 任务、数据库变更通知或第三方 Webhook——在无需人类实时监控的情况下完成多步骤推理与任务执行。
然而,构建企业级无感智能代理必须在“自主性”与“安全把控”之间取得平衡。当 AI 代理遇到高风险操作、模糊业务逻辑或低置信度输出时,系统必须能够自动挂起当前状态,触发人机协作(Human-in-the-Loop, HITL)流程,并在管理员审批或修正后再无缝恢复执行。借助 Amazon Bedrock AgentCore、Amazon SQS、AWS Lambda 与 Amazon DynamoDB,开发者可以构建出高可用、跨框架的无感智能代理,实现统一的 ask_human 升级机制与任务管理界面。
在追求极致稳定性与高并发推理的企业场景中,开发者还可以接入 n1n.ai 等多模型聚合平台,在云端配额受限或区域服务波动时,灵活切换或备份底层基础大模型,确保事件驱动流水线的高可靠运行。
整体系统架构设计
为保证高吞吐量与系统解耦,无感智能代理架构将事件接收、任务缓冲、Agent 推理与状态持久化进行分层设计。以下是包含人机协作在内的完整工作流:
[ 事件源 ] (S3 上传 / CloudWatch 定时 / Webhook)
│
▼
[ Amazon SQS 队列 ] (削峰填谷,控制 Agent 并发速率)
│
▼
[ AWS Lambda 执行节点 ] (初始化 Agent Core 运行时环境)
│
▼
[ Amazon Bedrock Agent Core ]
├── 自动执行基础工具 (API 调用 / 数据库检索)
└── 判断需要人工审核时调用 `ask_human` 工具
│
▼
[ Amazon DynamoDB ] (记录挂起状态 PENDING_HUMAN_INPUT)
│
▼
[ 人机协作 Jobs 管理后台 ] (运维/管理人员查看轨迹并输入指令)
│
▼
[ Lambda 回调处理程序 ] (恢复 Agent Core 上下文并继续完成任务)
核心组件职责:
- 事件缓冲队列(Amazon SQS):解耦事件生成与模型推理,防止突发大流量冲垮 LLM 端点配额,并提供失败重试机制。
- 执行工作节点(AWS Lambda):从 SQS 提取消息,构建 Agent 上下文,调用 Bedrock AgentCore 进行思考与操作。
- Agent Core 推理引擎:按需执行业务工具。一旦识别到边界情况或风险操作,立即触发
ask_human函数。 - 持久化状态库(Amazon DynamoDB):存储 Agent 的中间状态、推理链快照以及待人工处理的数据集。
- 人机协作(HITL)管理后台:为企业运维人员提供可视化界面,查看 Agent 的思考推导路径(Reasoning Trace),补充缺失信息或点击授权批准。
对话式 Agent 与无感 Agent 维度对比
| 比较维度 | 对话式 Agent (Conversational Agents) | 无感智能代理 (Ambient Agents) |
|---|---|---|
| 触发机制 | 用户手动输入对话 / HTTP 阻塞请求 | 系统异步事件(S3 上传、Cron 定时、SQS 队列) |
| 执行上下文 | 同步短连接 Session | 异步解耦长生命周期任务 |
| 人机交互模式 | 逐轮实时交互对话 | 异常驱动(仅在触发 ask_human 时挂起) |
| 状态管理 | 内存级会话上下文 | 外部持久化数据库(DynamoDB / Redis) |
| 高可用要求 | 用户感知超时快速报错 | 队列削峰、幂等重试、接入 n1n.ai 实施多模型容灾 |
ask_human 工具设计与实现
在 AgentCore 框架中,工具可以被定义为独立的标准函数。ask_human 本质上是一个熔断器机制,当 Agent 评估自身的置信度低于预设门槛时,通过调用该工具挂起推理循环。
步骤 1:定义框架无关的 ask_human 工具
import json
import boto3
import os
from datetime import datetime
dynamodb = boto3.resource('dynamodb')
jobs_table = dynamodb.Table(os.getenv('JOBS_TABLE_NAME', 'AmbientAgentJobs'))
def ask_human(job_id: str, prompt_question: str, context_data: dict, required_fields: list) -> dict: