基于 AWS Bedrock 构建多代理音乐生成流水线

随着组织从单用途智能体过渡到多智能体系统，基础设施需求也随之改变。一个独立智能体处理客户查询可以在无服务器环境中运行，会话生命周期很短。但当你需要三个智能体协作完成一个持续数天的创意工作流时，它们需要共享上下文并建立在彼此的输出之上——仅有几小时上限的无服务器会话就无法满足需求了。

在本文中，我们将带你完成一个音乐制作流水线的部署：一个智能体在实例自有 GPU 上运行生成式音频模型。另外两个智能体从共享卷中打开它写入的 .wav 文件。到最后，你会得到一首可以播放的曲目。你还将学会如何创建容量提供者、从不同制品类型部署智能体、使用共享会话编排智能体间的协作，以及跨多天持久化工作流。

Amazon Bedrock AgentCore 为托管智能体提供两种计算选项。MicroVM 是无服务器选项：冷启动快速、会话隔离、按量计费。Runtime Instances 是新选项：AWS 托管的 EC2 基础设施，适用于持久化、长时间运行的智能体工作流。两者使用相同的运行时 API，但 Instances 增加了多天会话、GPU、持久卷，以及在同一实例上共存多个智能体的能力。

Runtime Instances 与 MicroVM 的区别

两个选项都支持自定义框架（CrewAI、LangGraph、LlamaIndex、Strands Agents），可选择任意基础模型，与 MCP 和 A2A 集成，并且共享相同的 AgentCore 运行时 API。区别在于底层计算模型。

智能体是在会话中运行的工作负载。与每个运行时只托管一个智能体的 MicroVM 模型不同，单个 Instances 会话可以托管多个智能体。当两个智能体运行时共享同一容量提供者时，你可以使用相同的 runtimeSessionId 调用它们，使两个智能体落在同一个 EC2 实例上。在那里，它们共享文件系统并可以协作完成同一任务。

图 1：三个智能体流水线（Compose、Deliver、Screen）共享一个 GPU 实例和会话

我们将构建一个使用三个专业智能体的音乐制作系统：

Composition 智能体（Audio AI 团队）：使用 Claude Sonnet 4.6 将制作人的请求转化为音乐简报，然后使用生成式音乐模型（ACE-Step，一个用于音乐生成的开源基础模型 FM）在实例自有 GPU 上渲染实际音频。以 Amazon Elastic Container Registry（Amazon ECR）中的容器镜像形式打包。

Delivery 智能体（Audio Engineering 团队）：从共享文件系统读取渲染好的曲目并进行测量，然后基于这些测量值（而非音频本身）向 Claude Sonnet 4.6 请求一条处理链（EQ、压缩、限制）。真实的信号处理应用到该链上，结果会再次测量以确认达到了交付目标。以 Amazon Elastic Container Registry（Amazon ECR）中的容器镜像形式打包。

Compliance 智能体（Release Engineering 团队）：独立地重新测量完成的交付物，检查 Delivery 智能体声称的交付目标，并对照工作室自身目录进行谐波相似性筛查。如果筛查发现问题，智能体会回调 Composition 智能体生成替代版本并重新筛查。以 Amazon Simple Storage Service（Amazon S3）上的 zip 文件形式交付。

工作流：制作人发起一首曲目的制作。Composition 智能体编写简报并在实例 GPU 上渲染真实音频。Delivery 智能体打开该文件，测量它，应用从测量结果推导出的处理链，再次测量以证明结果命中目标。然后 Compliance 智能体独立重新测量，检查交付目标，并对照工作室目录对音频进行筛查。如果筛查标记出匹配项，它回调 Composition 智能体生成替代版本。最终制作人得到一个可播放的 .wav 文件和三份解释每项决策的报告。

使这一切在 Runtime Instances 上成为可能的关键特性：

通过共享会话 ID 实现共存。 每个智能体有自己的运行时，但通过在同一容量提供者上使用相同的 runtimeSessionId 调用它们，AgentCore 将它们放在同一实例上并挂载相同的卷。它们共享文件系统，可以访问彼此的输出。

可使用的 GPU。 Composition 智能体直接在实例的 NVIDIA L4 上运行 ACE-Step 基础模型，约 9 秒内渲染 20 秒 48 kHz 立体声音频。模型及其依赖保存在持久卷上。在会话中构建一次，然后在会话中的每次调用中复用，包括隔夜停止后。

