Building ambient agents with Amazon Bedrock AgentCore: From event-driven signals to human-in-the-loop workflows

使用 Amazon Bedrock AgentCore 构建环境代理：从事件驱动信号到人机协作工作流 | 人工智能

使用 Amazon Bedrock AgentCore 构建环境代理：从事件驱动信号到人机协作工作流

大规模处理文档的团队都熟悉这种流程：文件存入存储后，有人注意到，逐一打开每个文件，判断需求并安排审核。监控警报也以相同方式排队，等待人工处理。手动分类所浪费的小时数正是环境代理要解决的操作问题。想象一下文档被上传到你的 Amazon Simple Storage Service（Amazon S3）存储桶后，几秒钟内你的工作流页面就会出现一个待执行的任务（或根据配置已开始执行）。代理会分析文件，呈现发现结果，并在继续下一步前请求你的批准。事件本身即为提示。这就是环境代理：它响应事件流，仅在需要时通过单个 ask_human 工具暂停以获取人工输入，待人工回应后从暂停处继续执行。

图1：环境代理在 AgentCore 运行时响应事件并暂停等待人工输入的概览

目前大多数 AI 代理体验遵循不同模式：用户打开聊天界面，输入提示，等待响应。这种方式适用于一次性问题，但限制了代理同时只能处理一个对话，并要求人工先描述情况才能触发后续操作。对于需要代理响应基础设施中事件的场景（文件上传、数据库变更、计划任务、系统警报），这种仅依赖聊天的模型就难以适用。

环境代理描述了另一种范式，LangChain 等工具已对此进行了阐述。环境代理不等待用户发起对话，而是监听事件流并即时响应，可能同时处理多个事件。它们不仅由人工消息触发，多个代理可并行运行。关键的是，它们并非完全自主：生产环境设计会特别关注代理何时暂停与人工交互。当信号触发时，代理执行其工作流，仅在需要澄清、批准或审核时中断人工。这种人机协作组件降低了代理投入生产的风险，建立用户信任，并通过反馈使代理持续学习和改进。

目前在 AWS 上运行的组织已具备事件驱动的基础设施：Amazon S3 事件通知、Amazon EventBridge 规则、AWS Lambda 触发器和 Amazon DynamoDB 流。缺失的环节是将这些事件源连接到能够推理事件、采取行动并在需要时引入人工的智能代理。像 AWS Step Functions 这样的全自动化流水线可以编排工作流，但无法处理歧义或提出澄清问题。基于聊天的代理可以推理，但需要人工先发起对话。环境代理填补了这一空白。

Amazon Bedrock AgentCore 是一个平台，可用于构建、连接和优化大规模代理，支持任何框架或模型。AgentCore Runtime 提供了使这种模式得以实现的执行环境：基于容器的代理托管，支持长期运行的工作负载，内置会话隔离，并与 Amazon Bedrock 基础模型集成。AgentCore Runtime 支持足够长的会话，以覆盖此处所示的信号 → 代理 → 人工参与（HITL）流程。参考实现将每个代理回合限制在 Lambda 15 分钟超时内，这在实践中提供了充足的余量。结合 AWS Lambda 进行事件处理和 Amazon DynamoDB 进行状态管理，结果是一个完全无服务器的环境代理平台。

在本文中，我们将逐步介绍在 Amazon Bedrock AgentCore 上实现该模式的完整流程。您将了解以下内容：

• Amazon S3 或计划事件如何成为 AgentCore Runtime 上代理运行的任务，无论是否有人工参与。
• 一个 ask_human 工具加上标准响应封包即可支持完整的人工参与交互范围。
• 参考示例中包含的内容，以及您为自己的用例需要自行编写的部分。

先决条件

在部署参考实现之前，请确保已具备以下条件：

• 具有创建 AWS 身份和访问管理（IAM）角色、Lambda 函数、DynamoDB 表、S3 存储桶、Amazon 简单队列服务（Amazon SQS）队列、Amazon API Gateway API、Amazon CloudFront 分发、Amazon Cognito 用户池、Amazon 弹性容器注册表（Amazon ECR）存储库和 Bedrock AgentCore 运行时权限的 AWS 账户。在沙盒账户上拥有管理员权限是一个良好的起点。

• 已使用 AWS 命令行界面（AWS CLI）配置凭证的 AWS 账户，并将默认 AWS 区域设置为 us-east-1（示例默认配置针对该区域）。

• 在您的账户和区域中已安装并引导 AWS 云开发工具包（AWS CDK）v2（执行 cdk bootstrap）。

• 本地已安装并运行 Docker。代理容器作为部署的一部分将被构建并推送到 Amazon ECR。

• 后端 Lambda 函数和代理构建使用 Python 3.11 或更高版本，React 前端使用 Node.js 18 或更高版本。

• 在目标区域中可以访问 Anthropic Claude Sonnet 4.5 模型。如果您之前未使用过 Bedrock，请按照《管理对 Amazon Bedrock 基础模型（FMs）的访问》启用该模型。模型可用性因 AWS 区域而异。请查看 Amazon Bedrock 文档，获取目标区域当前支持的模型列表。稍后切换模型只需一行配置更改即可。

理解环境代理

在深入架构之前，了解环境代理与典型聊天机器人有何不同，并了解其余部分文章依赖的构建模块：事件驱动的触发模型、环境信号抽象，以及将它们连接在一起的单一人工参与工具。

