机器学习推理往往需要处理敏感数据——病历、商业机密、私密通信记录，这些信息一旦泄露后果严重。那么问题来了：能不能把推理任务放到云上执行，但同时让云服务本身完全无法接触到你的数据？更具体地说，能否实现整个推理过程中数据始终保持加密状态？这篇文章会给出一个可行的答案：利用Amazon SageMaker AI结合全同态加密（FHE）来完成机器学习推理。基于FHE的方案，查询、响应乃至中间计算结果全程加密，即便是SageMaker AI自身也无法窥探。

简单解释一下FHE：这是一种特殊的加密方式，允许在不解密的情况下直接对加密数据进行计算。放在ML推理的场景里，就是可以对加密的查询直接应用模型，生成加密的预测结果，整个过程不涉及任何解密步骤。来想几个实际场景，你就知道这种能力意味着什么了：

• 医疗健康：保险公司想为医生提供一个ML模型，根据诊断数据预测医疗手术结果。把模型部署到云上能省去很多运维麻烦，但医生绝不能把患者医疗信息暴露给第三方——隐私法规不是闹着玩的。
• 能源行业：某石油天然气公司用ML分析卫星照片，筛选潜在的钻探地点。把模型放在云端能大幅降低成本，但那些涉及敏感地理位置的卫星照片，绝对不能落到第三方手里。
• 电信领域：运营商想对客户邮件进行垃圾和钓鱼检测。用云上的ML可以轻松扩展规模，但数据保护法规要求客户消息在第三方处理时必须保持加密状态。

我们之前发过一篇博客，讲的是如何在SageMaker端点上实现FHE推理，当时是从头手搓一个线性回归算法，用的是底层库SEAL。这次会更进一步——用concrete-ml这个高层库，它专为FHE推理设计，开箱即用支持好几种常见模型类型，甚至API跟scikit-learn兼容，灵活性和可操作性都强了不少。

这篇文章会讲到：

• 用自定义容器在SageMaker AI中训练一个concrete-ml模型
• 把训练好的模型部署到SageMaker AI推理端点
• 创建一个专用于concrete-ml推理的自定义客户端
• 用这个客户端向推理端点发起查询

走完这些步骤，你就会拥有一套完整的端到端加密ML推理系统，核心正是concrete-ml在SageMaker AI中的落地实现。

解决方案概览

concrete-ml在SageMaker AI中的工作流程其实很清晰：

1. 模型所有者先准备好训练数据。concrete-ml要求所有特征归一化到同一量纲，比如[-1, 1]区间。
2. 用这些数据训练一个支持FHE的模型版本——这个模型专门设计来处理加密数据，而非明文。
3. 把训练好的模型部署到SageMaker AI。
4. 客户端利用模型支持的FHE方案加密查询。
5. 加密查询被发送到云端运行FHE模型的端点。
6. 模型在不解密的情况下，将加密查询转换为加密预测。
7. 加密响应返回客户端，由客户端解密并获取最终预测。

这套机制与AWS Nitro System提供的机密计算环境不太一样，但可以形成互补。AWS Nitro Encla ves的做法是：在硬化的隔离环境内解密并处理查询，通过CPU和内存隔离保障安全。而FHE的做法是：查询全程保持加密状态，安全性的根基在于数学而不是硬件或软件。

前提条件

要动手实现这套方案，你需要准备：

• 本机开发环境：Python 3.12、有pip安装包的权限、安装好Docker或其他容器构建工具。建议在虚拟环境中操作，但不是强制要求。
• 一个AWS账户，并准备好以下资源：
  • Amazon Elastic Container Registry（Amazon ECR）中的仓库，用于存放训练和推理容器镜像。
  • Amazon Simple Storage Service（Amazon S3）中的存储位置，用于存放：
    • 训练好的模型文件
    • 训练代码（如果希望与模型文件分开存储）
    • 密钥和加密数据

Amazon S3的安全最佳实践建议务必遵守。

• AWS Identity and Access Management（IAM）中的角色，分别对应：
  • 模型创建者
  • 推理端点创建者
  • 推理端点本身
  • 客户端用户

这些角色的IAM策略，以及一个基于MNIST手写数字集的完整示例，都可以在示例代码仓库中找到。