独立部署。 每个团队按自己的节奏推送自己的制品。Audio AI 团队推送新的 Composition 镜像无需与 Audio Engineering 或 Release Engineering 协调，另外两个团队继续不受影响地运行。

多天持久化。 制作人周一进行作曲，停止会话过夜，周二恢复交付工作。实例自动空闲并在下次调用时恢复。

混合制品。 ECR 中的容器和 S3 中的代码包可以在同一容量提供者上共存。团队选择适合其工作流的打包方式。

从现在开始，本文是实践导向的。你将准备 AWS 账户，然后运行一个三智能体流水线，渲染、生成并清理一首完成的曲目，产生一个你可以播放的 .wav 文件。按顺序完成这些步骤。首先确认前提条件，然后定义三个智能体（步骤 1），创建配置 GPU 实例及其持久卷的容量提供者（步骤 2），将每个智能体部署为独立的运行时（步骤 3），并使用共享会话 ID 调用它们，使它们可以在同一实例上共存并相互传递工作（步骤 4）。最后，步骤 5 展示任何一个团队如何在不干扰其他团队的情况下发布其智能体的新版本。完整示例在 AgentCore samples GitHub 仓库中。

在开始之前，确保你拥有：

• 一个 AWS 账户，拥有可以创建基础设施的主体凭证。本示例将创建 AWS Identity and Access Management（IAM）角色、一个 S3 桶、ECR 仓库，以及一个 AgentCore 容量提供者和运行时。
• AWS Command Line Interface（AWS CLI）已安装并配置完成。
• 一个虚拟私有云（VPC），至少有一个子网和安全组。
• 在 Amazon Bedrock 控制台中为 Anthropic Claude Sonnet 4.6 启用了模型访问。
• 在本地安装了 Finch 或其他 OCI 兼容的容器工具。
• 已安装 Python 3.10+。
• boto3 ≥ 1.36.0 或 botocore ≥ 1.43.72。旧版本缺少 create_capacity_provider，deploy.py 会失败。

完整可用的代码请参阅 GitHub 上的 AgentCore samples 仓库。

步骤 1：定义你的智能体

AgentCore Runtime Instances 支持任何智能体框架。在本示例中，每个智能体都是使用 Strands Agents 构建的 Python 应用。

有两个细节经常配置错误，而且两者都会产生令人困惑的失败：

@app.entrypoint
def invoke(payload, context):  # 参数必须命名为 "context"
    session_id = getattr(context, "session_id", None) or "local-session"

SDK 通过参数名分发（它检查 params[1] == "context"）。这是读取会话 ID 的唯一方式，智能体需要会话 ID 才能找到彼此的文件并相互调用。

其次，在处理器内部构建 Agent，而不是在模块作用域：

def build_agent(session_id: str, track_id: str) -> Agent:
    return Agent(
        name=AGENT_NAME,
        model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
        system_prompt=SYSTEM_PROMPT,
        session_manager=FileSessionManager(
            session_id=f"{session_id}-{AGENT_NAME}",
            storage_dir=str(track_path(track_id) / f".sessions-{AGENT_NAME}"),
        ),
    )

模块级别的 Agent 在并发请求之间共享，而 Strands 拒绝使用 Agent is already processing a request 进行可重入调用。历史记录通过 FileSessionManager 保存在卷上，这就是隔天后恢复的会话如何记住早期决策的方式。

Composition 智能体将制作人请求（prompt）使用 Claude Sonnet 4.6 转化为音乐简报，然后使用 ACE-Step 基础模型渲染实际音频：

@app.entrypoint
def invoke(payload, context):
    session_id = getattr(context, "session_id", None) or "local-session"
    track_id = payload.get("track_id", "demo-track")
    if payload.get("mode") == "prepare":
        return {"status": "ok", "model_stack": prepare_model_stack()}
    ensure_track_dir(track_id)
    brief = build_agent(session_id, track_id)(
        payload["prompt"], structured_output_model=CompositionBrief
    ).structured_output
    wav = track_path(track_id) / "composition.wav"
    render = render_audio(wav, brief, duration_s=payload.get("duration_s", 30.0))
    write_text(track_id, "composition.md", brief.to_markdown(render))
    return {"status": "ok", "render": render, "host": host_info(),
            "artifacts": [publish(track_id, wav)]}

