AWS Lambda 文件系统上的 AI Agent 实践

你以前肯定写过这样的代码。一个 S3 事件被触发，你的 Lambda 函数被唤醒，它做的第一件事就是把文件下载到 /tmp。处理文件。上传结果。清理 /tmp，以免空间耗尽。管道中的每个文件、每次调用、每个函数都要重复这一过程。

S3 Files 改变了这一切。你可以将 S3 存储桶挂载为本地文件系统，然后在 Lambda 代码中直接使用 open()。我构建了一组 AI 代码审查智能体，它们通过挂载的 S3 存储桶共享工作区，并由一个持久函数进行编排，而文件访问代码是整个项目中最平淡无奇的部分。这恰恰就是它的意义所在。

如果你构建过任何需要在 Lambda 上访问 S3 数据的东西，就一定熟悉这种模式。你需要一个文件，但 S3 提供的并不是文件，而是对象。因此，你要先把对象下载到 /tmp，完成处理，再把结果上传回去。

# The old way: every Lambda developer has written this
import boto3

s3 = boto3.client("s3")

def lambda_handler(event, context):
    bucket = event["bucket"]
    key = event["key"]

# Download to /tmp
    local_path = f"/tmp/{key.split('/')[-1]}"
    s3.download_file(bucket, key, local_path)

# Do your actual work
    with open(local_path) as f:
        content = f.read()
    result = process(content)

# Upload the result
    s3.put_object(Bucket=bucket, Key=f"output/{key}", Body=result)

# Clean up so you don't fill /tmp
    os.remove(local_path)

为了“读取一个文件并写入一个文件”，却要走这么多流程。而当多个函数需要处理同一份数据时，情况会变得更糟。每个函数都要下载自己的副本，各自管理自己的 /tmp。如果你正在处理大型代码仓库或数据集，很快就会耗尽 /tmp 的 10GB 空间上限。

如果不提那些能减轻这种痛苦的库，那就是我对你的不负责任。s3fs 和 smart_open 等工具可以抽象掉其中一些步骤，但它们在底层仍然需要进行 API 调用。你的代码依旧是通过 SDK 与 S3 通信，而不是通过文件系统。

S3 Files 是一项新功能，它可以将 S3 存储桶作为本地文件系统挂载到 Lambda 函数上。你的代码可以在 /mnt/workspace 这样的挂载路径中读写文件，而 S3 Files 会负责将更改同步回存储桶。你写入的更改会在几分钟内出现在 S3 中，对 S3 对象所做的更改则会在几秒内反映到文件系统上。

# The new way: just file paths
from pathlib import Path

WORKSPACE = Path("/mnt/workspace")

def lambda_handler(event, context):
    # Read directly from the mount
    content = (WORKSPACE / "source" / "app.py").read_text()
    result = process(content)

# Write directly to the mount
    (WORKSPACE / "output" / "result.json").write_text(result)

访问文件时不再需要 boto3，不再需要管理 /tmp，也不再需要上传步骤。文件系统本身就是接口。

在底层，S3 Files 构建于 Amazon EFS 之上。它会将你的工作集缓存在高性能存储中，从而为活跃使用的数据提供亚毫秒级延迟。对于大型顺序读取，它会直接从 S3 流式传输数据。你可以同时获得文件系统语义、S3 的持久性以及经济性。

不过，有一点需要注意：S3 Files 要求使用 VPC。你的 Lambda 函数必须与挂载目标位于同一个 VPC 中，而且还需要通过 NAT 网关访问外部互联网。

坦白说，作为一个做无服务器架构的人，我通常会尽量避开 VPC。但这些年来，AWS 已经消除了其中的大部分障碍。连接 VPC 的 Lambda 函数不再像过去那样承受冷启动惩罚。网络配置只是编写一次的样板代码。考虑到 S3 Files 带来的能力，这个取舍是值得的。给自己准备一份可复用的网络模板，然后继续往下做就行。

我想用一些比“读取 CSV”更有意思的场景来测试 S3 Files，于是构建了一个无服务器代码审查系统。你只需向它提供一个公开的 GitHub 代码仓库，接下来就会发生三件事：

一个持久编排器函数将代码仓库克隆到共享的 S3 Files 工作区

一个安全审查智能体和一个代码风格审查智能体并行分析代码

结果以 JSON 文件的形式写入同一个工作区，并同步回 S3

