企业官网建设流程全解析

• Mock
• 测试

导读

SageMaker Runtime 是 AWS 推理服务中负责向已部署端点发起调用（同步InvokeEndpoint、异步InvokeEndpointAsync）的运行时组件。在 moto 中，sagemaker-runtime后端为这两类调用提供了可编程的 Mock 能力：默认情况下返回静态数据，开发者可以借助 moto 专属的/moto-api/static/sagemaker/...接口预先配置"预期结果队列"，从而精确控制每次调用的响应内容、失败注入与返回头。读完本文，你将掌握该模块的全部已实现能力、结果队列的入队与出队规则、同步/异步调用在底层源码中的真实行为，以及如何在单元测试中按场景编排这些结果。本文以 docs/docs/services/sagemaker-runtime.rst 为骨架，结合 moto/sagemakerruntime/models.py、moto/sagemakerruntime/responses.py 与 tests/test_sagemakerruntime/test_sagemakerruntime.py 展开源码级讲解。

一、已实现功能总览

根据官方服务文档（见 docs/docs/services/sagemaker-runtime.rst），moto 对sagemaker-runtime的支持情况如下：

也就是说，moto 目前覆盖了推理调用中最常用的同步与异步两条路径；凡是请求了invoke_endpoint_with_response_stream的代码，在 Mock 环境下会得到"未实现"的响应，测试设计时需要避开该操作或自行做条件分支。

二、默认行为：静态数据

2.1 同步调用invoke_endpoint

在不做任何额外配置时，invoke_endpoint会返回一组写死的静态数据。其默认值定义在 moto/sagemakerruntime/models.py 的SageMakerRuntimeBackend.invoke_endpoint方法中：

self.results[endpoint_name][unique_repr] = ( "body", "content_type", "invoked_production_variant", "custom_attributes", )

返回值是一个四元组，四个元素会被 moto/sagemakerruntime/responses.py 的SageMakerRuntimeResponse.invoke_endpoint映射到 Boto3 客户端返回结构的不同字段：

这一点在 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint__default_results中得到验证：即使调用时传入了Accept="sth"、TargetModel="tm"，返回的Body仍是b"body"，CustomAttributes仍是"custom_attributes"。

从源码结构可以推断，"同一次 Mock 会话内、相同请求特征"的调用会命中缓存：invoke_endpoint内部以endpoint_name为键维护一个嵌套字典self.results，请求的唯一标识unique_repr则由请求头中所有以x-amzn-sagemaker开头的字段、Accept头以及请求体Body共同计算并 base64 编码而来（见 moto/sagemakerruntime/responses.py）。因此"相同 EndpointName + 相同请求头 + 相同 Body"的重复调用会复用第一次的结果。

2.2 异步调用invoke_endpoint_async

异步调用的默认行为同样是静态数据，但其返回结构完全不同。真实 AWS 的异步推理会将结果写入用户指定的 S3 位置，moto 模拟了这一流程：默认返回的数据是json.dumps({"default": "response"})，并会在当前账号/分区下自动创建一个名为sagemaker-output-{uuid}的 S3 Bucket，将结果以response.json写入（见 moto/sagemakerruntime/models.py）。

底层实现位于SageMakerRuntimeBackend.invoke_endpoint_async（moto/sagemakerruntime/models.py），它最终返回(output_location, failure_location)两个 S3 路径，其中：

• output_location：s3://sagemaker-output-{uuid}/response.json；
• failure_location：s3://sagemaker-output-{uuid}/failure.json（仅当本次结果被标记为失败时，数据才会写入该文件）。

对应的响应层 moto/sagemakerruntime/responses.py 会把这些位置放入响应头：

• X-Amzn-SageMaker-OutputLocation
• X-Amzn-SageMaker-FailureLocation
• InferenceId：若请求未携带X-Amzn-SageMaker-Inference-Id头，则由random.uuid4()生成

这与真实 AWS 的异步推理语义保持一致：客户端拿到OutputLocation后再到 S3 中读取推理结果。

三、用 moto-api 结果队列定制响应

3.1 队列机制的核心思想

无论是同步还是异步调用，moto 都提供了一套"预期结果队列"机制：通过向 moto 专属的 HTTP 接口 POST 一个 JSON 负载，将一系列结果按顺序压入后端队列；之后每次"新"的推理请求会从队列头部弹出一个结果。该机制的关键规则（官方文档与源码一致）是：

1. 相同请求 → 相同结果：后续使用相同请求特征的调用（同步场景看请求头/Body，异步场景看InputLocation）会返回与第一次完全相同的结果，而不会再次消费队列；
2. 不同请求 → 队列下一个结果：其他使用不同特征的新请求会取走队列中的下一条；
3. 队列为空 → 回退静态数据：队列耗尽后，新请求返回默认静态数据。