事件驱动与用户发起代理的对比

用户发起代理遵循请求-响应模式：

用户 → 提示 → 代理 → 响应 → 用户

环境代理遵循事件驱动模式：

事件 → 信号 → 代理 → [可选的人工交互] → 操作

关键区别在于触发机制。环境代理由系统事件而非明确的用户请求激活，这使它们非常适合文档处理流程、监控和告警、计划分析以及需要审批关卡的多步骤工作流。

环境信号：触发机制

环境信号是一种配置，将事件源映射到代理。当事件发生时，平台会自动为代理创建一个任务。接下来发生什么取决于信号的一个设置：

• 当 autoExecute: false（默认值）时，任务会进入“任务”页面并处于空闲状态，等待人工审核并执行。当信号可能在未知输入上触发，或代理拥有高风险工具时，这种安全的先审核流程是必要的。

• 当 autoExecute: true 时，信号处理器会直接将任务加入到工作者队列，代理立即运行，只有当代理本身调用 ask_human 时才会引入人工干预。这是完全自主的流程。

该模式涵盖多个信号事件源。参考示例中包含了前两个事件源，其余部分是通过编写新的处理 Lambda 函数并在“信号”页面上添加对应的表单字段来扩展的：

• Amazon S3 文件上传（已包含）：当文件上传到特定存储桶和前缀时触发。

• 计划事件（已包含）：按照类似 cron 的计划触发代理。由携带 jobType: "scheduled" 的任务驱动，而非“信号”页面上的信号。

• API Webhook（扩展点）：响应外部系统的通知。

• 数据库变更（扩展点）：响应 Amazon DynamoDB 流或 Amazon Relational Database Service（Amazon RDS）事件。

人工参与机制：一个工具、一个封装、一个视图

环境代理需要与人类交互的结构化方式。在此示例中，代理通过单一工具（ask_human）暴露这些交互，并返回标准的响应封装。在该封装中，状态可以是 completed（完成）、interrupted（中断）或 error（错误），对应的字段是 result（结果）、question（问题）或 error（错误）。平台还会在每个响应中传递 session_id 和 job_id，以便后续操作可以关联。这些是关联元数据，不属于代理必须实现的核心协议部分。当代理返回 interrupted 时，平台会将任务状态设为 interrupted，并将其 requiresAction 标志设为 true。参考 React 前端在“任务”页面的“中断”标签页中显示这些信息，每行带有警告标识，因此无需轮询独立的审核队列。相同的“任务”视图还显示待处理问题、等待批准的操作建议、最终结果和失败任务，为用户提供一个统一的界面，查看所有代理正在执行的操作，而不是监控多个聊天窗口或邮件线程。

同样的机制支持读者可能从更广泛的代理文献中熟悉的几种提示模式：一个仅报告结果的“通知”回合（Notify turn），一个询问澄清的“问题”回合（Question turn），一个提出操作建议并等待 APPROVE/REJECT/MODIFY 的“审核”回合（Review turn），以及一个将失败记录在任务记录中并让用户决定是否重试的“错误”回合（Error turn）。这些是代理编写问题的方式惯例，而非独立的运行时模式。在平台层面，只存在一条代码路径和一个封装。

图2：通过ask_human工具呈现的Notify、Question和Review三种人机协作模式

架构概述

该平台由一组无服务器组件通过事件管道连接而成。本节首先介绍端到端流程，然后依次描述每个组件。

事件在平台中的端到端流动方式如下：Amazon S3发出s3:ObjectCreated通知，该通知由信号处理器Lambda函数接收。信号处理器在ambient-signals表上查询全局二级索引（GSI），查找与事件存储桶匹配的信号，然后为每个匹配项创建作业记录。API层级（或调度器）将作业加入Amazon SQS队列。同一个Job Execution Lambda函数既处理API路径请求，也通过附加的SQS事件源从队列中取出任务，调用Amazon Bedrock AgentCore Runtime上的代理，并将结果（以及任何需要人工输入的请求）写回Amazon DynamoDB。通过Amazon CloudFront从Amazon S3提供的React前端会轮询一个小型的Amazon API Gateway和Lambda层级以获取更新，并允许用户响应待处理的交互。

主要组件包括：

• Amazon S3配合事件通知作为信号的入口点，当文件上传时触发。前缀和后缀过滤器被推送到存储桶的通知配置中，确保信号处理器仅对可能匹配信号的事件进行处理。

• Amazon SQS将API网关请求与代理调用解耦。作业执行队列保存待处理的任务。死信队列（DLQ）用于捕获在配置的重试次数后仍无法处理的消息。

• 三个管道Lambda函数将事件从接收端传递到代理（后面部分将描述API网关后的独立管理层级）：信号处理器将传入事件与配置的信号定义进行匹配并创建作业。作业执行Lambda函数包含两个入口路径：一个API处理程序用于入队消息，另一个SQS工作程序用于消费消息并使用作业上下文调用AgentCore Runtime。调度器按照每分钟一次的cron表达式触发，将到期的计划作业加入同一SQS队列。

• Amazon Bedrock AgentCore Runtime在隔离容器中运行代理代码，支持长期运行的工作负载。

• Amazon DynamoDB存储代理注册表、作业记录、环境信号定义、聊天线程、对话历史和Powertools幂等性记录。通过UpdateItem + list_append原子追加对话消息，确保并发写入时不会相互覆盖。