三个 Lambda 函数全部挂载同一个 S3 存储桶。编排器写入文件，智能体读取文件。函数之间不需要传递 S3 键，也不需要将文件下载到 /tmp。文件系统就是协调层。

这些智能体使用 Strands Agents SDK 和 Amazon Bedrock。每个智能体都会获得一组自定义文件工具，用于操作挂载路径；Claude 会决定读取哪些文件、分析什么内容，以及写入什么结果。编排器使用 Lambda 持久函数协调工作流，并自动执行检查点保存。

完整源代码位于 GitHub：singledigit/lambda-s3-files-example

IaC 是迭代次数最多的部分。S3 Files 才刚刚推出，CloudFormation 资源类型尚未加入 linter。下面是我总结出的经验。

要让 S3 Files 与 Lambda 配合工作，你需要五种资源：

启用了版本控制的 S3 Bucket（必需）

供 S3 Files 访问存储桶的 IAM Role

在存储桶与 NFS 之间建立桥接的 S3 Files FileSystem

每个 AZ 中的 Mount Targets（网络端点）

用于控制 Lambda POSIX 身份的 Access Point

这些资源类型分别是 AWS::S3Files::FileSystem、AWS::S3Files::MountTarget 和 AWS::S3Files::AccessPoint。你的 IDE 中的 CloudFormation linter 目前还无法识别它们。忽略那些红色波浪线即可。

S3 Files IAM 角色信任的是 elasticfilesystem.amazonaws.com，而不是 s3files.amazonaws.com。这点曾让我踩坑。S3 Files 构建于 EFS 之上，因此其信任关系使用的是 EFS 服务主体。

S3FilesRole:
  Type: AWS::IAM::Role
  Properties:
    Path: /service-role/
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Sid: AllowS3FilesAssumeRole
          Effect: Allow
          Principal:
            Service: elasticfilesystem.amazonaws.com
          Action: sts:AssumeRole
          Condition:
            StringEquals:
              aws:SourceAccount: !Ref AWS::AccountId
            ArnLike:
              aws:SourceArn: !Sub 'arn:aws:s3files:${AWS::Region}:${AWS::AccountId}:file-system/*'

该角色需要拥有读取和写入存储桶的 S3 权限。通过 aws:ResourceAccount 条件，将权限范围限定到你的特定存储桶 ARN。

下面是与 Lambda 相关的重要部分。Access Point 控制函数运行时使用的 POSIX 身份，并创建一个可写的根目录。没有它，Lambda 虽然能挂载文件系统，却无法向其中写入内容。

S3FilesAccessPoint:
  Type: AWS::S3Files::AccessPoint
  Properties:
    FileSystemId: !GetAtt S3FileSystem.FileSystemId
    PosixUser:
      Uid: '1000'
      Gid: '1000'
    RootDirectory:
      Path: /lambda
      CreationPermissions:
        OwnerUid: '1000'
        OwnerGid: '1000'
        Permissions: '755'

CreationPermissions 属性至关重要。它会在客户端首次连接时，自动创建具有正确所有权的 /lambda 目录。如果没有它，根目录将归 root（UID 0）所有，而 Lambda 通过 Access Point 以 UID 1000 运行，因此无法创建子目录。

我喜欢遵循的经验法则是（如果你用到了它们）：如果使用 EFS Access Point，就使用 CreationInfo；如果使用 S3 Files Access Point，就使用 CreationPermissions。概念相同，只是属性名称不同。

在 Lambda 端，FileSystemConfigs 接收的是 Access Point ARN，而不是 FileSystem ARN，此外还需要一个本地挂载路径：

OrchestratorFunction:
  Type: AWS::Serverless::Function
  DependsOn:
    - MountTargetA
    - MountTargetB
  Properties:
    FileSystemConfigs:
      - Arn: !GetAtt S3FilesAccessPoint.AccessPointArn
        LocalMountPath: /mnt/workspace
    VpcConfig:
      SecurityGroupIds:
        - !GetAtt NetworkingStack.Outputs.LambdaSGId
      SubnetIds:
        - !GetAtt NetworkingStack.Outputs.PrivateSubnetAId
        - !GetAtt NetworkingStack.Outputs.PrivateSubnetBId

针对 Mount Targets 设置 DependsOn 很重要。在 Mount Targets 可用之前，Lambda 无法挂载文件系统，而创建它们大约需要五分钟。