mode=prepare 调用将栈构建到卷上，这是一个带有 CUDA PyTorch 和 ACE-Step 的 virtualenv。渲染作为子进程在该卷的解释器下运行：

proc = subprocess.run(
    [status["python"], status["runner"], "--out", str(out_path),
     "--prompt", brief.style_tags, "--duration", str(duration_s)],
    capture_output=True, text=True, check=False,
    timeout=RENDER_TIMEOUT_S, env=gpu_env(),
)

交付智能体从共享文件系统读取渲染后的音轨并进行测量。它使用 Claude Sonnet 4.6 来选择 EQ 频段、压缩器设置，以及哪些部分保持不变。数字信号处理（DSP）随后被应用。然后再次对输出进行测量，因此计划是被验证的，而不是被信任的。

before = audio.measure(str(source))  # ITU-R BS.1770-4
plan = build_agent(session_id, track_id)(
    f"Prepare this for {platform} delivery.\n{json.dumps(before.to_dict())}",
    structured_output_model=DeliveryPlan,
).structured_output
data, rate = audio.read_audio(str(source))
bands = [b.model_dump() for b in plan.eq_bands]
if bands:
    data = audio.apply_filters(data, rate, bands)
if plan.compressor.enabled:
    data, _ = audio.compress(
        data, rate,
        threshold_db=plan.compressor.threshold_db,
        ratio=plan.compressor.ratio,
        attack_ms=plan.compressor.attack_ms,
        release_ms=plan.compressor.release_ms,
        knee_db=plan.compressor.knee_db,
    )
data, _ = audio.normalise_loudness(data, rate, plan.target_lufs)
data, _ = audio.limit(data, rate, ceiling_dbtp=plan.target_true_peak_dbtp)
audio.write_audio(str(delivery), data, rate, subtype="PCM_24")
after = audio.measure(str(delivery))  # verify, don't trust

合规智能体独立地重新测量完成的交付物，根据交付智能体声称的交付目标进行检查，并针对工作室自身目录进行谐波相似性筛查。当它发现相似性时，会提出标记并回调到作曲智能体进行修复。

client = boto3.client(
    "bedrock-agentcore", region_name=REGION,
    config=Config(read_timeout=600, retries={"max_attempts": 3, "mode": "standard"}),
)
response = client.invoke_agent_runtime(
    agentRuntimeArn=COMPOSITION_RUNTIME_ARN,  # injected at deploy time, not hard-coded
    qualifier=COMPOSITION_QUALIFIER,
    runtimeSessionId=session_id,  # the caller's own --- routes to this instance
    payload=json.dumps({
        "mode": "remediate", "track_id": track_id,
        "issue": issue,
        "avoid": avoid or {},
    }).encode(),
)
body = json.loads(response["response"].read())  # "response", not "body"

步骤 2：创建容量提供者

容量提供者告诉 AgentCore 为你的智能体配置什么计算基础设施。你指定实例类型和 VPC 放置位置。AgentCore 处理配置和生命周期管理。

你需要两个 IAM 角色，区别很重要：

一个操作员角色，AgentCore 假定它代表你配置 EC2：启动、标记和终止实例及其网络接口。附加托管策略 BedrockAgentCoreRuntimeInstancesOperatorRolePolicy。

一个执行角色，你的智能体进程在运行时假定它来调用 Bedrock 和 S3。CreateAgentRuntime 需要它，没有它会失败。

两者都信任 bedrock-agentcore.amazonaws.com。

现在创建容量提供者。注意名称必须使用下划线（不允许使用连字符）：

control = boto3.client("bedrock-agentcore-control", region_name=REGION)
resp = control.create_capacity_provider(
    name="music_production_capacity",
    permissionsConfiguration={"capacityProviderOperatorRoleArn": operator_arn},
    computeConfiguration={
        "ec2Configuration": {
            "launchTemplateSource": {
                "launchParameters": {
                    "operatingSystem": "LINUX_X86_64", "instanceRequirements": {"allowedInstanceTypes": ["g6.xlarge"]},
                }
            },
            "vpcConfiguration": {"subnets": subnets, "securityGroups": groups},
            "volumes": [
                {"ebsConfiguration": {"name": "tracks", "sizeGiB": 20,
                    "volumeType": "gp3", "encrypted": True}},
                {"ebsConfiguration": {"name": "models", "sizeGiB": 60,
                    "volumeType": "gp3", "encrypted": True,
                    "throughput": 500}},
            ],
            "rootVolume": {"freeSpaceGiB": 30, "volumeType": "gp3"},
            "lifecycleConfiguration": {"idleInstanceTimeout": 600,
                "maxLifetime": 86400},
        }
    },
)