• 位于Amazon API Gateway后的五个管理层级Lambda函数，为前端提供REST API（agent_management、job_management、signal_management、chat_management、conversation_management），以及一个被chat_management异步调用的chat_execution工作Lambda函数，使聊天API调用能够立即返回。有关它们的部署方式，请参见生产部署部分。

• 通过Amazon CloudFront从Amazon S3提供的React前端，提供代理管理UI，用户可在此监控作业、与代理聊天、回答问题并审核待处理操作。

图3：事件从Amazon S3经由Amazon SQS和AWS Lambda流向AgentCore Runtime的流程，Amazon DynamoDB中保存状态信息，React前端提供交互界面

构建事件基础设施

在了解架构之后，下一步是连接事件源，将Amazon S3上传转换为代理任务。本节将创建存储桶，将其事件通知指向信号处理器Lambda函数，并逐步讲解处理器如何将事件与配置的信号进行匹配。

设置Amazon S3信号触发器

创建Amazon S3存储桶并配置驱动信号处理器的事件通知：

aws s3 mb s3://amzn-s3-demo-bucket-$(date +%s)

该存储桶配置为直接将s3:ObjectCreated:*事件发送到信号处理器Lambda函数。在此示例中，当信号被创建或更新时，signal_management Lambda函数会动态安装通知配置，因此为新前缀添加新信号不需要重新部署。

信号处理器Lambda函数

信号处理器接收Amazon S3事件，在DynamoDB中查找匹配的信号定义，并为每个匹配项创建任务。其核心处理逻辑如以下示例所示。后端/functions/multi_agent/signal_processor.py中的实际处理程序还使用AWS Lambda Powertools进行结构化日志记录和幂等性处理，会查询信号表上的bucketName-signalId-index全局二级索引，应用每个信号配置的前缀和后缀检查，并写入signal_triggered任务记录。

def process_s3_signal(event):
    """Process S3 file-upload events and create agent jobs."""
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]

        for signal in find_matching_signals(bucket, key):
            create_agent_job(signal, {"bucket": bucket, "key": key})

信号配置数据模型

信号以以下结构存储在DynamoDB中：

{
  "signalId": "sig-123abc",
  "userId": "user-456def",
  "agentId": "agent-789ghi",
  "signalName": "Document Processor",
  "signalType": "s3_file_upload",
  "enabled": true,
  "autoExecute": false,
  "bucketName": "ambient-agent-documents",
  "configuration": {
    "bucketName": "ambient-agent-documents",
    "prefix": "invoices/",
    "suffix": ".pdf"
  },
  "triggerCount": 42,
  "lastTriggered": "2026-04-15T10:30:00Z",
  "createdAt": "2026-04-01T00:00:00Z"
}

通过API创建信号时，仅需设置configuration.bucketName。前面示例中顶层的bucketName由平台自动填充。由于DynamoDB GSI分区键不能嵌套在映射属性中，signal_management会在每次写入时将configuration.bucketName镜像到顶层bucketName，使bucketName-signalId-index GSI能够针对每个Amazon S3事件扩展到匹配信号。

autoExecute标志是决定代理是否自主触发或需要人工先审核任务的唯一开关。在代理管理UI的信号表单中，该标志作为复选框与常见的enabled设置并列显示，因此更改行为只需快速编辑，无需修改代码或数据库。

图4：信号定义页面上的autoExecute切换开关

在AgentCore Runtime上部署代理

Amazon Bedrock AgentCore Runtime 主机托管代理容器，并通过 InvokeAgentRuntime API 暴露该容器，该 API 由处理每个任务的 worker Lambda 函数调用。本节将逐步介绍随示例提供的代理布局、驱动其运行的配置，以及将 LangGraph 工具调用转换为平台响应封装的小型编排器。

代理打包与结构

AgentCore Runtime 为代理提供基于容器的执行环境，因此您可以使用任何 Python 代理框架。该示例采用模块化设计：

agent/
├── agent.py             # 入口点
├── config.yaml          # 代理配置
├── requirements.txt     # 依赖项
├── core/                # 平台集成
│   ├── agent_core.py        # 主代理逻辑
│   ├── tool_factory.py      # 配置驱动的工具创建
│   └── execution_control.py # 会话状态 + 循环检测
└── tools/               # 自定义工具
    ├── calculator.py        # 数学运算
    ├── human_input.py       # 人机协作
    └── s3_reader.py         # 导出 list_s3_files + read_s3_file

代理配置

config.yaml 定义代理行为、工具和系统提示。该示例默认使用 Amazon Bedrock 上的 Anthropic Claude Sonnet 4.5 进行本操作指南，这非常适合依赖人机协作工作流的多步骤工具调用和长上下文推理。切换到 Claude Haiku、Amazon Nova 或 Amazon Bedrock 上的其他工具调用模型（各区域可用性不同）只需将以下配置中的 model_id 修改为一行。max_iterations: 10 为图提供了足够的空间，使其在停止前可以进行大约十次模型到工具的往返（编排器会将该值翻倍以计算 LangGraph 的递归限制，因为每次往返会遍历两个图节点），这涵盖了示例中 S3、计算器和 ask_human 工具预期的多步骤工具使用。该文件还包含一个 execution: 块（循环检测器、熔断器、会话缓存阈值），此处为简洁起见省略。完整文件请参阅 agent/config.example.yaml。