对于 IAM，Lambda 需要针对 FileSystem ARN 的 s3files:ClientMount 和 s3files:ClientWrite 权限，而不是 elasticfilesystem:ClientMount。即使信任策略使用 EFS 主体，IAM 操作仍然使用 s3files 命名空间。我知道，这毕竟是一项新服务。

用于持久函数的异步 API Gateway

这里不应该使用同步 API 调用。编排器会启动可能运行数分钟的智能体，你不希望它在智能体工作期间一直空转，那是在浪费钱。你也不希望调用方一直盯着加载动画。启动任务，然后继续处理其他事情。异步才是正确的方式。

诀窍是在 OpenAPI 定义中将 X-Amz-Invocation-Type 请求头设置为 Event：

x-amazon-apigateway-integration:
  type: aws
  httpMethod: POST
  uri: !Sub 'arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${OrchestratorFunction.Alias}/invocations'
  credentials: !GetAtt ApiGatewayRole.Arn
  requestParameters:
    integration.request.header.X-Amz-Invocation-Type: "'Event'"

API 会立即返回 202。持久函数则在后台运行。你可以到 S3 存储桶中查看结果。

编排器是一个负责协调流水线的持久函数。它会运行三个步骤：克隆仓库、并行调用两个审查智能体，以及写入汇总摘要。

克隆步骤使用标准 Python 下载 GitHub tarball，并将其解压到挂载的工作区中。不需要 boto3，只使用 tarfile、pathlib 和 requests。

@durable_step
def clone_repo(step_ctx: StepContext, repo_url: str, review_id: str):
    """Download a public GitHub repo tarball and extract to the workspace."""
    tarball_url = repo_url.rstrip("/") + "/archive/refs/heads/main.tar.gz"
    response = requests.get(tarball_url, timeout=120, stream=True)

if response.status_code == 404:
        tarball_url = repo_url.rstrip("/") + "/archive/refs/heads/master.tar.gz"
        response = requests.get(tarball_url, timeout=120, stream=True)

response.raise_for_status()

source_dir = Path(WORKSPACE) / review_id / "source"
    source_dir.mkdir(parents=True, exist_ok=True)

tarball_bytes = io.BytesIO(response.content)
    with tarfile.open(fileobj=tarball_bytes, mode="r:gz") as tar:
        for member in tar.getmembers():
            if not member.isfile():
                continue
            # Extract code files to the mounted workspace
            dest = source_dir / relative_path
            dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_bytes(tar.extractfile(member).read())

这个 source_dir.mkdir() 调用会写入 /mnt/workspace/{repo}/source/。这里就是 S3 Files 挂载点。文件会写入文件系统，并自动同步回 S3 存储桶。

并行步骤使用 context.invoke() 同时调用两个智能体函数：

@durable_execution
def handler(event: dict, context: DurableContext) -> dict:
    # Step 1: Clone the repo
    clone_result = context.step(clone_repo(repo_url, review_id))

# Step 2: Run reviews in parallel
    def security_review(ctx: DurableContext):
        return ctx.invoke(SECURITY_AGENT_ARN, {"review_id": review_id}, name="security-review")

def style_review(ctx: DurableContext):
        return ctx.invoke(STYLE_AGENT_ARN, {"review_id": review_id}, name="style-review")

results = context.parallel(
        [security_review, style_review],
        name="parallel-reviews",
        config=ParallelConfig(max_concurrency=2),
    )

如果编排器在克隆完成后、审查开始前被中断，持久函数会从检查点继续执行，而不会重新克隆。这正是持久函数的核心理念。

每个审查智能体都是一个 Strands 智能体，并配有三个操作 S3 Files 挂载点的自定义工具：list_files、read_file 和 write_review。智能体自行决定读取哪些文件、分析什么内容，以及何时写入审查结果。

@tool
def list_files(path: str = ".") -> str:
    """List files and directories at a path in the source code workspace."""
    source_dir = Path(WORKSPACE) / review_id / "source"
    target = source_dir / path
    entries = []
    for item in sorted(target.iterdir()):
        if item.is_dir():
            entries.append(f"  [dir]  {item.name}/")
        else:
            entries.append(f"  [file] {item.name} ({item.stat().st_size} bytes)")
    return "\n".join(entries)

@tool
def read_file(path: str) -> str:
    """Read the contents of a source code file from the workspace."""
    source_dir = Path(WORKSPACE) / review_id / "source"
    return (source_dir / path).read_text(encoding="utf-8", errors="ignore")

