音乐生成并不只是调用一次模型：编曲、音色渲染、混音母带往往需要不同的上下文、模型和失败恢复策略。Amazon Bedrock AgentCore Runtime Instances 为这类长流程提供 AWS 托管的 EC2 基础设施、GPU、持久卷和可持续多天的会话，因此可以把多个智能体放在同一台 GPU 实例上，通过共享文件系统逐步交付作品。
为什么把三个智能体放在同一实例
一个可落地的音乐流水线可以划分为三种职责：
- 编曲智能体：生成速度、调式、段落、音符以及乐器配置，输出结构化制作计划。
- 渲染智能体：读取计划，调用 GPU 音频模型生成分轨、MIDI 或音频片段。
- 混音智能体：检查所有分轨，执行增益调整、混合、限制和最终导出。
三者共置在一个 Runtime Instance 上有几个直接好处：
- 大型音频文件不需要在服务之间反复上传和下载。
- GPU 模型可以在同一运行环境中调度，减少跨节点通信。
- 中间分轨写入持久卷后，即使某个阶段重试，也不必从头生成。
- 多日会话适合需要人工试听、修改提示词再继续执行的制作过程。
代价也很明确：单实例会成为共享故障域，而且多个智能体可能争抢 GPU 显存。生产环境中应让一个调度器串行执行重型推理，或者根据模型显存占用设置并发上限，而不是让三个进程无约束地同时加载模型。
把共享目录设计成智能体之间的协议
共享文件系统不应只是一个随意堆放 WAV 文件的目录。更稳妥的做法是让每次制作拥有独立的运行目录，并规定明确的输入输出：
/workspace/runs/song-2025-001/
├── plan.json
├── stems/
│ ├── lead.wav
│ ├── bass.wav
│ └── pulse.wav
├── final/
│ └── finished_track.wav
└── status.json
plan.json 是编曲智能体交给渲染智能体的契约；stems/ 是渲染结果；只有在全部分轨验证成功后，混音智能体才写入 final/。状态文件应记录当前阶段、模型版本、提示词摘要、随机种子和错误信息，以便安全重试和复现结果。
文件写入最好采用“临时文件加原子重命名”的方式。这样，下游智能体不会误读尚未写完的 JSON。对于大型音频文件，还可以同时写入校验和与采样率、声道数等元数据。
一个可直接运行的三智能体骨架
下面的示例只使用 Python 标准库，可以在本地直接运行。它用正弦波代替真正的 GPU 音乐模型，重点演示三个逻辑智能体如何通过共享目录交接产物。部署到 Runtime Instance 时，可以把 render_agent 中的合成代码替换成实际的模型推理调用。
将以下内容保存为 pipeline.py：
import json
import math
import os
import sys
import wave
from array import array
from pathlib import Path
RATE = 44100
WORKSPACE = Path(os.getenv('WORKSPACE', './workspace'))
RUN_ID = os.getenv('RUN_ID', 'demo-track')
RUN_DIR = WORKSPACE / 'runs' / RUN_ID
STEMS_DIR = RUN_DIR / 'stems'
FINAL_DIR = RUN_DIR / 'final'
def write_json_atomic(path, payload):
path.parent.mkdir(parents=True, exist_ok=True)
temp = path.with_suffix(path.suffix + '.tmp')
with temp.open('w', encoding='utf-8') as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2)
os.replace(temp, path)
def write_wav(path, samples):
path.parent.mkdir(parents=True, exist_ok=True)
data = array('h', samples)
if sys.byteorder != 'little':
data.byteswap()
with wave.open(str(path), 'wb') as output:
output.setnchannels(1)
output.setsampwidth(2)
output.setframerate(RATE)
output.writeframes(data.tobytes())
def read_wav(path):
with wave.open(str(path), 'rb') as source:
assert source.getnchannels() == 1
assert source.getsampwidth() == 2
assert source.getframerate() == RATE
data = array('h', source.readframes(source.getnframes()))
if sys.byteorder != 'little':
data.byteswap()
return list(data)
def composer_agent():
plan = {
'tempo_bpm': 100,
'key': 'A minor',
'notes': [
{'frequency': 440.00, 'seconds': 0.4},
{'frequency': 523.25, 'seconds': 0.4},
{'frequency': 659.25, 'seconds': 0.4},
{'frequency': 523.25, 'seconds': 0.4},
{'frequency': 440.00, 'seconds': 0.8},
{'frequency': 329.63, 'seconds': 0.8}
]
}
write_json_atomic(RUN_DIR / 'plan.json', plan)
write_json_atomic(RUN_DIR / 'status.json', {'stage': 'composed'})
def render_agent():
with (RUN_DIR / 'plan.json').open(encoding='utf-8') as handle:
plan = json.load(handle)
roles = {'lead': 1.0, 'bass': 0.5, 'pulse': 2.0}
for role, frequency_ratio in roles.items():
samples = []
for note in plan['notes']:
count = int(note['seconds'] * RATE)
frequency = note['frequency'] * frequency_ratio
for index in range(count):
attack = min(1.0, index / max(1, int(0.02 * RATE)))
release = min(1.0, (count - index) / max(1, int(0.03 * RATE)))
envelope = max(0.0, min(attack, release))
value = math.sin(2 * math.pi * frequency * index / RATE)
samples.append(int(32767 * 0.22 * envelope * value))
write_wav(STEMS_DIR / f'{role}.wav', samples)
write_json_atomic(
RUN_DIR / 'status.json',
{'stage': 'rendered', 'stems': sorted(roles)}
)
def mixer_agent():
paths = sorted(STEMS_DIR.glob('*.wav'))
if not paths:
raise RuntimeError('No stems found')
tracks = [read_wav(path) for path in paths]
frame_count = min(len(track) for track in tracks)
mixed = [sum(track[i] for track in tracks) for i in range(frame_count)]
peak = max(1, max(abs(value) for value in mixed))
master = [int(value * 32767 * 0.9 / peak) for value in mixed]
output = FINAL_DIR / 'finished_track.wav'
write_wav(output, master)
write_json_atomic(
RUN_DIR / 'status.json',
{'stage': 'finished', 'output': str(output)}
)
return output
if __name__ == '__main__':
composer_agent()
render_agent()
result = mixer_agent()
print(f'Created {result}')
运行方式：
export WORKSPACE="$PWD/workspace"
export RUN_ID="song-$(date +%Y%m%d-%H%M%S)"
python pipeline.py
cat "$WORKSPACE/runs/$RUN_ID/status.json"
最终文件位于 workspace/runs/<RUN_ID>/final/finished_track.wav。这个示例把三个智能体放在一个进程中以便演示；实际部署时，可以将它们拆成独立进程或任务，由一个编排器根据 status.json 和文件是否存在来触发下一阶段。
部署时不要忽略这些边界
迁移到 AgentCore Runtime Instances 时，可以把入口程序、模型依赖和系统音频库打包到同一个运行环境，并将持久卷挂载为 /workspace。具体实例创建参数应以当前 AWS 控制台、SDK或基础设施即代码接口为准，不要把本地目录当成持久存储。
上线前建议检查：
- 幂等性：重试渲染阶段时，是否会覆盖已经确认的分轨？
- GPU 调度：是否限制了同时加载的模型数量，并监控显存峰值？
- 恢复能力：实例进程退出后，能否从
status.json指示的阶段继续？ - 产物验证：混音前是否检查采样率、声道数、长度和文件完整性？
- 可观测性：日志是否包含运行 ID、智能体名称、耗时、模型版本和随机种子？
- 安全与版权：用户上传的参考音频是否隔离，模型和素材是否允许商业使用？
- 成本控制：等待人工反馈时，是否仍有必要持续占用 GPU 实例？
这种架构最适合中间文件较大、GPU 推理较重且工作流可能持续数小时甚至数天的任务。若每一步都很短、可以完全无状态化，拆分成独立服务可能更容易扩缩容；如果分轨共享和连续迭代才是主要成本，那么单实例、多智能体、持久工作区会更直接。