# Amazon Bedrock 配置
aws:
  bedrock:
    model_id: "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
    region_name: "us-east-1"

# 代理行为
agent:
  verbose: true
  max_iterations: 10
  handle_parsing_errors: true

# 工具配置（启用配置驱动的工具组合）
tools:
  calculator:
    enabled: true
    type: "calculator"
    name: "calculator"
    description: "执行数学计算。"

  human_input:
    enabled: true
    type: "human_input"
    name: "ask_human"
    description: "向人类用户请求输入或澄清。"

  s3_list:
    enabled: true
    type: "s3_list"
    name: "list_s3_files"
    description: "列出 Amazon S3 存储桶中的文件。"

  s3_reader:
    enabled: true
    type: "s3_reader"
    name: "read_s3_file"
    description: "从 Amazon S3 读取和分析文件。"

prompts:
  system_template: |
    您是一个拥有工具集访问权限的有用 AI 助手。
    仅在工具能增加价值时调用工具。优先提供基于工具结果的简洁回答。

代理基于 langchain.agents.create_agent 构建，这是一个编译为 LangGraph 的工具调用代理。LangChain 负责编排层，因为它提供了预构建的工具调用模式、广泛的开放集成生态系统以及许多团队已经熟悉的 API，而 AgentCore Runtime 则提供底层的托管主机、会话隔离和扩展能力。这两层是互补的。围绕图的平台包装器执行三项任务：它为当前回合构建消息列表（包括任何对话历史记录，对于信号触发的任务，还包括触发信号的 Amazon S3 存储桶和密钥，使模型无需被告知即可获取文件）、调用图，并扫描工具输出中的 ask_human 守卫符，以便将工具调用转换为中断响应。

最简单形式（agent/core/agent_core.py 中的完整版本还处理每会话历史缓存、循环检测和执行跟踪），编排器如下所示：

from bedrock_agentcore import BedrockAgentCoreApp
from langchain.agents import create_agent
from langchain_aws import ChatBedrock
from langchain_core.messages import ToolMessage

from core.tool_factory import create_tools_from_config, load_config
from tools.human_input import HUMAN_INPUT_SENTINEL

class Agent:
    def __init__(self):
        self.config = load_config()
        self.llm = ChatBedrock(
            model_id=self.config["aws"]["bedrock"]["model_id"],
            region_name=self.config["aws"]["bedrock"]["region_name"],
        )
        # 工具从 config.yaml 构建，因此启用工具只需配置更改而非代码更改。
        self.tools, _ = create_tools_from_config()
        self.agent_graph = create_agent(
            model=self.llm,
            tools=self.tools,
            system_prompt=self.config["prompts"]["system_template"],
        )

    def invoke(self, payload):
        """为一个回合运行代理图。简化版：实际方法还会加载对话历史记录，在 ContextVar 上绑定执行状态供工具读取，记录执行跟踪，并在每次响应中传递 ``session_id`` 和 ``job_id``，以便平台关联后续回合。"""
        session_id = payload.get("session_id", "default")
        job_id = payload.get("job_id")
        result = self.agent_graph.invoke(
            {"messages": self._build_messages(payload)},
        )

        for msg in result["messages"]:
            if isinstance(msg, ToolMessage):
                content = str(msg.content)
                if content.startswith(HUMAN_INPUT_SENTINEL):
                    question = content[len(HUMAN_INPUT_SENTINEL):].strip()
                    return {
                        "status": "interrupted",
                        "question": question,
                        "session_id": session_id,
                        "job_id": job_id,
                    }

        return {
            "status": "completed",
            "result": self._extract_final_output(result["messages"]),
            "session_id": session_id,
            "job_id": job_id,
        }

app = BedrockAgentCoreApp()
agent = Agent()

@app.entrypoint
def invoke(payload):
    """AgentCore Runtime 入口点。委托给代理包装器。"""
    return agent.invoke(payload)

部署代理

使用提供的脚本将代理部署到 AgentCore Runtime：

cd agent

 /think

# 在 .env 中配置 AWS 设置，并在 config.yaml 中进行任何覆盖设置。
# 然后构建容器，推送到 Amazon ECR，并将其注册到
# Bedrock AgentCore Runtime。
./deploy_agent.sh

该脚本会返回一个代理运行时Amazon资源名称（ARN），后端会将其存储在代理注册表中，以便作业执行Lambda函数可以调用它。

使用 DynamoDB 进行状态管理

Amazon DynamoDB 是所有需要超越单次 Lambda 调用生命周期的数据的系统记录：已注册的代理、正在运行的作业、提供代理跨轮次连续性的对话历史，以及 UI 读取的信号定义和聊天线程。以下小节将描述表结构、会话模型，以及作业执行 Lambda 函数如何利用这两者来驱动作业完成。

数据库设计

状态层由 DynamoDB 支撑。本文中涉及的核心表包括：

代理注册表（每个已注册代理运行时一条记录）：

{
  "agentId": "agent-123abc",
  "agentName": "Document Analyzer",
  "agentArn": "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/abc123",
  "status": "active"
}

作业注册表（每个代理调用一条记录）。默认情况下，当 autoExecute: false 时，新触发的信号作业会以空闲状态进入此表，等待用户选择执行：