提醒一点：截至本文写作时，concrete-ml由Zama提供，用于原型开发或非商业用途时不需要购买许可证。但商业用途可能需要获取商用许可证。

训练

构建并部署训练容器

构建训练容器的步骤如下：

1. 切换身份，承担模型训练者角色。
2. 在本机创建 Dockerfile.training 文件。
3. 在 Dockerfile.training 中添加以下内容：

   FROM python:3.12RUN apt-get update && apt-get upgrade -y && apt-get cleanRUN apt-get -y install --no-install-recommends cmakeRUN pip install sagemaker_training==5.1.1 concrete-ml==1.9.0 concrete-python==2.10.0 torch==2.3.1

   务必核对版本号的一致性。concrete-ml库要求整个系统中Python版本、concrete-ml包版本和concrete-python包版本完全对应。

4. 构建容器镜像：

   docker build -f ./Dockerfile.training

5. 将镜像推送至Amazon ECR：
   1. 执行认证命令，让Docker登录到你的Amazon ECR仓库：

   aws ecr get-login-password --region | docker login --username AWS --password-stdin .dkr.ecr..amazonaws.com

   2. 给镜像打上仓库名标签：

   docker tag .dkr.ecr..amazonaws.com/:latest

   3. 推送标签后的镜像：

   docker push .dkr.ecr..amazonaws.com/:latest

验证容器可用

aws ecr describe-images --repository-name

正常的话应该会看到JSON格式的输出，其中包含你的镜像，imageDigest字段不为空，并且有latest标签。

训练模型

接下来是训练步骤。实际上，concrete-ml在这一步跟其他ML框架没什么两样，训练容器也跟自定义训练容器没什么区别。需要注意的是，训练过程基于明文数据进行——也就是说，concrete-ml除了归一化外，不需要对数据做额外预处理。但如果常规训练需要额外预处理，那在这里也同样需要（要么在训练作业开始前做，要么作为训练作业的一部分）。

编写训练脚本

1. 创建 training_script.py 文件。
2. 在文件中加入如下模板代码：

   import argparseimport osimport numpyfrom concrete.ml.sklearn import from concrete.ml.deployment import FHEModelDevdef do_training(model_dir, train):# 从train目录加载数据# 训练模型实例，然后用下面这行保存FHEModelDev(model_dir, model).sa ve()def model_fn(model_dir):# SageMaker AI需要这个函数存在，但不会实际调用raise NotImplementedErrorif __name__ == '__main__':parser = argparse.ArgumentParser()parser.add_argument('--model-dir', type=str, default=os.environ['SM_MODEL_DIR'])parser.add_argument('--train', type=str, default=os.environ['SM_CHANNEL_TRAINING'])args = parser.parse_args()do_training(args.model_dir, args.train)

3. 在 do_training 函数中实现数据加载逻辑。
4. 在 do_training 函数中实现模型训练逻辑。

创建自定义框架

为了方便将训练容器集成到SageMaker AI中，建议创建一个自定义框架：

1. 创建 framework.py 文件。
2. 在文件中加入以下内容：

   from sagemaker.estimator import Frameworkclass Concrete(Framework):def __init__(self,entry_point,source_dir=None,hyperparameters=None,py_version="py312",framework_version="1.9.0",distributions=None,**kwargs,):self.image_uri = super(Concrete, self).__init__(entry_point, source_dir, hyperparameters,image_uri=self.image_uri,**kwargs)self.framework_version = framework_versionself.py_version = py_versiondef training_image_uri(self, region=None):return self.image_uridef create_model(self,model_server_workers=None,role=None,vpc_config_override=None,entry_point=None,source_dir=None,dependencies=None,image_name=None,**kwargs,):return None

3. 将 image_uri 的值替换为你的Amazon ECR训练容器位置。

启动训练作业

这一节会演示如何用Python脚本启动训练作业。当然，通过控制台或AWS CLI也能完成。（注意：训练作业会根据实例类型和运行时长产生费用。）

1. 为Python 3.12创建一个虚拟环境。
2. 激活虚拟环境。
3. 用pip安装以下包：

   boto3==1.37.38sagemaker==2.243.2

