在AI基础设施和模型部署的实践中,我们经常使用Python框架(如FastAPI或Flask)来构建API服务。当模型推理结果返回或日志数据需要通过HTTP响应传递时,数据必须被序列化为JSON格式。
然而,Python标准库的json模块对某些特定类型是无能为力的,尤其是那些在数据科学和模型推理中极为常见的类型,例如 NumPy数组 (np.ndarray)、NumPy整数/浮点数 (np.int64)、以及 datetime对象。
本文将深入探讨如何高效且鲁棒地解决这一核心的序列化问题,提供一个适用于生产环境的通用JSON编码器。
1. 为什么会发生序列化错误?
标准JSON(JavaScript Object Notation)只支持少数基本类型:字符串、数字、布尔值、数组(列表)和对象(字典)。当Python尝试将一个不兼容的类型(如NumPy数组)转换为JSON时,就会抛出如下错误:
TypeError: Object of type ndarray is not JSON serializable
让我们通过一个简单的例子来重现这个错误,模拟模型返回了NumPy格式的置信度得分和记录时间戳:
import json
import numpy as np
from datetime import datetime
# 模拟模型输出
model_result = {
"scores": np.array([0.95, 0.05]),
"prediction_id": np.int64(42),
"request_time": datetime.now()
}
try:
# 尝试直接序列化
json.dumps(model_result)
except TypeError as e:
print(f"发生错误: {e}")
# 发生错误: Object of type ndarray is not JSON serializable
2. 解决方案:使用自定义JSON编码器
解决这个问题的核心方法是为json.dumps函数提供一个自定义的默认处理函数(default参数),或者定义一个继承自json.JSONEncoder的类。
2.1 基础实现:通用编码函数
在模型部署场景中,我们需要关注三类最常见的非序列化对象:
1. np.ndarray 或 np.generic (NumPy标量):
2. datetime.datetime 或 datetime.date:
3. set (集合类型):
我们可以定义一个通用的函数来处理这些类型:
import json
import numpy as np
from datetime import datetime, date
from uuid import UUID
def ai_infra_json_encoder(obj):
"""处理常见的AI/ML部署中非JSON兼容类型的编码器。"""
# 1. 处理 NumPy 数组和标量
if isinstance(obj, (np.ndarray, np.generic)):
# 将NumPy数组转换为标准Python列表
# 将NumPy标量转换为标准Python int/float
return obj.tolist()
# 2. 处理 datetime/date 对象
elif isinstance(obj, (datetime, date)):
return obj.isoformat() # 转换为 ISO 8601 字符串
# 3. 处理 set 对象
elif isinstance(obj, set):
return list(obj) # 转换为列表
# 4. 处理 UUID 对象 (常见于请求ID)
elif isinstance(obj, UUID):
return str(obj)
# 如果是其他未知类型,让标准序列化器抛出错误
raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
# 示例数据 (包含多种非兼容类型)
example_data = {
"inference_vector": np.array([0.1, 0.5, 0.4]),
"execution_time_ms": np.float64(12.34),
"feature_ids": {101, 202},
"log_time": datetime.now()
}
# 使用自定义编码器进行序列化
json_output = json.dumps(
example_data,
default=ai_infra_json_encoder,
indent=4
)
print(json_output)
2.2 结果输出 (可运行)
运行上述代码,我们将得到一个完美序列化的JSON字符串:
{
"inference_vector": [
0.1,
0.5,
0.4
],
"execution_time_ms": 12.34,
"feature_ids": [
202,
101
],
"log_time": "2023-11-20T10:30:00.123456"
// (时间戳取决于运行时间)
}
3. 集成到Web框架中的最佳实践
在现代AI部署框架中,如使用FastAPI或Starlette,框架通常提供了内置的序列化处理机制(基于Pydantic)。然而,如果你需要在中间件或日志记录中直接处理NumPy对象,上述ai_infra_json_encoder函数仍然是最佳选择。
FastAPI/Starlette 提示:
如果你使用的是FastAPI,并且想让框架全局知道如何处理NumPy类型,你可以将此逻辑集成到自定义的JSONResponse中,或配置底层的orjson库(如果使用)来处理NumPy。
例如,在标准库环境中,如果你想使用类继承的方式(更面向对象),可以这样做:
class AIJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (np.ndarray, np.generic)):
return obj.tolist()
if isinstance(obj, (datetime, date)):
return obj.isoformat()
if isinstance(obj, set):
return list(obj)
return json.JSONEncoder.default(self, obj)
# 使用类编码器
json.dumps(example_data, cls=AIJSONEncoder, indent=4)
通过采用这种自定义编码策略,我们可以确保无论模型返回的是NumPy数组还是标准Python类型,API都能可靠地将数据转换为有效的JSON响应,从而避免在生产环境中出现意外的序列化错误。
汤不热吧