{
  "jobId": "job-789ghi",
  "agentId": "agent-123abc",
  "sessionId": "session-xyz789",
  "jobName": "Signal: Document Processor - invoice.pdf",
  "jobType": "signal_triggered",
  "status": "idle",
  "requiresAction": false,
  "metadata": {
    "signalId": "sig-123abc",
    "triggerPayload": {
      "bucket": "documents",
      "key": "invoices/invoice.pdf",
      "eventSource": "s3"
    }
  }
}

当作业运行时，工作者会将状态更改为 busy。如果代理通过调用 ask_human 暂停，状态会变为 interrupted，requiresAction 变为 true，这会使作业显示在 Jobs 页面的 Interrupted 选项卡中。

对话存储表（每个会话的完整消息历史，设置 30 天的生存时间（TTL））：

{
  "sessionId": "session-xyz789",
  "agentId": "agent-123abc",
  "messages": [
    { "type": "human", "content": "Analyze this invoice", "timestamp": "2026-04-15T10:30:00Z" },
    { "type": "ai", "content": "Which category should I focus on?", "timestamp": "2026-04-15T10:30:30Z" }
  ],
  "ttl": 1778803200
}

还有其他表用于环境信号、聊天线程和 Powertools 去重记录。计划执行通过 job-registry 表的 jobType 和 nextRun GSI 重用，而不是使用单独的表。

会话管理

会话在作业执行和聊天轮次之间提供对话连续性。conversation_management Lambda 函数负责 DynamoDB 的持久化：每个轮次（人类 + AI 对）都会通过原子的 UpdateItem + list_append 追加到会话中，这样同一会话的两个并发写入者不会互相覆盖，而 30 天的 TTL 会处理清理。

带对话连续性的作业执行

作业执行 Lambda 函数有两个入口路径。API 路径会发送一个 SQS 消息（包含 Powertools 幂等性处理和 userId 所有权检查），并立即返回 202 Accepted 状态码，因此前端无需等待模型处理结果。SQS 工作线程路径才是实际执行工作的地方：它加载对话历史记录，将任何用户回复折叠为延续提示，调用 AgentCore 运行时，并将结果写回。该工作线程大致如下：

# 工作线程路径 - 消费 SQS 队列并调用 AgentCore 运行时
def execute_task(job, human_response=None):
    session_id = job["sessionId"]
    agent_id = job["agentId"]
    if human_response:
        prompt = (
            "Previous conversation:\n"
            f"{load_conversation_context(session_id, agent_id)}\n\n"
            f"Human: {human_response}"
        )
    else:
        prompt = job["prompt"]

    response = bedrock_agentcore.invoke_agent_runtime(
        agentRuntimeArn=job["agentArn"],
        runtimeSessionId=session_id,
        # SDK 期望将负载作为 JSON 编码的字符串
        payload=json.dumps(
            {"prompt": prompt, "metadata": job.get("metadata", {})},
        ),
    )
    result = json.loads(response["response"].read())
    save_conversation_turn(
        session_id, agent_id, prompt, result.get("result", ""),
    )

    return {
        "status": result.get("status", "completed"),
        "sessionId": session_id,
        "result": result.get("result"),
        "question": result.get("question"),
    }

图 5：作业执行流程，从 API 入队到 SQS 工作线程和 AgentCore 调用，最终写回 DynamoDB

实现人机协作模式

前面的代理代码片段在图产生 ask_human 工具调用时返回了基于哨兵的中断信封。本节将聚焦 ask_human 工具本身，展示协调器如何在不破坏推理循环的情况下捕获该哨兵。

人工输入工具

人机协作功能的核心是 ask_human 工具。它不会抛出异常：由于 LangGraph 的工具节点会将工具异常捕获为错误观察并反馈给模型，该工具反而返回一个哨兵字符串。协调器在图执行完成后会在 ToolMessage 流中检测到该哨兵并将其转换为中断响应。每次调用的状态（指标、会话 ID 等）不是工具参数。协调器通过 ContextVar 绑定这些信息，工具通过 current_execution_state.get() 读取，这确保了同一容器内并发调用之间的隔离。

from core.execution_control import current_execution_state

HUMAN_INPUT_SENTINEL = "__HUMAN_INPUT_REQUIRED__::"

def create_human_input_tool_func():
    """创建 ask_human 工具函数。"""

    def ask_human_wrapper(question: str) -> str:
        """向人类用户请求输入或澄清。

        返回一个协调器会转换为中断的哨兵字符串。模型永远不会看到这个哨兵，因为协调器在观察到它时会停止图执行。
        """
        state = current_execution_state.get()
        if state is not None:
            state.execution_metrics.record_human_interaction()
        return f"{HUMAN_INPUT_SENTINEL}{question}"

    return ask_human_wrapper

代理端使用

代理工具在工具调用循环中自然地使用该工具。例如，在分析包含多个条目发票时，代理可能会调用：

Tool call: ask_human
Arguments: {
  "question": "这张发票涉及三个费用类别。我需要详细分析硬件、服务还是软件许可？"
}

该工具返回哨兵值后，协调器会停止流程图，将问题作为最新AI轮次持久化存储到对话库中，并在UI上显示中断的作业。当用户回复后，作业执行Lambda函数会将答案折叠到延续提示中，再次调用代理。

构建代理管理UI

代理管理UI是平台的人机交互界面：用户在此浏览作业、回答待处理问题并直接与代理对话。这是一个使用相同REST API的React单页应用，该API在之前的内容中已做过描述。接下来的两个小节将介绍UI布局和将信号、作业与人工输入关联起来的用户体验流程。