@tool
def write_review(filename: str, content: str) -> str:
    """Write review findings to a JSON file in the reviews directory."""
    reviews_dir = Path(WORKSPACE) / review_id / "reviews"
    reviews_dir.mkdir(parents=True, exist_ok=True)
    (reviews_dir / filename).write_text(content)
    return f"Written to {reviews_dir / filename}"

每个工具都只使用 pathlib 和 open()。智能体从 /mnt/workspace/{repo}/source/ 读取文件，并写入 /mnt/workspace/{repo}/reviews/。这些路径都位于 S3 Files 挂载点上。智能体既不知道，也不关心自己实际上是在与 S3 交互。

处理程序创建一个带有系统提示词的 Strands 智能体，并让它开始运行：

model = BedrockModel(model_id=MODEL_ID, max_tokens=4096)

agent = Agent(
    model=model,
    tools=[list_files, read_file, write_review],
    system_prompt=SECURITY_SYSTEM_PROMPT,
)

response = agent(
    f"Review the source code for security issues. "
    f"Start by listing the root directory, then read the key files "
    f"and write your findings to 'security.json'."
)

Claude 会自主探索代码库。它会列出目录、读取感兴趣的文件，并写入结构化的 JSON 审查结果。安全智能体会查找硬编码密钥、注入漏洞和权限过于宽松的 IAM 策略。代码风格智能体则会检查命名约定、文档缺失和代码组织方式。

两个智能体会在同一个挂载工作区上并行运行。一个写入 security.json，另一个写入 style.json。除了文件系统之外，不需要任何协调机制。

完整源代码已发布在 GitHub 上，README 中包含部署说明。克隆代码后运行 sam build --use-container && sam deploy --guided，大约 15 分钟即可部署完整技术栈。将它指向任意公开的 GitHub 仓库，然后到 S3 存储桶中查看结果。

何时适合在 Lambda 中使用 S3 Files

当 Lambda 函数需要共享访问存储在 S3 中的数据时，S3 Files 非常适用。例如，多个函数读写相同的文件；需要探索目录树的智能体；要求使用文件路径而不是字节流的库；以及数据已经存储在 S3 中、你又不想将其复制到其他位置的工作负载。

如果你只需要处理来自 S3 事件通知的单个小文件，S3 Files 就没那么理想。如果函数只是下载一个对象、处理它，然后上传结果，那么传统的 boto3 模式更简单。你不需要为此配置 VPC、挂载目标或接入点。

成本构成很直观。你需要为 S3 Files 高性能存储（即活跃工作集）、同步过程中的 S3 请求以及标准 VPC 成本付费，其中 NAT 网关是最大的成本项。对于原本需要在 S3 和 EFS 之间复制数据的工作负载，S3 Files 最多可节省 90% 的成本，因为你不再需要维护一套独立的文件系统。

有一个值得特别指出的生产环境注意事项：S3 Files 会在几分钟内将写入内容同步回 S3，而不是在几毫秒内完成。如果另一个系统需要在 Lambda 写入结果后立即读取，就必须将这段延迟考虑在内。对于我们的代码审查场景，这没有问题。编排器会在两个智能体都完成后写入摘要，而用户会在方便时查看存储桶。

别再写样板代码了

多年来，对于“能否在 Lambda 中挂载 S3？”这个问题，答案一直只有“不行”。现在可以了。而实现它所需的代码，反而是函数中最不起眼的部分。

你的智能体可以像读取本地文件一样读取文件。你的流水线可以共享一个工作区，而无须到处传递 S3 键。那些要求使用文件路径的库也能直接工作。

完整源代码已发布在 GitHub 上。克隆、部署，然后让它审查你自己的仓库。SAM 模板会处理一切：VPC、S3 Files、挂载目标、接入点、持久编排器和 Strands 智能体。

别再把文件从 S3 下载到 /tmp 了。直接挂载你的存储桶。

如果你正在使用 AI 编程助手，下面这些资源涵盖了本文涉及的技术：

Kiro Powers（从 Kiro Powers 面板安装）：

aws-sam——SAM 模板编写、部署和最佳实践

aws-lambda-durable-functions——持久函数模式、步骤操作、并行执行和部署

strands——使用 Strands SDK、自定义工具和 Bedrock 集成构建智能体

获取本文配套的智能体文件。你可以将其用作 Kiro steering、CLAUDE.md，或任何其他智能体指令格式。