无法本地部署Vertex AI自定义预测器问题求助

问题分析 #

你遇到的ImportError: sys.meta_path is None, Python is likely shutting down错误，是因为Python进程即将关闭时，LocalEndpoint的__del__析构函数尝试调用Docker API停止容器，但此时Python的模块导入系统已被销毁，导致无法加载必要模块完成操作。健康检查返回b'{}'说明容器本身已正常启动，错误出现在进程收尾阶段。

解决方案 #

1. 手动调用stop()方法，避免依赖自动析构 #

不要依赖Python垃圾回收自动触发容器停止，主动在代码中调用local_endpoint.stop()，确保操作在Python shutdown前执行：

使用with语句的调整版本：

from google.cloud.aiplatform.prediction import LocalModel
from google.cloud import aiplatform

from model.predictor import MyPredictorBasic

display_name = "test_model"
model_path = "./model/"
project_id = "ai-play-430308"
repository = "test-repo"
image = "test-image"
region = "europe-west2"
output_image_uri = f"{region}-docker.pkg.dev/{project_id}/{repository}/{image}"
requirements_path = model_path + "requirements.txt"

aiplatform.init(project=project_id, location=region)

local_model = LocalModel.build_cpr_model(
    src_dir=model_path,
    output_image_uri=output_image_uri,
    predictor=MyPredictorBasic,
    requirements_path=requirements_path,
)

spec = local_model.get_serving_container_spec()
print(spec)

with local_model.deploy_to_local_endpoint() as local_endpoint:
    health_check_response = local_endpoint.run_health_check()
    print("health_check_response", health_check_response.content)
    
    # 手动调用stop，提前完成容器清理
    local_endpoint.stop()

非with语句的调整版本：

local_endpoint = local_model.deploy_to_local_endpoint()
try:
    health_check_response = local_endpoint.run_health_check()
    print("health_check_response", health_check_response.content)
    # 在这里可以添加预测调用逻辑
    # predictions = local_endpoint.predict(instances=[[1], [2]])
finally:
    # 确保无论是否出错，都手动停止容器
    local_endpoint.stop()

2. 完善requirements.txt内容 #

空的requirements.txt可能导致镜像构建时缺少必要依赖，建议添加基础依赖：

google-cloud-aiplatform>=1.40.0

3. 验证Docker环境 #

• 确认Docker daemon正在运行：执行docker ps查看是否能正常返回容器列表
• 确保当前用户拥有Docker操作权限（无需sudo），可通过groups命令检查是否在docker组中

额外测试：添加预测调用验证功能 #

在健康检查通过后，可添加预测调用验证自定义预测器是否正常工作：

with local_model.deploy_to_local_endpoint() as local_endpoint:
    health_check_response = local_endpoint.run_health_check()
    print("health_check_response", health_check_response.content)
    
    # 测试预测功能
    predictions = local_endpoint.predict(instances=[{}, {}])
    print("predictions", predictions)
    
    local_endpoint.stop()

内容的提问来源于stack exchange，提问作者Andy T