前端架构

前端是基于React和Cloudscape Design框架的单页应用，通过Amazon CloudFront从Amazon S3提供服务。它提供五个标签页的导航：

• 工作流：用于管理将代理和信号分组的工作流级定义的画廊和CRUD界面。

• 聊天：独立聊天页面（/chat, /chat/:threadId），用于用户与已注册代理发起对话。

• 代理：管理已注册代理的运行时信息（名称、ARN、能力、状态）。

• 作业：列出所有作业并提供状态过滤功能，详情视图包含交互式聊天标签页、执行追踪和元数据。

• 信号：定义和切换环境信号。预置表单包含Amazon S3前缀和后缀过滤器及计划设置。当连接Webhook、Amazon EventBridge或DynamoDB流扩展点时，会添加对应字段。

独立聊天页面和作业详情聊天标签页复用了相同的聊天组件。当作业处于运行状态（status === "busy"）时，面板会以短时间间隔轮询/conversations/:sessionId接口，确保代理或第二个浏览器标签页的回复能在几秒内显示。作业结束后轮询停止。由于两个位置使用的是相同组件，用户可以在作业运行期间直接从作业详情视图与代理自由对话，而不仅限于回答单个待处理问题。

用户体验流程

• 信号触发：新作业以jobType: "signal_triggered"类型出现在作业列表中。若信号设置autoExecute: false（默认值），作业将进入空闲状态，需用户手动运行。若设置autoExecute: true，作业工人在列表刷新前已开始起草响应。用户发起的作业标记为jobType: "user_initiated"，并使用不同徽章标识以便区分。

• 请求人工输入：任何调用ask_human的作业会进入"中断"标签页，其requiresAction标志设为true，对应行会显示警告标识。

• 响应处理：详情视图提供"提供响应"按钮，用户无需离开作业上下文即可回答待处理问题。

• 执行反馈：状态转换（空闲→忙碌→完成/中断/错误）会被轮询检测，并在列表和聊天标签页标题中实时反映。

• 交互式聊天：工作详情的聊天标签页和/chat页面都会渲染完整的对话历史记录，并接受新的用户消息，因此人类可以在不离开用户界面的情况下回答待处理的问题，或通过提供更多上下文信息来引导代理。

• 响应提交：发送消息会调用聊天执行或工作执行Lambda函数，该函数会在AgentCore运行时上使用工作上下文和用户响应作为下一步的输入来调用代理。

• 完成与审计：最终结果会存储在工作记录中，完整的对话历史记录会保留在会话中，并在聊天标签页中提供用于审计和重复使用。

身份验证由Amazon Cognito和API网关的CognitoUserPoolsAuthorizer处理（参见backend/infrastructure/multi_agent_stack.py）。

实时更新通过轻量级轮询实现，轮询对话和工作端点，因此不需要运行WebSocket基础设施。

扩展示例

参考实现特意设计得较小，以便您在开始添加功能前能看清合同内容。本节描述了开箱即用的功能、您需要构建的内容，以及插入新事件源或工具的位置。

与您构建的内容相比，示例中包含的功能

在扩展之前，了解连接点所在位置会有所帮助。该示例是环境代理模式的参考实现，而非即开即用的产品。开箱即用的功能包括：

• 平台：信号输入、基于SQS的工作流程管道、HITL中断机制、DynamoDB中的工作和对话状态、带有工作页面和聊天界面的React用户界面，以及与Cognito身份验证集成的AgentCore。

• 参考代理：一个容器化的LangChain代理，包含四个工具（计算器、询问人类、列出S3文件、读取S3文件）和一个通用系统提示。它适用于验证整个管道，但不适合直接解决您的业务问题。

• 通过信号界面连接的一个信号源（Amazon S3文件上传，通过存储桶、前缀和后缀匹配），以及一个针对jobType为" scheduled"的工作的每分钟cron作业级调度器。如果Amazon S3上传和cron不足以满足需求，需要在后端和信号页面表单中扩展signalType枚举以添加新的信号类型，如Webhook、Amazon EventBridge事件或DynamoDB流。

您需要构建的内容，以及平台对每部分的期望：

• 您的代理：一个针对您领域定制的容器化Python代理，具有自己的工具和系统提示。它必须遵循平台依赖的两种约定：返回包含三个状态的信封（完成、中断或错误，分别带有匹配的结果、问题或错误字段），并使用ask_human哨兵请求人工输入。这大约需要在您偏好的代理框架周围编写约50行适配器代码。

• 如果Amazon S3上传和cron不足以满足需求，新的信号类型：一个新的处理Lambda函数，用于查询信号表并写入signal_triggered工作行（如下模板所示），以及在信号页面上的对应表单字段，以便用户进行配置。

• 您代理的新工具：在agent/tools/下的新模块、agent/core/tool_factory.py中的注册条目，以及config.yaml中的启用标志。编排器和HITL流程不会改变。任何返回字符串的工具都会参与相同的图。

简而言之：平台负责底层实现，你只需提供智能。你在三个地方编写代码（代理逻辑、工具实现和新的信号处理器）都可以无缝嵌入现有架构，无需改动其他部分。

以配置替代代码

为现有代理启用新功能只需修改配置而非编写代码。代理的 config.yaml 文件可以开关工具并提供描述，因此添加新的 Amazon S3 前缀处理器或新的计算器模式无需重新构建容器。信号定义存储在 DynamoDB 中，可通过 UI 的 Signals 页面进行编辑。