4. 创建 start_training.py 文件。
5. 在文件中加入以下内容：

   from sagemaker import sessionfrom framework import Concretesagemaker_session = session.Session()concrete = Concrete(entry_point="training_script.py",instance_count=1,instance_type="ml.m5.xlarge",# 小模型用ml.m5.xlarge，大模型用ml.m5.4xlargerole="arn:aws:iam::123456789012:role/SageMakerModelTrainerRole",# 使用前提条件中的模型训练者角色ARNsagemaker_session=sagemaker_session,hyperparameters={},output_path="s3://my-model-bucket/concrete-ml/models/",# 使用前提条件中的模型桶code_location="s3://my-model-bucket/concrete-ml/scripts/",# S3路径，用于存放训练脚本)concrete.fit(inputs=)

6. 根据你的实际配置，更新 instance_type、role、output_path、code_location 和 inputs 的值。
7. 运行这个文件：

   python start_training.py

8. 验证训练是否成功完成：

   aws sagemaker describe-training-job --training-job-name

   看到 TrainingJobStatus: Completed 就对了。然后确认输出文件是否存在：

   aws s3 ls s3://my-model-bucket/concrete-ml/models/

   确保 server.zip 和 client.zip 两个文件都在。

训练完成后，训练容器会向模型桶写入两个文件：server.zip（供推理端点使用）和 client.zip（客户端用户加密查询用）。

构建并部署推理容器

基于FHE的ML推理比标准ML推理要复杂一些，原因在于几个新的技术约束：

• 客户端需要 client.zip 中的模型专用信息才能生成加密密钥。
• FHE密文的大小可能超过SageMaker AI的查询大小限制，因此客户端和服务端需要通过SageMaker AI API调用之外的方式传输密文。
• FHE求值所需时间可能超过SageMaker AI的超时设置，因此推理会使用SageMaker AI的异步推理机制。
• 端点需要客户端提供求值密钥（一种公钥）才能执行FHE求值。

为了应对这些新需求并优化用户体验，我们将构建一个系统，其中：

• 自定义客户端负责加密查询，并将求值密钥附着在查询中。
• 自定义训练端点能够在需要时获取 client.zip，并用它来对FHE模型进行求值。
• 同一个自定义客户端负责解密来自训练端点的预测结果。
• 客户端和端点通过Amazon S3相互传输密文和密钥。

要部署并使用这套系统，请完成以下步骤。

编写Predictor

创建 predictor.py 文件，内容如下：

from flask import Flaskimport flaskimport loggingimport jsonfrom concrete.ml.deployment import FHEModelServerfrom sagemaker.s3 import S3Uploader, S3Downloader# 加载模型try:model = FHEModelServer("/opt/ml/model/")except Exception:logging.exception("Failed to initialize FHEModelServer")raiseapp = Flask(__name__)@app.route('/ping', methods=['GET'])def ping():return flask.Response(response='n', status=200, mimetype='application/json')@app.route('/invocations', methods=['POST'])def transformation():try:input_json = flask.request.get_json()if not input_json or not isinstance(input_json, dict):return flask.Response(response=json.dumps({"error": "Invalid JSON"}),status=400,mimetype="application/json",)required_keys = ["evaluation_keys_uri","encrypted_query_uri",]for key in required_keys:if key not in input_json:return flask.Response(response=f'Missing required field: {key}',status=400)if (not isinstance(input_json[key], str)or not input_json[key].startswith('s3://')):return flask.Response(response=f'Invalid Amazon S3 URI for {key}', status=400)evaluation_keys_uri = input_json["evaluation_keys_uri"]encrypted_query_uri = input_json["encrypted_query_uri"]downloader = S3Downloader()try:evaluation_keys = downloader.read_bytes(evaluation_keys_uri)encrypted_query = downloader.read_bytes(encrypted_query_uri)except Exception as e:logging.error(f"Failed to download from S3: {e}")return flask.Response(response='Failed to retrieve data from Amazon S3',status=500)prediction = model.run(encrypted_query, evaluation_keys)return flask.Response(response=prediction, status=200, mimetype="application/octet-stream")except KeyError as e:return flask.Response(response=json.dumps({"error": f"Missing key: {str(e)}"}),status=400,mimetype="application/json",)except Exception as e:return flask.Response(response=json.dumps({"error": "Internal server error"}),status=500,mimetype="application/json",)

这个Predictor的逻辑是：接收一个包含三个Amazon S3位置的“查询”——加密查询的位置、对应求值密钥的位置，以及写入预测结果的位置。Predictor从S3下载查询和密钥，执行FHE模型求值，然后将预测结果写回S3。

将Predictor打包成容器

将Predictor打包成容器的步骤如下：

1. 切换身份，承担端点创建者角色。
2. 为容器文件创建一个新目录。
3. 将 predictor.py 复制到新目录中。
4. 获取所需的样板文件（nginx.conf、serve、wsgi.py），可以从示例仓库下载，也可以从SageMaker AI关于自定义推理容器的文档中复制。（需要注意：文档中的nginx.conf超时值需要调大一些，否则FHE求值可能完不成。）
5. 在该目录下创建 Dockerfile.inference 文件。
6. 在 Dockerfile.inference 中添加以下内容：

   FROM python:3.12RUN apt-get -y update && apt-get install -y --no-install-recommends nginx ca-certificates cmake && rm -rf /var/lib/apt/lists/*RUN pip install flask gevent gunicorn sagemaker sagemaker_training==5.1.1 concrete-ml==1.9.0 concrete-python==2.10.0RUN rm -rf /root/.cache# 设置环境变量ENV PYTHONUNBUFFERED=TRUEENV PYTHONDONTWRITEBYTECODE=TRUEENV PATH="/opt/program:${PATH}"COPY / /opt/programRUN chmod +x /opt/program/serveWORKDIR /opt/program

7. 构建容器镜像：

   docker build -f ./Dockerfile.inference

8. 将镜像推送至Amazon ECR。
   1. 执行认证命令，让Docker登录到你的Amazon ECR仓库：

   aws ecr get-login-password --region | docker login --username AWS --password-stdin .dkr.ecr..amazonaws.com

   2. 给镜像打上仓库名标签：

   docker tag .dkr.ecr..amazonaws.com/:latest

   3. 推送标签后的镜像：

   docker push .dkr.ecr..amazonaws.com/:latest

   4. 验证容器是否可用：

   aws ecr describe-images --repository-name

   正常的话会看到JSON输出，包含你的镜像，imageDigest字段不为空，且有latest标签。

部署推理端点

（重要：端点一旦创建就会持续产生费用，直至删除。费用根据实例类型、训练时长和端点运行时间而变化。详细定价信息请查阅Amazon SageMaker AI定价页面。测试结束后务必删除端点以避免不必要的开销。）继续使用端点创建者角色：

1. 创建一个虚拟环境。
2. 激活这个虚拟环境。
3. 用pip安装以下包：

   boto3==1.37.38sagemaker==2.243.2

4. 创建 start_inference_endpoint.py 文件，内容如下：

   from sagemaker.session import Sessionfrom sagemaker.model import Modelfrom sagemaker.predictor import Predictorfrom sagemaker.async_inference.async_inference_config import AsyncInferenceConfigsagemaker_session = Session()model = Model(image_uri="123456789012.dkr.ecr.us-east-1.amazonaws.com/concrete-inference:latest",# 使用上一步构建的ECR URImodel_data="s3://my-model-bucket/concrete-ml/models/model.tar.gz",# 训练作业保存模型的路径role="arn:aws:iam::123456789012:role/SageMakerEndpointRole",# 使用前提条件中的端点角色ARNsagemaker_session=sagemaker_session,predictor_cls=Predictor,)async_config = AsyncInferenceConfig(max_concurrent_invocations_per_instance=1,output_path=,failure_path=,)endpoint = model.deploy(initial_instance_count=1,# 测试时先开1个实例instance_type="ml.m5.xlarge",# 用于FHE的最低推荐配置；追求性能可以用ml.m5.24xlargewait=True,endpoint_logging=True,async_inference_config=async_config,)print(f"Endpoint name: {endpoint.endpoint_name}")

5. 执行脚本：

   python start_inference_endpoint.py

6. 验证端点是否处于服务中：

   aws sagemaker describe-endpoint --endpoint-name

   等 EndpointStatus 显示为 InService 再继续往下走，这可能需要几分钟时间。

脚本会打印出端点的名称，记下来留给客户端用。

创建客户端

用户使用你的系统时，不应该需要了解FHE的任何细节。因此，客户端要把所有FHE相关的复杂性隐藏起来。具体来说，客户端负责：

• 从Amazon S3获取 client.zip。
• 用 client.zip 生成密钥。
• 用这些密钥加密查询。
• 将加密后的查询和对应的求值密钥写入Amazon S3。
• 把这些位置信息发送给推理端点，然后接收加密预测结果的S3位置。
• 从S3获取加密预测结果并进行解密。

要创建这个客户端：

1. 创建 client.py 文件。
2. 在文件中添加以下模板代码：

   import tempfileimport tarfileimport osimport jsonimport sagemakerfrom sagemaker.s3 import S3Uploader, S3Downloaderfrom sagemaker.base_deserializers import BytesDeserializerfrom sagemaker.base_serializers import JSONSerializerfrom sagemaker.predictor import Predictorfrom sagemaker.predictor_async import AsyncPredictorfrom sagemaker.async_inference.waiter_config import WaiterConfigfrom concrete.ml.deployment import FHEModelClientsagemaker_session = sagemaker.Session()predictor = AsyncPredictor(Predictor(,serializer=JSONSerializer(),deserializer=BytesDeserializer(),sagemaker_session=sagemaker_session,))model_location = def get_query():# 返回需要加密的查询的代码...# 下载并解压客户端配置with tempfile.TemporaryDirectory() as config_dir_name:try:S3Downloader().download(model_location,local_path=config_dir_name,sagemaker_session=sagemaker_session,)tf = tarfile.open(os.path.join(config_dir_name, "model.tar.gz"),mode="r:gz")tf.extract("client.zip", config_dir_name)except FileNotFoundError as e:except tarfile.TarError as e:except Exception as e:with tempfile.TemporaryDirectory() as key_dir_name:concrete_client = FHEModelClient(config_dir_name,key_dir=key_dir_name)# 生成并上传求值密钥eval_keys_location = concrete_client.generate_private_and_evaluation_keys()eval_keys = concrete_client.get_serialized_evaluation_keys()uploader = S3Uploader()uploader.upload_bytes(eval_keys,eval_keys_location,sagemaker_session=sagemaker_session)# 加密并上传查询encrypted_query_location = plaintext_query = get_query()encrypted_query = concrete_client.quantize_encrypt_serialize(plaintext_query)uploader.upload_bytes(encrypted_query,encrypted_query_location,sagemaker_session=sagemaker_session)# 向端点发送请求query = {'evaluation_keys_uri': eval_keys_location,'encrypted_query_uri': encrypted_query_location,}query_json = json.dumps(query)try:async_response = predictor.predict_async(data=query_json,input_path="",initial_args={"ContentType": "application/json"},)# 等待端点返回结果encrypted_result = async_response.get_result(waiter_config=WaiterConfig(""))prediction = concrete_client.deserialize_decrypt(encrypted_result)except TimeoutError as e:except Exception as e:

3. 实现 get_query() 函数，用于获取明文查询。
4. 更新Amazon S3位置、端点名称和模型位置的占位值。
5. 根据应用程序需求，为 占位块编写异常处理代码，覆盖TimeoutError、FileNotFoundError和TarError等场景。

（有一点你可能注意到了：客户端和端点对加密查询和响应的处理方式不一样。客户端通过手动将加密查询写入Amazon S3，然后把S3位置作为实际查询提交给端点。而端点直接返回加密结果，让SageMaker AI负责读写S3。为什么会有这种差异？因为加密响应是一个字节串，SageMaker AI可以自然处理。但客户端的查询本身是一个JSON结构，必须包含求值密钥的位置。加密查询要想嵌入同一个JSON，就得做Base64之类的编码，这会增加不必要的处理和网络开销。所以示例代码选择直接处理加密查询，绕过了这步编码。）

接下来：

1. 创建一个虚拟环境。
2. 激活虚拟环境。
3. 安装所需包：

   boto3==1.37.38sagemaker==2.243.2concrete-ml==1.9.0concrete-python==2.10.0

最后：

1. 切换身份，承担客户端角色。
2. 执行脚本：python client.py
3. 将预测输出与预期结果进行比较，验证FHE加密是否正常工作。

清理资源

为避免产生后续费用，请删除你创建的资源：

1. 通过SageMaker AI控制台或SDK删除推理端点。
2. 确认端点已被删除：

   aws sagemaker describe-endpoint --endpoint-name

   返回错误，提示端点不存在，就对了。

3. 通过SageMaker AI控制台或SDK删除端点配置。
4. 确认端点配置已被删除：

   aws sagemaker list-endpoint-configs

   应该看不到有匹配的端点配置。

5. 通过SageMaker AI控制台或SDK删除SageMaker AI模型。
6. 确认模型已被删除：

   aws sagemaker list-models

   应该看不到有匹配的模型。

7. 通过Amazon S3控制台或AWS CLI删除模型工件、加密查询、加密响应和求值密钥。
8. 确认Amazon S3对象已被删除：

   aws s3 ls s3:///

   应该为空或没有匹配的对象。

9. 通过Amazon ECR控制台或AWS CLI删除容器镜像。
10. 确认容器镜像已被删除：

    aws ecr describe-images --repository-name

    应该看不到有匹配的镜像。

常见问题

• 推理时出现TimeoutError：增加WaiterConfig的max_attempts，或者换用更大的实例类型。
• AccessDenied错误：检查IAM角色是否拥有正确的S3和SageMaker AI权限。
• 容器构建失败：确认Docker有足够内存（8GB以上）。
• 推理时出现服务器错误：检查concrete-ml各个包的版本是否一致。

性能与安全考量

FHE提供的是密码学层面的保护，代价是性能上的权衡。开销大小取决于具体模型，但通常可以预期比明文推理慢到100,000倍。有两种方式可以缓解这个问题。一是增加实例的vCPU数量。二是利用经典的ML技术“量化”，降低模型推理时的数值精度。concrete-ml的运行时间会随数值精度上升而增加，因此量化的效果在FHE场景下甚至比常规ML推理还明显——当然，量化可能影响模型精度，但FHE本身的转换不会改变精度。在我们的测试中，模型代码中的量化将开销降到了2800倍（ml.m5.xlarge实例上，从67ms增加到187秒），且未观察到精度损失。如果再增加vCPU数量，在ml.m5.24xlarge实例上还能进一步降到500倍（46秒）。

对某些应用来说，这个延迟仍然不小。由于这一开销，FHE目前还不适合交互式、对延迟敏感的应用。但对于异步或批量处理场景，当隐私需求比延迟更重要时，它就是一把利器。回到文章开头提到的几个场景：

• 为医生提供基于诊断数据预测手术结果的模型。
• 分析卫星照片筛选潜在钻探地点。
• 检测邮件中的垃圾和钓鱼信息。

这些场景都能容忍额外几秒钟的延迟。

需要注意的是，客户端务必确保解密后的查询和预测结果不被泄露——concrete-ml的加密和对应的明文解密结合在一起，可能暴露密钥信息。另外，这套系统并不保护模型本身的机密性。查询和响应在SageMaker AI看来是加密且不可读的，但concrete-ml并没有对模型本身进行加密，模型在SageMaker AI眼里仍然是可见的。那些能够看到明文查询和响应的人，也有可能通过“模型窃取”攻击获得模型信息。最后，concrete-ml不提供电路隐私：密文有可能泄露模型信息。不过，客户仍然可以借助AWS为Amazon S3和SageMaker AI提供的标准安全机制来保护模型和密文。记住：安全是AWS与客户共担的责任。按照最佳实践，客户应当：

• 创建IAM角色时遵循最小权限原则。只为每个角色授予完成其功能所需的最低权限。参考仓库中的示例IAM策略，根据你的实际用例调整资源ARN和操作。
• 对那些不是FHE密文的数据，启用Amazon S3桶加密。这包括为所有存储模型、数据和求值密钥的S3桶启用默认加密，以保护静态数据。
• 将S3桶的权限缩减到系统所需的最小范围。

结论

在SageMaker AI中使用FHE工具，可以对加密数据执行推理，且数据在整个过程中保持不可读状态。这种做法能让你享受SageMaker AI带来的敏捷性、可扩展性和托管基础设施，同时实现从查询到响应的全程密码学保护。

想了解更多关于AWS安全和加密的内容，可以参考以下资源：

• AWS为客户提供了哪些密码学服务？
• AWS密码学工具文档
• AWS的密码学计算
• 全同态加密概览
• 在Amazon SageMaker端点上启用全同态加密，实现安全的实时推理
• Amazon S3安全
• Amazon Elastic Container Service安全
• 配置Amazon SageMaker AI安全
• IAM安全最佳实践