⚠️ GPU 容量。如果在尝试分配 GPU 实例时跨多个可用区（AZ）遇到 InsufficientInstanceCapacity，可以切换到其他 GPU 实例类型，如 g5.xlarge。相应地更新 allowedInstanceTypes。

步骤 3：部署智能体运行时

这里的每个智能体都有自己的运行时。运行时在调用时汇聚在一起。当两个运行时共享一个容量提供者且你使用相同的 runtimeSessionId 调用它们时，AgentCore 会将两个智能体放在同一个 EC2 实例上，在那里它们共享文件系统并可以在同一任务上协作。这就是交付智能体如何读取作曲智能体写入的 .wav 文件的方式。

接下来，将每个运行时指向容量提供者并声明它挂载哪些卷：

composition = control.create_agent_runtime(
    agentRuntimeName="music_production_composition",
    roleArn=execution_arn,  # required
    agentRuntimeArtifact={"containerConfiguration": {"containerUri": image_uri}},
    protocolConfiguration={"serverProtocol": "HTTP"},
    capacityProviderConfiguration={"capacityProviderArn": cp_arn},
    filesystemConfigurations=[
        {"capacityProviderVolume": {"volumeName": "tracks", "mountPath": "/mnt/tracks"}},
        {"capacityProviderVolume": {"volumeName": "models", "mountPath": "/mnt/models"}},
    ],
    lifecycleConfiguration={"idleRuntimeSessionTimeout": 600, "maxLifetime": 86400},
    environmentVariables={"AWS_REGION": REGION, "WORKSPACE_DIR": "/mnt/tracks",
        "MODELS_DIR": "/mnt/models", "MODEL_ID": MODEL_ID},
)

合规智能体的调用相同，只是制品不同，是 zip 而不是镜像，且只需要工作空间卷：

agentRuntimeArtifact={"codeConfiguration": {
    "code": {"s3": {"bucket": bucket, "prefix": key}},
    "runtime": "PYTHON_3_12",
    "entryPoint": ["compliance_agent.py"],
}},

步骤 4：使用共享会话编排多智能体工作流

现在调用这些智能体。这正是三个智能体成为管道的地方：你将相同的 runtimeSessionId 传递给每次调用。第一次调用是慢的，因为它需要配置实例。之后每次调用都会路由到已经运行的实例。

client = boto3.client(
    "bedrock-agentcore", region_name=REGION,
    config=Config(read_timeout=900, retries={"max_attempts": 2, "mode": "standard"}),
)
session_id = f"music-production-{uuid.uuid4()}"  # 33-100 characters

def invoke_agent(runtime_arn, payload):
    response = client.invoke_agent_runtime(
        agentRuntimeArn=runtime_arn, qualifier="DEFAULT",
        runtimeSessionId=session_id, payload=json.dumps(payload).encode(),
    )
    return json.loads(response["response"].read())

invoke_agent(COMPOSITION_ARN, {"mode": "prepare"})
invoke_agent(COMPOSITION_ARN, {"mode": "catalogue", "track_id": track_id, "duration_s": 20})
invoke_agent(COMPOSITION_ARN, {"mode": "compose", "track_id": track_id, "duration_s": 30,
    "prompt": "Create an upbeat electronic track with heavy bass and synth melodies."})
invoke_agent(DELIVERY_ARN, {"track_id": track_id, "platform": "spotify",
    "prompt": "Prepare this for streaming delivery."})
result = invoke_agent(COMPLIANCE_ARN, {"track_id": track_id,
    "prompt": "Screen this delivery for release."})
print(result["outcome"])  # cleared | review_required | not_cleared

以下是它在 us-east-2 中实况 g6.xlarge 上产生的结果：

1. prepare model stack (GPU instance + torch + weights)  239s
   host : ip-172-31-11-83.us-east-2.compute.internal
   stack : ready=True in 125s
2. render back-catalogue  66s
3. compose (renders audio on the GPU)  25s
   rendered : NVIDIA L4 in 8.98s (peak VRAM 7.63 GiB)
   audio 20.062s 48000Hz 2ch -7.5 LUFS peak 0.42 dBTP