添加新的信号类型

目前信号可触发于 Amazon S3 事件和定时任务。添加新的信号类型（如 Webhook、Amazon EventBridge 规则、DynamoDB 流、Kafka 主题）都遵循相同的三步协议。下游管道对信号类型透明：从 SQS 工作者到 AgentCore 调用再到 UI，所有 signal_triggered 任务都采用统一处理方式，你只需编写从事件源到任务行的适配器。

# backend/functions/multi_agent/webhook_signal_processor.py (新文件)
import json, os, uuid, boto3
from datetime import datetime, timezone
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource("dynamodb")
sqs = boto3.client("sqs")
signals = dynamodb.Table(os.environ["AMBIENT_SIGNALS_TABLE"])
# 由于历史原因，任务表的环境变量名为 TASK_REGISTRY_TABLE；
# 该表本身存储的正是博文其他部分提到的相同任务行。
jobs = dynamodb.Table(os.environ["TASK_REGISTRY_TABLE"])
QUEUE = os.environ["JOB_EXECUTION_QUEUE_URL"]

def handler(event, _context):
    """Webhook -> ambient job. 通过 API Gateway 触发."""
    payload = json.loads(event.get("body") or "{}")
    source = payload.get("source", "unknown")

    # 1. 查询启用的 Webhook 信号，匹配该事件源
    resp = signals.query(
        IndexName="signalType-source-index",  # 在 CDK 中添加此 GSI
        KeyConditionExpression=Key("signalType").eq("webhook") & Key("source").eq(source),
        FilterExpression="enabled = :e",
        ExpressionAttributeValues={":e": True},
    )

    for signal in resp.get("Items", []):
        job_id, session_id = str(uuid.uuid4()), str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()
        auto = bool(signal.get("autoExecute", False))

        # 2. 使用与 S3 路径相同的结构写入任务行
        jobs.put_item(Item={
            "jobId": job_id,
            "userId": signal["userId"],
            "agentId": signal["agentId"],
            "jobName": f"Signal: {signal['signalName']} - {source}",
            "jobType": "signal_triggered",
            "status": "busy" if auto else "idle",
            "sessionId": session_id,
            "prompt": f"Webhook from {source}:\n{json.dumps(payload)}",
            "requiresAction": False,
            "createdAt": now, "updatedAt": now,
            "metadata": {"signalId": signal["signalId"],
                         "signalType": "webhook",
                         "triggerPayload": payload,
                         "autoExecute": auto},
        })

3. 如果信号配置为自动执行，则入队

if auto: sqs.send_message(QueueUrl=QUEUE, MessageBody=json.dumps({"jobId": job_id, "signalTriggered": True, "signalId": signal["signalId"]}))

return {"statusCode": 202, "body": "accepted"}

你将 Lambda 函数及其 API Gateway 路由添加到 CDK 堆栈中，扩展前端 TypeScript 联合类型和 Signals 页面表单中的 signalType，其余部分（Jobs 页面 UI、ask_human 中断处理、对话持久化、autoExecute 开关、契约测试）均可直接使用。与生产环境的 Signal Processor 一样，实际的 Webhook 处理程序会添加 AWS Lambda Powertools 的幂等性功能，以防止重复交付创建重复任务。

## 生产部署

在本地组件整合完成后，下一步是在账户中进行部署。平台以单个 AWS CDK 堆栈的形式进行部署。以下小节将描述该堆栈创建的内容、部署命令、运行时重要的指标，以及示例中自带的安全性和扩展性默认配置。

### 基础设施即代码

整个平台定义在 backend/infrastructure/multi_agent_stack.py 中，是一个 AWS CDK 堆栈。在存储和消息传递方面，它会创建具有按请求付费计费和时间点恢复功能的 DynamoDB 表（代理注册表、任务注册表、对话存储、环境信号、聊天线程、幂等性记录），以及任务执行队列及其死信队列。

在计算和交付方面，它会创建架构部分提到的三个管道 Lambda 函数（Signal Processor、Job Execution、Scheduler），支持 REST API 的五个管理层 Lambda 函数（agent_management、job_management、signal_management、chat_management、conversation_management），以及 chat_management 异步调用的 chat_execution 工作 Lambda，一个基于 Cognito 的 REST API，以及为 React 前端提供服务的 CloudFront 分发。

IAM 权限直接来源于代码：每个表授予 DynamoDB 的读写权限，根据需要授予 SQS 的发送和消费消息权限，以及在工作角色上设置范围限定的 bedrock-agentcore:InvokeAgentRuntime 策略。

### 部署步骤

克隆示例

git clone https://github.com/aws-samples/sample-ambient-agent.git cd sample-ambient-agent

部署后端（CDK）并构建/部署前端

cd backend pip install -r requirements.txt cdk bootstrap # 仅首次执行 cdk deploy --all

cd ../frontend npm install npm run build

将代理容器部署到 Bedrock AgentCore Runtime

cd ../agent ./deploy_agent.sh

### 监控与可观测性

Lambda 函数会在 AmbientAgents 命名空间下发出自定义 CloudWatch 指标，用于监控该模式中的关键事件：信号匹配、任务入队、任务完成、任务中断和任务错误。结合 Lambda 和 SQS 的默认指标（调用次数、错误数、队列深度、死信队列深度），这些指标可以生成一个仪表板，完整展示信号 → 任务 → 代理 → 人类的整个循环。将死信队列深度设置为警报，作为防止静默故障的第一道防线。

