用 Amazon Bedrock 与 AWS Lambda 构建受治理的智能应用部署器

2026-08-07 51 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

让非技术员工用自然语言描述需求,然后在数秒内得到一个已经部署、支持多租户的 Web 应用,这不只是“让大模型写代码”。PDI Technologies 构建的 PDI Brew 更接近一套智能交付系统:Amazon Bedrock 负责理解意图并生成结构化计划,AWS Lambda 中的配置代理负责把计划转换成受约束、可审计的云资源。

真正值得关注的,是它把生成式 AI 放在了一个有明确边界的部署流程里。模型可以规划,但不能随意创建资源;代理可以执行,但只能调用经过批准的模板与服务。

从一句需求到可执行部署计划

用户输入可能非常模糊,例如:

创建一个门店检查工具,让区域经理填写检查结果,并按门店查看历史记录。

部署系统不能直接把这句话交给具有管理员权限的代理。更稳妥的做法是先经过一个可插拔的规划器,将自然语言转换成结构化规范:

{
  "application_type": "inspection_tracker",
  "tenant_mode": "isolated_by_tenant_id",
  "roles": ["regional_manager", "administrator"],
  "entities": [
    {
      "name": "inspection",
      "fields": ["store_id", "score", "notes", "created_at"]
    }
  ],
  "capabilities": ["create_record", "list_history"],
  "deployment_profile": "serverless-standard"
}

结构化计划是系统的关键边界。它可以被 JSON Schema、业务规则和权限策略逐项校验,也能保存下来用于审批、审计和故障重放。

“可插拔规划器”还意味着意图理解与资源配置不必绑定。团队可以更换 Bedrock 模型、调整提示词,或者针对不同应用类型增加专用规划器,而不必改写底层配置逻辑。

Lambda 配置代理应该执行模板,而不是自由发挥

配置代理的职责不是生成任意 CloudFormation,而是从获批的部署目录中选择模板,并填入经过校验的参数。例如,一个应用类型可以映射到固定的前端、API、身份认证和数据存储组合。

下面是一个可以改造的最小 Python Lambda 示例。这里明确做出一个实践假设:CloudFormation 模板已经由平台团队审核并存放在受控的 S3 存储桶中,Bedrock 只负责把自然语言转换为部署计划。

运行前需要设置三个环境变量:BEDROCK_MODEL_IDTEMPLATE_BUCKETSTACK_EXECUTION_ROLE_ARN。Lambda 执行角色还需要调用 Bedrock、读取指定 S3 对象、创建 CloudFormation Stack,以及向 CloudFormation 传递执行角色的权限。

import json
import os
import re
import uuid

import boto3

bedrock = boto3.client('bedrock-runtime')
cloudformation = boto3.client('cloudformation')

ALLOWED_APP_TYPES = {
    'inspection_tracker': 'templates/inspection-tracker.yaml',
    'request_portal': 'templates/request-portal.yaml',
}


def plan_application(description):
    prompt = f'''You are an application planner.
Return JSON only with this shape:
{{
  "application_type": "inspection_tracker or request_portal",
  "tenant_name": "lowercase short name",
  "display_name": "human-readable name"
}}

User request:
{description}
'''

    response = bedrock.converse(
        modelId=os.environ['BEDROCK_MODEL_ID'],
        messages=[
            {
                'role': 'user',
                'content': [{'text': prompt}],
            }
        ],
        inferenceConfig={'temperature': 0, 'maxTokens': 300},
    )

    text = response['output']['message']['content'][0]['text']
    return json.loads(text)


def validate_plan(plan):
    app_type = plan.get('application_type')
    tenant_name = plan.get('tenant_name', '')

    if app_type not in ALLOWED_APP_TYPES:
        raise ValueError('Unsupported application type')

    if not re.fullmatch(r'[a-z0-9-]{3,30}', tenant_name):
        raise ValueError('Invalid tenant name')

    return plan