4. delivery (real DSP, verified by measurement)  41s
   read : composition.wav <- written by another agent
   before -7.5 LUFS peak 0.42 dBTP
   after -14.0 LUFS peak -3.2 dBTP
   targets : loudness met, true peak held
5. compliance screen  28s
   verdict : REVIEW REQUIRED
   screen : 2 reference(s), closest catalogue_00.wav distance 0.0665 (review)

-- collocation --  All 5 steps were served by one instance, as intended.

调用 StopRuntimeSession 时，实例会自动进入空闲状态。空闲期间不收取计算费用。再次调用该会话时，AgentCore 会恢复实例运行，前提是实例落在同一个可用区。Amazon Elastic Block Store（Amazon EBS）卷是按可用区锁定的。如果原始可用区容量不足，卷无法重新挂载，持久化存储将丢失。使用按可用区固定的 ODCR 或 MODELS_SNAPSHOT_ID，可以确保重新放置后能够恢复。会话最长可保留 14 天。

第 5 步：独立更新 Agent

该架构的优势之一是独立部署。当 Audio AI 团队发布了 composition agent 的新版本时，他们只需更新自己的运行时：

# Audio AI team pushes new container version
finch build --platform linux/amd64 \
  -f Dockerfile.composition \
  -t "$ECR/music-production/composition-agent:v2" .
finch push "$ECR/music-production/composition-agent:v2"

control.update_agent_runtime(
    agentRuntimeId=composition_id,  # all three are required
    roleArn=execution_arn,
    agentRuntimeArtifact={"containerConfiguration": {"containerUri": new_uri}},
    capacityProviderConfiguration={"capacityProviderArn": cp_arn},
)

delivery agent 和 compliance agent 继续运行，不受影响。无需协调，无需共享部署流水线，也不会因为自己的更新而破坏其他团队的 agent。

首先删除会话。这是停止 EC2 和 Amazon EBS 费用最直接的方式。删除会话会取消配置 EC2 资源：实例、网络接口和 Amazon EBS 卷。删除容量提供商也会停止并删除其关联的会话及其持久化存储。但是，必须先将该容量提供商上的每个运行时和运行时版本取消关联，且该分离操作是异步的。会话删除是快速路径，如果想避免持续产生费用，应使用该方式。

data = boto3.client("bedrock-agentcore", region_name=REGION)
data.delete_capacity_provider_session(capacityProviderId=cp_id, sessionId=session_id)
for rid in runtime_ids:
    control.delete_agent_runtime(agentRuntimeId=rid)
# Versions detach from the capacity provider asynchronously, so poll here.
control.delete_capacity_provider(capacityProviderId=cp_id)

然后删除 ECR 仓库、S3 制品、IAM 角色和 Amazon CloudWatch 日志组。

在本文中，我们在 Amazon Bedrock AgentCore Runtime Instances 上部署了一个多 Agent 音乐制作系统。三个由不同团队使用不同打包格式构建的 Agent，在单个 GPU 实例上的同一个会话中协作，生产出了一首可以播放的曲目。

该架构展示了几种可以超越音乐制作场景的模式：

通过共享会话实现共置。同一容量提供商上的多个 Agent 运行时，在使用相同会话 ID 调用时会共享一个实例，这意味着它们共享同一个主机：相同的 GPU、相同的本地卷、相同的进程空间。

独立团队部署。每个 Agent 有自己的运行时生命周期。团队按自己的节奏发布版本，无需协调。

混合制品类型。容器和代码包可以在同一基础设施上共存。使用适合团队工作流程的打包方式即可。

多日持久化。工作暂停时停止会话，再次调用时恢复。会话最长可保留 14 天（恢复取决于是否落在同一可用区）。

音乐只是一个便捷的载体，但该架构本身与音乐无关。换掉渲染步骤，同样的三 Agent 形态也适用于 AWS 为 GPU Instances 标注的工作负载：3D 渲染、仿真、模型推理、媒体处理。或者一个长时运行的作业——管道产生一个大制品，交给第二个 Agent 转换，再由第三个 Agent 在发送前检查结果。

如需了解更多，请参阅 Amazon Bedrock AgentCore 文档。获取本文的完整可运行代码，请探索 AgentCore samples 仓库。对于 Graph、Swarm 或 Workflow 等高级编排模式，请参阅 Strands Agents 文档。