### 安全最佳实践

 /think

- IAM 最小权限：仅为每个 Lambda 函数授予其所需的权限（对其自身的表具有读取权限，针对 bedrock-agentcore:InvokeAgentRuntime 的定向调用，以及针对 Amazon S3 的特定路径前缀）。

- 静态加密：所有 DynamoDB 表使用 AWS 管理的密钥。Amazon S3 存储桶使用 S3 管理的加密。CloudFront 日志存储在专用的日志存储桶中。

- 传输加密：所有 API Gateway 端点和 CloudFront 分发都需要 HTTPS。Amazon S3 存储桶策略强制使用 SSL。

- 审计日志：AWS CloudTrail 记录整个堆栈中的 API 调用。可以在作业表上启用 DynamoDB 流，以获得更丰富的作业生命周期审计信息。

- 负责任的 AI：通过 Amazon Bedrock 防护措施对代理的输入和输出进行过滤，确保内容过滤、禁止主题和上下文关联检查在发现结果出现在作业页面或通过 ask_human 提示传递给审阅者之前完成。这保护了人机协作交互，并确保代理输出始终基于源文档。

### 扩展策略

- Lambda 并发：为作业执行工作线程的 Lambda 函数保留并发配额，以限制每个账户的下游 Bedrock 调用在可预测的上限内。

- DynamoDB 容量：按请求付费是默认模式。当流量模式趋于稳定时，可切换为带有自动扩展的预置容量。

- 成本优化：使用 DynamoDB TTL 自动淘汰旧对话，使用 Amazon S3 生命周期策略淘汰已处理文档，并使用 CloudWatch 指标过滤器跟踪每个代理模型调用的成本。

## 清理

评估完示例后，为避免持续产生费用，请按照与创建顺序相反的顺序拆除堆栈。

首先，从 Amazon Bedrock AgentCore 中删除代理运行时，使容器停止计费：

aws bedrock-agentcore delete-agent-runtime \ --agent-runtime-arn <your-agent-runtime-arn>

接下来，清空堆栈创建的 Amazon S3 存储桶（文档存储桶、前端托管存储桶和 CloudFront 日志存储桶），以便 CloudFormation 可以删除它们：

aws s3 rm s3://<amzn-s3-demo-documents-bucket> --recursive aws s3 rm s3://<amzn-s3-demo-frontend-bucket> --recursive aws s3 rm s3://<amzn-s3-demo-logs-bucket> --recursive

然后销毁 AWS CDK 堆栈本身，这将删除 Lambda 函数、API Gateway、CloudFront 分发、DynamoDB 表、SQS 队列、Cognito 用户池和 IAM 角色：

cd backend cdk destroy --all

最后，如果您不打算重新部署，删除由代理部署脚本创建的 Amazon ECR 仓库：

aws ecr delete-repository \ --repository-name ambient-agent \ --force

如果您仅为本次操作启用了 Amazon Bedrock 模型访问权限且不再需要，可以在 Amazon Bedrock 控制台的“模型访问”页面中撤销该权限。

## 总结

环境代理将 AI 自动化从“等待用户”转变为“响应信号”。操作响应时间从小时级缩短至秒级，全天候监控无需持续人工关注即可实现。每个代理独立运行，使用隔离会话，因此平台可以并行处理大量并发事件而无需协调开销。只有在真正需要人工输入时，人类才会介入流程。统一的作业页面消除了在多个工具之间切换上下文的需要，完整的对话历史记录确保您始终能追踪代理的行为及其暂停原因。

### 何时使用环境模式

以下用例描述了环境代理模式的一般应用场景。当前参考示例已实现 Amazon S3 和调度器的基础功能。Webhook、数据库变更和外部集成是需要在顶层进行扩展的连接点。

该模式适用于以下场景：当环境中发生的事件需要驱动模型可推理的工作流程，并且需要人类偶尔进行干预时。文档处理流水线是典型示例：文件到达后，代理会读取并总结文件内容，在采取行动前请求批准。监控和告警流程同样适用这种模式，代理会先对告警进行分类，仅在需要升级时请求人工介入。定时处理、分析和报告天然适合 cron 路径，任何需要审批环节或人工检查点的多步骤工作流都可以通过 ask_human 中断机制清晰映射。

某些工作负载使用环境代理并不是最佳选择。实时聊天应用应使用专为低延迟对话交互设计的传统聊天代理。简单的请求-响应 API 完全不需要代理。普通的 API 端点更快、更便宜且更容易理解。对于不需要大语言模型（LLM）推理的纯确定性工作流，AWS Step Functions 仍然是更优选择。需要注意的是，带有 autoExecute: true 配置的环境代理已经覆盖了完全自主运行的场景，如果您需要 LLM 推理但不需要人工干预环节。

### 下一步操作

克隆参考实现，按照顶层 README 的指引进行部署和配置。

进一步阅读：

- Amazon Bedrock AgentCore 文档
- LangChain 文档
- AWS CDK 文档

从小处着手。选择一个具体用例（文档处理或监控告警），然后逐步扩展。克隆示例代码，将其指向您自己的事件源，看看 Jobs 页面能为您减少多少重复性工作。

## 作者信息

'"`