--- title: 实现过程 description: "本节详细介绍通过构造仅包含待验证算子的单算子网络进行算子验证的方法。" url: https://www.hiascend.com/document/detail/zh/canncommercial/latest/others/tbeaicpudevg/atlasopdev_10_0105.html sourcePath: /source/zh/canncommercial/900/others/tbeaicpudevg/atlasopdev_10_0105.html indexId: b5a8eb06e3ede0d8a28fadd8ebb535e7e26615d66a0653d031242f09b0b2869373 --- # 实现过程 #### 简介 本节详细介绍通过构造仅包含待验证算子的单算子网络进行算子验证的方法。 #### 验证思路 图1 验证思路 1. 开发者进行算子开发并将相关交付件部署到Host侧，包括算子实现文件、算子插件（仅当测试前端为TensorFlow网络时需要）、算子原型库、算子信息库。 2. 选择一个需要测试的用例，明确shape、dtype等。 3. 使用numpy或其他工具生成随机的数据作为算子输入数据。 4. 使用输入数据，在TensorFlow环境或用numpy模拟算子计算流程，执行此算子，得到算子的预期运行结果。 5. 通过TensorFlow前端构造只包含当前已开发的TBE算子的单算子网络，并把 3 数据作为输入，执行单算子网络。TensorFlow 调用GE/FE等，最终在AI处理器 所在硬件环境上执行单算子网络，得到算子的实际运行结果，大致过程如下： a. TF Adapter适配和插件加载：如果使用TensorFlow前端，会调用TF Adapter适配的接口生成TensorFlow原图，并下发到GE，由GE初始化过程加载，调用算子插件解析算子，并映射为昇腾AI处理器支持的图。 b. GE调用算子原型库中的infershape()/verify()做shape推导、参数合法性校验。 c. FE加载算子信息库，算子信息库中注册了算子信息，该文件指定了算子在AI处理器上执行过程中的入参能够支持shape、type、format信息。 d. FE调用TBE算子库中的算子进行UB融合，将算子实现文件编译生成算子kernel。 e. GE将从FE获取的taskinfo下发给RunTime，并传递到Device侧，在AI处理器上执行单算子网络，得到算子的实际运行结果。 f. 把算子在AI处理器上的执行结果和 4 的预期结果进行对比，查看功能是否正确、精度是否满足要求。 #### 操作步骤（TensorFlow 1.15版本） 本节以Add算子为例，介绍构造单算子的TensorFlow 1.15版本网络并进行验证的方法，完整的代码样例获取方法：单击Gitee(https://gitee.com/ascend/samples/)或Github(https://github.com/Ascend/samples)，进入Ascend samples样例仓，参见README中的“版本说明”下载配套版本的sample包，从“cplusplus/level1_single_api/4_op_dev/1_custom_op/tbe/testcases/tf1.15_test/add”目录中获取样例。 1. 进入“tbe/testcases/tf1.15_test/算子名称”目录，编写测试用例“tf_add.py”。 2. 导入python库。 1 2 3 4 import logging # Python标准库日志模块 import tensorflow as tf # 导入TensorFlow开源库 from npu_bridge.estimator import npu_ops # 导入TensorFlow开源库中的npu_ops模块 import numpy as np # 导入Python的数学基础库 3. 设置np.allclose比较函数的公差参数。 1 2 3 4 #np.allclose比较函数的相对公差参数 atol = 0.001 #np.allclose比较函数的绝对公差参数 rtol = 0.001 4. 通过config()定义昇腾AI处理器和cpu上的运行参数。 当“execute_type”为“ai_core”时，代表在AI处理器上运行单算子网络，调用的是TBE算子。 当“execute_type”为“cpu”时，代表在Host侧的CPU运行单算子网络，调用的是TensorFlow算子。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 def config(execute_type): if execute_type == 'ai_core': session_config = tf.ConfigProto( allow_soft_placement=True, log_device_placement=False,) custom_op = session_config.graph_options.rewrite_options.custom_optimizers.add() custom_op.name = "NpuOptimizer" custom_op.parameter_map["enable_data_pre_proc"].b = True # 开启数据预处理下沉到Device侧执行 custom_op.parameter_map["mix_compile_mode"].b = True custom_op.parameter_map["use_off_line"].b = True # True表示在AI处理器上执行训练 elif execute_type == 'cpu': session_config = tf.ConfigProto( allow_soft_placement=True, log_device_placement=False) return session_config 5. 单算子网络测试用例主函数。 ``` def main(unused_argv): shape_params = (2, 2, 2) dtype_params = np.float16 # 构造Add算子的两个输入数据,shape为shape_params，范围在[-2,2]之间的随机数 x_data = np.random.uniform(-2, 2, size=shape_params).astype(dtype_params) y_data = np.random.uniform(-2, 2, size=shape_params).astype(dtype_params) # 分别对Add算子的两个输入数据进行占位 x = tf.compat.v1.placeholder(dtype_params, shape=shape_params) y = tf.compat.v1.placeholder(dtype_params, shape=shape_params) # 计算算子输出 out = tf.math.add(x, y) # 在Host侧CPU上运行单算子，得到期望运行结果 with tf.compat.v1.Session(config=config('cpu')) as session: result_cpu = session.run(out, feed_dict={x: x_data, y: y_data}) # 在昇腾AI处理器上运行单算子，得到实际运行结果 with tf.compat.v1.Session(config=config('ai_core')) as session: result_ai_core = session.run(out, feed_dict={x: x_data, y: y_data}) np.array(result_ai_core).astype(dtype_params) np.array(result_cpu).astype(dtype_params) print('====================================') # 通过np.allclose比较昇腾AI处理器上运行的实际结果和cpu上运行的期望结果，其中atol和rtol为np.allclose比较函数的相对公差参数和绝对公差参数，请见步骤3设置。 cmp_result = np.allclose(result_ai_core, result_cpu, atol, rtol) print(cmp_result) print('====================================') ``` - 算子输入请根据算子实际输入个数及shape进行构造。 - 算子输出的计算，请根据算子逻辑调用TensorFlow相关接口进行实现。 6. 运行单算子网络。 1 2 if __name__ == "__main__": tf.app.run() #### 操作步骤（TensorFlow 2.6版本） 本节以Add算子为例，介绍构造单算子的TensorFlow 2.6版本网络并进行验证的方法，完整的代码样例获取方法：单击Gitee(https://gitee.com/ascend/samples/)或Github(https://github.com/Ascend/samples)，进入Ascend samples样例仓，参见README中的“版本说明”下载配套版本的sample包，从“cplusplus/level1_single_api/4_op_dev/1_custom_op/tbe/testcases/tf2.6_test/add”目录中获取样例。 1. 进入“tbe/testcases/tf2.6_test/算子名称”目录，编写测试用例“tf_add.py”。 2. 导入python库。 1 2 3 4 import logging # Python标准库日志模块 import tensorflow as tf # 导入TensorFlow开源库 from npu_device.compat.v1 import * # 导入NPU相关库 import numpy as np # 导入Python的数学基础库 3. 禁用TensorFlow 2.x行为并设置np.allclose比较函数的公差参数。 1 2 3 4 5 6 # 禁用TensorFlow 2.x行为 tf.compat.v1.disable_v2_behavior() # np.allclose比较函数的相对公差参数 atol = 0.001 # np.allclose比较函数的绝对公差参数 rtol = 0.001 4. 通过config()定义昇腾AI处理器和cpu上的运行参数。 当“execute_type”为“ai_core”时，代表在AI处理器上运行单算子网络，调用的是TBE算子。 当“execute_type”为“cpu”时，代表在Host侧的CPU运行单算子网络，调用的是TensorFlow算子。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 def config(execute_type): if execute_type == 'ai_core': session_config = tf.compat.v1.ConfigProto( allow_soft_placement=True, log_device_placement=False,) custom_op = session_config.graph_options.rewrite_options.custom_optimizers.add() custom_op.name = "NpuOptimizer" custom_op.parameter_map["enable_data_pre_proc"].b = True # 开启数据预处理下沉到Device侧执行 custom_op.parameter_map["mix_compile_mode"].b = True custom_op.parameter_map["use_off_line"].b = True # True表示在AI处理器上执行训练 custom_op.parameter_map["min_group_size"].b = 1 elif execute_type == 'cpu': session_config = tf.compat.v1.ConfigProto( allow_soft_placement=True, log_device_placement=False) return session_config 5. 单算子网络测试用例主函数。 ``` def main(unused_argv): shape_params = (2, 2, 2) dtype_params = np.float16 # 构造Add算子的两个输入数据,shape为shape_params，范围在[-2,2]之间的随机数 x_data = np.random.uniform(-2, 2, size=shape_params).astype(dtype_params) y_data = np.random.uniform(-2, 2, size=shape_params).astype(dtype_params) # 分别对Add算子的两个输入数据进行占位 x = tf.compat.v1.placeholder(dtype_params, shape=shape_params) y = tf.compat.v1.placeholder(dtype_params, shape=shape_params) # 计算算子输出 out = tf.math.add(x, y) # 在Host侧CPU上运行单算子，得到期望运行结果 with tf.compat.v1.Session(config=config('cpu')) as session: result_cpu = session.run(out, feed_dict={x: x_data, y: y_data}) # 在昇腾AI处理器上运行单算子，得到实际运行结果 with tf.compat.v1.Session(config=config('ai_core')) as session: result_ai_core = session.run(out, feed_dict={x: x_data, y: y_data}) np.array(result_ai_core).astype(dtype_params) np.array(result_cpu).astype(dtype_params) print('====================================') # 通过np.allclose比较昇腾AI处理器上运行的实际结果和cpu上运行的期望结果，其中atol和rtol为np.allclose比较函数的相对公差参数和绝对公差参数，请见步骤3设置。 cmp_result = np.allclose(result_ai_core, result_cpu, atol, rtol) print(cmp_result) print('====================================') ``` - 算子输入请根据算子实际输入个数及shape进行构造。 - 算子输出的计算，请根据算子逻辑调用TensorFlow相关接口进行实现。 6. 运行单算子网络。 1 2 if __name__ == "__main__": tf.compat.v1.app.run()