从源码看，这一规则在 moto/sagemakerruntime/models.py（同步）与第 107-144 行（异步）中以同样的模式实现：先查缓存字典，命中即返回；未命中再从队列pop(0)，否则生成默认值，最后把结果写回缓存。

3.2 配置同步结果队列（endpoint-results）

向/moto-api/static/sagemaker/endpoint-results发起 POST 即可配置同步调用invoke_endpoint的结果队列。官方文档给出的示例负载如下（原文见 docs/docs/services/sagemaker-runtime.rst）：

expected_results = { "account_id": "123456789012", # 默认账号，可省略 "region": "us-east-1", # 默认区域，可省略 "results": [ { "Body": "first body", "ContentType": "text/xml", "InvokedProductionVariant": "prod", "CustomAttributes": "my_attr", }, # 可按需添加更多结果 ], } requests.post( "http://motoapi.amazonaws.com/moto-api/static/sagemaker/endpoint-results", json=expected_results, ) client = boto3.client("sagemaker-runtime", region_name="us-east-1") details = client.invoke_endpoint(EndpointName="asdf", Body="qwer")

每个结果条目支持的字段与对应关系如下：

account_id与region缺省时分别使用默认账号与us-east-1，这一点在 moto-api 的解析代码 moto/moto_api/_internal/responses.py 中可以看到：account_id = body.get("account_id", DEFAULT_ACCOUNT_ID)、region = body.get("region", "us-east-1")，随后逐条把结果追加到sagemakerruntime_backends[account_id][region].results_queue（见 moto/moto_api/_internal/models.py）。也就是说，队列是按"账号 + 区域"隔离的，多账号/多区域测试时需各自配置。

3.3 配置异步结果队列（async-endpoint-results）

异步调用的结果队列通过/moto-api/static/sagemaker/async-endpoint-results配置，每个条目额外支持is_failure字段来模拟推理失败。官方文档示例：

expected_results = { "account_id": "123456789012", # 默认账号，可省略 "region": "us-east-1", # 默认区域，可省略 "results": [ { "data": json.dumps({"first": "output"}), }, { "is_failure": True, "data": "second inference failed", }, # 可按需添加更多结果 ], } requests.post( "http://motoapi.amazonaws.com/moto-api/static/sagemaker/async-endpoint-results", json=expected_results, ) client = boto3.client("sagemaker-runtime", region_name="us-east-1") details = client.invoke_endpoint_async(EndpointName="asdf", InputLocation="qwer")

字段语义：

队列消费时，is_failure=True的结果会走失败分支，把data原样写入failure.json；成功结果写入response.json。测试 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint_async验证了这条完整链路：配置两个结果 → 第一次调用拿到OutputLocation→ 从 S3 读回内容为{"first": "output"}→ 换一个InputLocation触发第二条结果（is_failure=True）→ 从FailureLocation读回"second failure"，同时InferenceId与请求携带的保持一致。

3.4 向后兼容：异步队列为空时回退同步队列

这是异步调用特有的规则，官方文档明确说明："如果异步队列为空，将使用已配置的同步队列"（for backward compatibility）。底层实现见 moto/sagemakerruntime/models.py：当async_results_queue为空但results_queue非空时，会从同步队列弹出条目，并把四元组包装成 JSON 写入 S3：

elif self.results_queue: # Backward compatibility is_failure = False body, _type, variant, attrs = self.results_queue.pop(0) data = json.dumps( { "Body": body, "ContentType": _type, "InvokedProductionVariant": variant, "CustomAttributes": attrs, } )

对应测试test_invoke_endpoint_async_should_read_sync_queue_if_async_not_configured（tests/test_sagemakerruntime/test_sagemakerruntime.py）验证了该回退路径：只配置同步队列、不配置异步队列时，异步调用依然能读到同步队列里的Body内容。

四、路由与请求协议：URL 是如何被匹配的

理解路由有助于在 Server Mode / Proxy Mode 下调试问题。sagemaker-runtime 的 URL 规则定义在 moto/sagemakerruntime/urls.py：

url_bases = [ r"https?://runtime\.sagemaker\.(.+)\.amazonaws\.com", ] url_paths = { "{0}/endpoints/(?P<name>[^/]+)/async-invocations$": response.dispatch, "{0}/endpoints/(?P<name>[^/]+)/invocations$": response.dispatch, }

即：

• 同步调用InvokeEndpoint→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/invocations；
• 异步调用InvokeEndpointAsync→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/async-invocations。

异步响应层正是从路径中解析端点名的：endpoint_name = self.path.split("/")[2]（见 moto/sagemakerruntime/responses.py），并从请求头读取X-Amzn-SageMaker-InputLocation与X-Amzn-SageMaker-Inference-Id。因此测试中给InputLocation传任意字符串都是允许的——它只作为"请求特征"参与缓存判定，并不校验其是否为真实 S3 路径。

此外，moto-api 的静态配置接口注册在 moto/moto_api/_internal/urls.py：

"{0}/moto-api/static/sagemaker/endpoint-results": response_instance.set_sagemaker_result, "{0}/moto-api/static/sagemaker/async-endpoint-results": response_instance.set_sagemaker_async_result,

这解释了为什么在默认（非 Server Mode）环境下配置队列要访问http://motoapi.amazonaws.com/moto-api/...：moto 会把对motoapi.amazonaws.com域名的请求拦截并交给 moto-api 后端处理。而在 Server Mode（TEST_SERVER_MODE）下，测试代码则使用localhost:5000作为 base URL——这一差异在测试文件中通过settings.TEST_SERVER_MODE做了分支（见 tests/test_sagemakerruntime/test_sagemakerruntime.py）。

五、完整可运行的测试示例

综合以上机制，下面是一段覆盖同步 + 异步 + 失败注入 + S3 落盘校验的完整示例（逻辑参照 tests/test_sagemakerruntime/test_sagemakerruntime.py）：

import json import boto3 import requests from moto import mock_aws, settings @mock_aws def test_sagemaker_runtime_queue(): client = boto3.client("sagemaker-runtime", region_name="us-east-1") base_url = "localhost:5000" if settings.TEST_SERVER_MODE else "motoapi.amazonaws.com" # 1. 配置同步结果队列：两条结果 requests.post( f"http://{base_url}/moto-api/static/sagemaker/endpoint-results", json={ "results": [ {"Body": "first body", "ContentType": "text/xml"}, {"Body": "second body"}, ] }, ) # 相同请求 -> 返回队列第一条（并缓存） r1 = client.invoke_endpoint(EndpointName="asdf", Body="qwer") assert r1["Body"].read() == b"first body" r1_again = client.invoke_endpoint(EndpointName="asdf", Body="qwer") assert r1_again["Body"].read() == b"first body" # 不同请求（换了 Accept/TargetModel）-> 取队列第二条 r2 = client.invoke_endpoint(EndpointName="asdf", Body="qwer", Accept="sth") assert r2["Body"].read() == b"second body" # 2. 配置异步结果队列：一个成功、一个失败 requests.post( f"http://{base_url}/moto-api/static/sagemaker/async-endpoint-results", json={ "results": [ {"data": json.dumps({"first": "output"})}, {"is_failure": True, "data": "second inference failed"}, ] }, ) async_r1 = client.invoke_endpoint_async(EndpointName="asdf", InputLocation="qwer") s3 = boto3.client("s3", "us-east-1") # 从 OutputLocation 读回成功结果 output_bucket, output_key = async_r1["OutputLocation"].replace("s3://", "").split("/", 1) out = s3.get_object(Bucket=output_bucket, Key=output_key)["Body"].read().decode("utf-8") assert json.loads(out) == {"first": "output"} # 不同 InputLocation -> 触发失败结果 async_r2 = client.invoke_endpoint_async( EndpointName="asdf", InputLocation="asf", InferenceId="sth" ) assert async_r2["InferenceId"] == "sth" failure_bucket, failure_key = async_r2["FailureLocation"].replace("s3://", "").split("/", 1) fail = s3.get_object(Bucket=failure_bucket, Key=failure_key)["Body"].read().decode("utf-8") assert fail == "second inference failed"

使用要点小结：

• 使用@mock_aws装饰器即可启用本模块（与 moto 其他服务一致）；
• 结果队列按(account_id, region)隔离，多环境测试需分别配置；
• 队列消费是"惰性"的：只有出现新的请求特征时才会弹出下一条；重复请求永远命中缓存，这在幂等测试中非常有用；
• 异步结果最终落在由 moto 自动创建的 S3 Bucket（sagemaker-output-{uuid}）中，因此可以直接用boto3.client("s3")断言落盘内容，包括失败场景；
• invoke_endpoint_with_response_stream尚未实现，涉及流式推理的用例需绕开。

六、扩展阅读

• 服务实现核心：moto/sagemakerruntime/models.py（SageMakerRuntimeBackend及两条队列的实现）
• 响应层与头映射：moto/sagemakerruntime/responses.py
• 路由规则：moto/sagemakerruntime/urls.py
• moto-api 静态配置接口的路由与解析：moto/moto_api/_internal/urls.py、moto/moto_api/_internal/responses.py、moto/moto_api/_internal/models.py
• 行为验证测试：tests/test_sagemakerruntime/test_sagemakerruntime.py
• 官方服务文档：docs/docs/services/sagemaker-runtime.rst

• Mock
• 测试