def lambda_handler(event, context):
    description = event.get('description', '').strip()
    if not description:
        return {'statusCode': 400, 'body': 'description is required'}

    plan = validate_plan(plan_application(description))
    app_type = plan['application_type']
    tenant_name = plan['tenant_name']
    stack_name = f'app-{tenant_name}-{uuid.uuid4().hex[:8]}'

    cloudformation.create_stack(
        StackName=stack_name,
        TemplateURL=(
            f'https://{os.environ["TEMPLATE_BUCKET"]}.s3.amazonaws.com/'
            f'{ALLOWED_APP_TYPES[app_type]}'
        ),
        Parameters=[
            {'ParameterKey': 'TenantId', 'ParameterValue': tenant_name},
            {
                'ParameterKey': 'ApplicationName',
                'ParameterValue': plan.get('display_name', tenant_name),
            },
        ],
        RoleARN=os.environ['STACK_EXECUTION_ROLE_ARN'],
        Capabilities=['CAPABILITY_NAMED_IAM'],
        Tags=[
            {'Key': 'ManagedBy', 'Value': 'agentic-deployer'},
            {'Key': 'TenantId', 'Value': tenant_name},
        ],
    )

    return {
        'statusCode': 202,
        'body': json.dumps(
            {'stack_name': stack_name, 'plan': plan},
            ensure_ascii=False,
        ),
    }

可以用下面的测试事件调用该函数:

{
  "description": "Create a store inspection tracker for the north region"
}

这个示例刻意限制了模型的权限:模型只能从两个应用类型中选择,租户名称必须通过正则校验,实际基础设施来自固定模板。生产环境还应使用 JSON Schema 或 Bedrock 的结构化输出能力约束响应,避免依赖自由文本中的 JSON。

多租户不是加一个 tenant_id 就结束了

PDI Brew 交付的是多租户应用,因此租户边界必须贯穿身份、数据、日志和成本归属。常见方案是在认证令牌中加入租户声明,并要求每一次数据访问都从受信任的身份上下文获取 tenant_id,不能接受客户端任意提交的租户标识。

以 DynamoDB 为例,可以把租户放进分区键:

from boto3.dynamodb.conditions import Key


def list_inspections(table, tenant_id):
    response = table.query(
        KeyConditionExpression=Key('pk').eq(f'TENANT#{tenant_id}')
    )
    return response.get('Items', [])

这里的 tenant_id 应来自经过验证的 JWT 声明或 API Gateway Authorizer 上下文。若直接使用请求体中的值,攻击者可能尝试读取其他租户的数据。

多租户还需要明确隔离等级:

  • 共享计算与共享数据表,依靠租户键做逻辑隔离,成本最低但策略要求严格。
  • 共享计算但每个租户使用独立数据资源,隔离更强,资源数量和运维成本更高。
  • 每个租户独立部署完整栈,边界最清晰,但部署速度、配额和成本管理更复杂。

智能部署器可以根据数据敏感度、租户规模和合规要求选择部署配置,但可选范围应由平台团队预先定义,而不是让模型自行发明隔离策略。

治理能力决定它能否进入生产环境

自然语言降低了创建应用的门槛,也会显著提高资源扩张速度。平台至少要在配置代理周围建立以下控制:

  • 使用允许列表限制可部署的应用类型、AWS 服务、区域和实例规格。
  • 为 CloudFormation 配置独立执行角色和权限边界,避免 Lambda 持有长期管理员权限。
  • 在执行前校验计划,在高风险变更前加入人工审批。
  • 为每次请求保存原始意图、模型输出、校验结果、模板版本和 Stack ID。
  • 设置租户级预算、资源配额、超时与自动回收策略。
  • 对提示词注入、异常参数、跨租户访问和重复请求进行测试。
  • 为创建过程增加幂等键,避免重试产生多套相同资源。

还应把“应用已创建”和“应用可以使用”区分开。CloudFormation 接受请求只代表开始配置。更完整的工作流可以通过 Step Functions 等编排机制等待栈完成、执行冒烟测试、写入应用目录,然后再向用户返回访问地址。

落地时从有限目录开始

构建这类平台时,最稳妥的起点不是支持任意应用,而是选择两三个重复率高、数据模型明确的内部工具,例如检查表、审批入口或请求跟踪器。

先为每类工具建立受控模板、参数 Schema、租户隔离策略和自动测试,再让 Bedrock 负责意图分类与参数提取。随着成功率和治理能力提升,可以逐步扩充应用目录和规划器能力。

这套架构的核心分工很清楚:自然语言负责表达目标,模型负责生成受约束的计划,Lambda 代理负责执行经过批准的动作,而 AWS 权限、模板和审计系统负责守住边界。只有把这四层分开,智能应用部署器才能同时获得速度、可维护性和生产环境需要的控制力。


相关推荐