在AWS Bedrock上构建生产级AI代理——值得关注的架构与代码决策
背景
模型是无状态的。它们处理一个请求，生成文本结果，然后遗忘。它们不会执行操作，默认不会与你的数据集成，也不会自主协调多步骤工作流。智能体通过将模型封装在运行时系统中来解决这个问题，该系统为模型提供工具、记忆和推理循环。
这个应用是一个无服务器租赁管理机器人——房东通过Telegram和电子邮件发送自然语言消息并接收通知。最棒的是，当前付款、债务和趋势的分析由Claude驱动的智能体完成。它能为你提供数据洞察，记住你的偏好，是一个真正的协作者。这就是每个智能体都能做到的事情。聊天机器人只是文本，没有与你的数据或系统的深度集成——从定义上来说，能力更弱。
GitHub代码：https://github.com/jorgetovar/whatsapp-rental-manager
智能体容易构建但并不简单。我们应该牢记在后端系统和生产应用中一直做的一切：安全性、集成、内聚性、耦合性、数据管理、可扩展性、可靠性、成本、运维卓越性、延迟和性能——所有非功能性需求仍然适用。
架构经验
1. 使用Bedrock而非直接调用模型API
运维层面的答案：IAM认证。没有需要轮换、泄露或存储在Secrets Manager中的API密钥。你用于DynamoDB的同一个boto3会话模式也适用于Bedrock。跨区域推理（us.anthropic.claude-sonnet-4-6）会自动路由到延迟最低的美国区域，并提供比单区域端点更高的吞吐量限制。
def create_agent(model_id: str = None, extra_context: str = "") -> Agent:
model_id = model_id or os.environ.get("BEDROCK_MODEL_ID", "us.anthropic.claude-sonnet-4-6")
boto_session = boto3.Session(
profile_name=os.environ.get("AWS_PROFILE"),
region_name=os.environ.get("AWS_REGION_NAME", "us-east-1"),
)
model = BedrockModel(model_id=model_id, boto_session=boto_session)
return Agent(model=model, system_prompt=SYSTEM_PROMPT + extra_context, tools=[...])2. 智能体记忆
Lambda是无状态的。每次调用都是冷启动。为了让Claude记住对话历史，历史记录必须持久化到外部存储，并在每次请求时重新加载。
我使用DynamoDB来跟踪消息，但有更好的选择——AWS Strands内置了会话处理功能，可以总结历史记录或保留精确数量的消息，这正是我们当前的做法。
_MAX_MESSAGES = 20 # 约10轮对话——足够的上下文，安全的DynamoDB项目大小
_TTL_SECONDS = 86400 # 24小时——对话隔夜重置
def save_history(chat_id: str, messages: list) -> None:
trimmed = _sanitize_messages(_strip_metadata(messages)[-_MAX_MESSAGES:])
get_table("conversations").put_item(Item={
"chatId": str(chat_id),
"messages": json.dumps(trimmed),
"ttl": int(time.time()) + _TTL_SECONDS,
})DynamoDB TTL自动完成清理——无需cron作业，无需成本。
3. 动态上下文注入 vs. 对话历史
有两种方式向模型提供当前状态：在每次请求时注入到系统提示中，或让其在对话历史中累积。
对话历史对于状态来说不可靠。它会被截断。它包含噪声。模型在长上下文中的注意力会衰减。如果房东在三轮对话前添加了新房产，你无法保证模型现在还记得。
有效的模式：
def _build_user_context(caller: dict) -> str:
landlord = get_table("landlords").get_item(Key={"landlordId": landlord_id}).get("Item")
active = sum(1 for l in leases if l.get("status") == "active")
return (
"
用户状态（系统固定）：
"
f"- 角色：房东 — {name}
"
f"- 活跃房产：{active}
"
f"- 已接受数据政策。不要重复欢迎信息。"
)
agent = create_agent(extra_context=user_ctx)
agent.messages = load_history(chat_id)
response = agent(text)4. 工具设计：文档字符串即API契约
使用Strands SDK，工具是用@tool装饰器装饰的Python函数。文档字符串不是文档——它是与模型的API契约。要明确具体。这能引导智能体朝正确方向前进，产生可靠的工具调用。
@tool
def log_payment(
tenant_name_or_phone: str,
amount: int,
period: str = "",
method: str = "other",
) -> str:
"""记录从租户收到的付款。
示例：tenant_name_or_phone='Juan', amount=800000。
period格式为YYYY-MM（空=当月）。
method: nequi/daviplata/efectivo/other
"""
...当用户说"Juan付了80万"时，Claude推断出log_payment(tenant_name_or_phone="Juan", amount=800000)，因为文档字符串明确说明了每个参数的含义。模糊的文档字符串会产生错误的工具调用。
5. 智能体安全：模型不是信任边界
智能体设计中最危险的模式：让模型决定用户是谁或他们被允许做什么。
在这个系统中，身份来自电话号码，Telegram在每个webhook中提供该号码。角色通过电话号码从DynamoDB解析——绝不来自模型输入或输出。
def resolve_caller_sync(phone: str) -> list[CallerContext]:
contexts: list[CallerContext] = []
# 检查是否已注册的房东
resp = landlords_table.get_item(Key={"landlordId": phone})
if resp.get("Item"):
contexts.append(CallerContext(role="landlord", landlordId=phone))
# 检查是否活跃租户
for item in paginated_query(leases_table, IndexName="tenantPhone-index", ...):
if item.get("status") == "active":
contexts.append(CallerContext(role="renter", landlordId=item["landlordId"], ...))
return contexts这与传统后端应用的原则相同——JWT令牌在每个请求中传递，你从中提取声明来获取customer_id——你永远不会从请求体中接受它，因为那将是一个欺骗漏洞。
Webhook请求在处理之前也会通过时间安全的HMAC检查进行验证：
def _verify_signature(secret: str, header: str) -> bool:
return hmac.compare_digest(secret, header or "")hmac.compare_digest无论字符串在何处不同都花费恒定时间——攻击者无法通过测量响应延迟来逐字节探测密钥。
6. 基础设施即代码
每个Lambda函数获得恰好所需的权限——逾期租户提醒函数读取租赁和租户信息；它不能写入。部署只需一个命令即可完成。任何人都可以轻松复制这个应用。
策略：
- DynamoDBReadPolicy： # 提醒Lambda只读 TableName: !Ref LeasesTable
- DynamoDBCrudPolicy： # webhook Lambda完全CRUD TableName: !Ref PaymentsTable
- !Ref SesPolicy # 所有提醒函数共享的托管策略
Secrets Manager动态引用在部署时解析：
TELEGRAM_BOT_TOKEN: !Sub '{{resolve:secretsmanager:miarriendobot/telegram:SecretString:token}}'注意：密钥值在部署时被烘焙到Lambda环境变量中。在Secrets Manager中轮换密钥不会自动更新Lambda——你必须重新部署或直接更新环境变量。
7. 可观测性
Agent 并非确定性的。现在比以往任何时候都更需要理解系统中发生了什么——集成点在哪里、是否有故障、如何获得通知。设置告警、链路追踪和良好的日志始终至关重要。
一个非显而易见的缺口：Webhook Lambda 必须始终返回 HTTP 200——Telegram 会在其他任何响应上无限期重试投递。副作用是，即使每次调用都抛出异常，API Gateway 看起来也始终健康。标准的可用性监控对此完全失效。
WebhookErrorAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
MetricName: Errors
Namespace: AWS/Lambda
Dimensions:
- Name: FunctionName
Value: !Ref WebhookFunction
Statistic: Sum
Period: 300
EvaluationPeriods: 1
Threshold: 3
ComparisonOperator: GreaterThanOrEqualToThreshold
AlarmActions:
- !Ref AlertTopic针对 Lambda Errors 指标的 CloudWatch 告警是唯一真正的信号。没有它，你只能通过用户投诉而非告警页面发现机器人已损坏。
8. DynamoDB 数据模型设计
反范式化决策是一个真正的权衡。即使 Property 也包含 canon，但将其直接存储在 Lease 上意味着提醒 Lambda 只需获取一个条目就能拥有所有信息——无需二次读取，无需关联。
class Lease(BaseModel):
leaseId: str
landlordId: str
propertyId: str
tenantPhone: str
canon: int # 从 Property 复制——有意为之：提醒只需一次读取
dueDay: int
startDate: str
status: str = "active"显式成本：update_canon 必须写入两个表。DynamoDB 不是关系型数据库——请根据读取模式设计，而非规范化。
9. 可测试性
任何依赖于日历日期的工作在 CI 中都无法测试，除非有绕过参数。job_monthly_summary(force=True) 会立即运行，无论日期如何。从第一天起就为可测试性设计。
def job_late_tenant_reminder(force: bool = False):
today = datetime.now(timezone.utc)
for lease in _get_all_active_leases():
reminder_date = _due_date_this_month(due_day) + timedelta(days=5)
if not force and today.date() != reminder_date.date():
continue # 跳过——不是正确的日期
# 发送提醒...同一个函数在本地通过 APScheduler 运行，在生产中通过 EventBridge 运行——无需模拟，完全一致。
10. AWS 凭证——在两个环境中都有效的一行代码
boto3.Session(profile_name=os.environ.get("AWS_PROFILE"))这一行代码在本地开发和 Lambda 中都能工作，无需任何分支。关键在于 os.environ.get("AWS_PROFILE")——当 Lambda 中未设置 AWS_PROFILE 时，.get() 返回 None，而 boto3.Session(profile_name=None) 会正确回退到通过 IMDS 获取的 IAM 执行角色。
需要避免的两个错误：
os.environ.get("AWS_PROFILE", "")返回""而非None——boto3 随后会尝试查找名为""的配置文件并静默失败。- 切勿将
AWS_PROFILE设置为 Lambda 环境变量——boto3 会查找 Lambda 运行时中不存在的凭证文件。
本地，你的 .env 设置 AWS_PROFILE=default-c1，并使用命名配置文件。在 Lambda 中，该变量不存在，执行角色自动接管。无需 if local else lambda 分支，无需硬编码配置文件名称，无需在代码中包含凭证。
11. 上下文是任何 Agent 最重要的部分
工具和良好的上下文使 Agent 强大。我们传递足够但重要的信号，以实现连贯的对话。模型不需要一切；它需要在正确的时间获得正确的信息。
- 角色: 房东 — Jorge Tovar
- 活跃房产: 2
- 已接受数据政策。不要重复欢迎语。模型知道用户的角色、当前状态以及不应询问的内容。这就是它正确行为所需的全部上下文。
12. 切勿在工具调用中间截断对话历史
始终在完整交换边界处截断——绝不要在 toolUse/toolResult 对中间截断。我们曾未实现此功能就发布，结果 Bedrock 在第 10 轮对话时抛出 ValidationException。修复方法是在截断列表后遍历并删除尾部任何不完整的交换：
def _sanitize_messages(messages: list) -> list:
# 丢弃仅包含 toolResult 块的起始用户消息
# （其对应的助手 toolUse 已被截断）
while start < len(messages):
msg = messages[start]
has_text = any(isinstance(b, dict) and "text" in b for b in msg.get("content", []))
if msg.get("role") != "user" or has_text:
break
start += 1
# 丢弃任何未被下一条消息的 toolResult 匹配的助手 toolUse
if tool_use_ids and not tool_use_ids.issubset(result_ids):
break # 缺少 toolResult——从此处丢弃完整推理循环的结束是唯一安全的截断点。
13. Hook 是首选且确定性的验证方式
模型不应负责那些可以在代码中确定性验证的事情。请求作用域的上下文在每次调用开始时设置一次，并自动传播到每个工具——无需线程，安全，因为 Lambda 在每个执行环境中一次只处理一个请求。
# lambda_handler.py — 在每次 agent 调用前设置
CURRENT_CONTEXT.clear()
CURRENT_CONTEXT.update(caller.model_dump())
# 任何工具——从同一个字典读取
def _landlord_id() -> str:
return CURRENT_CONTEXT.get("landlordId", "")模型接收该字符串，理解它是一个业务规则，并自然地响应用户——“抱歉，该功能仅对房东可用。”
14. 结构化错误响应很重要
明确错误是可重试的还是业务规则违规。模型以不同方式处理它们，用户也是如此。
结论
在 AWS Bedrock 上构建 Agent 与构建任何生产后端系统并无本质区别——非功能性需求不会因为中间有一个语言模型而消失。安全性、可观测性、可测试性和数据完整性同样重要，在某些情况下甚至更重要，因为 Agent 的行为比确定性代码更难预测。
代码是开源的——如果你正在构建类似的东西，这个架构可供你采用和调整。
现在从未有过更好的时机成为一名工程师，并通过软件为社会创造价值。
- GitHub
如果你喜欢这些文章，请访问我的博客 jorgetovar.dev。
