Repository: lf-edge/ekuiper
Pull request #1400 (2022-09-13 03:52:00)

doc(plugin): tutorial to use python plugin
Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/docs/directory.json b/docs/directory.json
index 38a12d8b..77be8c35 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -93,6 +93,10 @@
 						{
 							"title": "使用 eKuiper 原生插件实现图像标注",
 							"path": "tutorials/ai/tensorflow_lite_tutorial"
+						},
+						{
+							"title": "使用 Python 函数插件运行 AI 算法",
+							"path": "tutorials/ai/python_tensorflow_lite_tutorial"
 						}
 					]
 				},
@@ -651,6 +655,10 @@
 						{
 							"title": "Label image by tensorflow lite model with eKuiper native plugin",
 							"path": "tutorials/ai/tensorflow_lite_tutorial"
+						},
+						{
+							"title": "Running AI Algorithms with Python Function Plugins",
+							"path": "tutorials/ai/python_tensorflow_lite_tutorial"
 						}
 					]
 				},
diff --git a/docs/en_US/tutorials/ai/python_tensorflow_lite_tutorial.md b/docs/en_US/tutorials/ai/python_tensorflow_lite_tutorial.md
new file mode 100644
index 00000000..8ea72a3a
--- /dev/null
+++ b/docs/en_US/tutorials/ai/python_tensorflow_lite_tutorial.md
@@ -0,0 +1,284 @@
+# Running AI Algorithms with Python Function Plugins
+
+[LF Edge eKuiper](https://www.lfedge.org/projects/ekuiper/) is an edge lightweight IoT data analytics / streaming
+software which can be run at all kinds of resource constrained IoT devices.
+
+[TensorFlow Lite](https://www.tensorflow.org/lite/guide) is a set of tools to help developers run TensorFlow models on
+mobile, embedded, and IoT devices. It enables on-device machine learning inference with low latency and a small binary
+size.
+
+By integrating eKuiper and TensorFlow Lite, users can analyze the data in stream by AI with prebuilt TensorFlow models.
+In this tutorial, we will walk you through building a eKuiper plugin to label pictures (binary data) produced by an edge
+device in stream by pre-trained image recognition TensorFlow model. In earlier [tutorial](./tensorflow_lite_tutorial.md), we have implemented the model inference using a GO language native plugin. In this tutorial, we will use a Python plugin to implement the similar functionality.
+
+The completed plugin package can be downloaded [here](https://github.com/lf-edge/ekuiper/blob/master/docs/resources/pyai.zip), which also contains the full source code.
+
+## Prerequisite
+
+To run the TensorFlow lite interpreter, we need a trained model. We won't cover how to train and cover a model in this tutorial, you can check [tflite converter](https://www.tensorflow.org/lite/convert) for how to do that. We can either train a new model or pick one online. In this tutorial, we will use the model in [TensorFlow image classification example](https://www.tensorflow.org/lite/examples/image_classification/overview).
+
+Before starting the tutorial, please prepare the following products or environments.
+1. Install the Python 3.x environment.
+2. Install the nng, ekuiper and tensorflow lite packages via `pip install nng ekuiper tflite_runtime`.
+
+By default, the portable plugin for eKuiper will run with the `python` command. If your environment does not support the `python` command, please use the [configuration file](../../operation/config/configuration_file.md#portable-plugin-config) to modify the Python command, such as `python3`.
+
+If you are developing with Docker, you can use the `lfedge/ekuiper:<tag>-slim-python` version. This version includes both the eKuiper and python environments.
+
+## Develop the Plugin
+
+To integrate eKuiper with TensorFlow Lite, we will develop a custom eKuiper function plugin to use with eKuiper rules. For example, we will create the `labelImage` function whose input is binary type data representing an image and whose output is a string representing the label of the image. So if the input image has a peacock in it, `labelImage(col)` will output `peacock`.
+
+To develop the function plugin, we need to：
+
+1. implement the business logic in Python and wrap it as an eKuiper function.
+2. Package the relevant files according to the plugin format.
+
+Create Python files that implement the extended interface (source, sink, or function). In this tutorial, we are developing a function plugin, so we need to implement the function extension interface.
+- Writing Python image classification functions
+- Wrapping an existing function as an eKuiper function plugin
+
+### Implement the business logic
+
+Our target function wants to take the binary data of an image as an input parameter, perform image preprocessing and other operations in the function, call the TensorFlow Lite model for inference, extract the most likely classification result from the inference result, and output it. We need to implement this function using Python in the same way as we would write a normal Python function.
+
+1. Download the [Image Classification Model](https://storage.googleapis.com/download.tensorflow.org/models/tflite/mobilenet_v1_1.0_224_quant_and_labels.zip), unzip it, and place it in the plug-in project. It contains a model file `mobilenet_v1_1.0_224.tflite` and a classification text file `labels.txt`. 
+2. Implement the image classification inference business logic. Create a Python file label.py, and implement the function `label(file_bytes)` in it.
+
+The label function will receive the base64 encoded image data passed by the eKuiper rule and perform inference on the classification. The pseudocode for its implementation is as follows:
+
+```python
+def label(file_bytes):
+    # Load the model file
+    interpreter = tf.Interpreter(
+        model_path= cwd + 'mobilenet_v1_1.0_224.tflite')
+    
+    # Preprocess the input image, turn it into tensors. Here the code is omitted.
+    
+    # Set model input, call inference, get result tensor
+    interpreter.set_tensor(input_details[0]['index'], input_data)
+    interpreter.invoke()
+    output_data = interpreter.get_tensor(output_details[0]['index'])
+    
+    # Post process the result and turn it into the output format. Here the code is omitted.
+    return result
+```
+
+The above code is only related to the business logic and can be tested without calling eKuiper's SDK. We just need to make sure that the input and output is of a type that can be converted to JSON format. For example, if the return value is a numpy array, it needs to be converted to a list type first. Developers can add main functions or unit tests to their business logic file or to another file for testing. For example, the following main function can be used to test the above business logic.
+
+```python
+# To test the logic
+if __name__ == '__main__':
+    file_name = "peacock.jpg"
+    with open(file_name, "rb") as f:
+        cwd = './'
+        result = label(base64.b64encode(f.read()))
+        print(json.dumps(result))
+```
+
+This file uses the `peacock.jpg` image file as input, calls the label function to test it, and converts the result to a json string and prints it. This allows us to see if the function works as expected. Here, we should get a json array and sort the recognition results by confidence level.
+
+```json
+[{"confidence": 0.9999935626983643, "label": "85:peacock"}, {"confidence": 2.156877371817245e-06, "label": "8:cock"}, {"confidence": 1.5930896779536852e-06, "label": "81:black grouse"}, {"confidence": 9.999589565268252e-07, "label": "92:coucal"}, {"confidence": 7.304166160793102e-07, "label": "96:jacamar"}]
+```
+
+See the `lable.py` file in the full code for details.
+
+### Plugin Implementation
+
+Like native plugins, Python plugins need to implement the corresponding interfaces; Python plugins also support the Source, Sink and Function interfaces, [interface definition](../../extension/portable/python_sdk.md#development) is similar to the native plugins. Here, what we need to implement is the function interface.
+
+Create the `label_func.py` function to wrap the functions implemented in the previous section. Import the Function class from eKuiper's plugin SDK and create the corresponding implementation class. The validate function is used to validate the parameters; is_aggregate is used to define whether the function is an aggregate function. The key implementations are in the exec function. Here, we take the data in the eKuiper stream as an argument, call the logic implemented above, and return the result to eKuiper.
+
+Note that the version of the eKuiper python SDK imported here should be the same as the target running version of eKuiper.
+
+```python
+from typing import List, Any
+
+from ekuiper import Function, Context
+
+from label import label
+
+# Inherit Function class from eKuiper SDK and implement it.
+class LabelImageFunc(Function):
+
+    def __init__(self):
+        pass
+
+    def validate(self, args: List[Any]):
+        if len(args) != 1:
+            return "invalid arg length"
+        return ""
+
+    def exec(self, args: List[Any], ctx: Context):
+        return label(args[0])
+
+    def is_aggregate(self):
+        return False
+
+# Create an instance for the class to be called by the entry function later
+labelIns = LabelImageFunc()
+```
+
+Once the code is implemented, we also need to add a description file for each function, which is placed in the functions directory. This plugin only implements one function, so you only need to create the `labelImage.json` file. Note that the file name is the name of the function and not the name of the plugin. This file will help the eKuiper manager to automatically produce the UI associated with the plugin.
+
+### Plugin Packaging
+
+At this point, we have completed the development of the main functionality, and next we need to package these files into a plugin format. Plugin packaging requires several steps to be completed.
+
+1. If the plugin has additional dependencies, such as TensorFlow Lite in this case, you need to create the dependency installation script `install.sh`. When the plugin is installed, eKuiper will look for an installation script file `install.sh` in the plugin package and execute the it if there is one. In this case, we create a `requirements.txt` file listing all the dependency packages. The installation of the dependencies is done in `install.sh` by calling `pip install -r $cur/requirements.txt`. For other plugins, you can reuse this script to update `requirements.txt` if you have no special requirements.
+2. Create a Python entry file that exposes all the implemented interfaces. Because multiple extensions can be implemented in a single plugin, you need an entry file that defines the implementation classes for each extension. The content is a main function, which is the entry point for the plugin runtime. It calls the methods in the SDK to define the plugin, including the plugin name, and a list of keys for the source, sink, and function implemented in the plugin. Here only a function plugin named `labelImage` is implemented, and its corresponding implementation method is `labelIns`. The Python plug-in process is independent of the eKuiper main process.
+    ```python
+    if __name__ == '__main__':
+        # Define the plugin
+        c = PluginConfig("pyai", {}, {},
+                         {"labelImage": lambda: labelIns})
+        # Start the plugin instance
+        plugin.start(c)
+    ```
+3. Create a plugin description file in JSON format to define the metadata of the plugin. The file name must be the same as the plugin name, i.e. `pyai.json`. The function names defined therein must correspond exactly to the entry file, and the contents of the file are as follows. Where, executable is used to define the name of the plugin's executable entry file.
+    ```json
+    {
+      "version": "v1.0.0",
+      "language": "python",
+      "executable": "pyai.py",
+      "sources": [
+      ],
+      "sinks": [
+      ],
+      "functions": [
+        "labelImage"
+      ]
+    }
+    ```
+
+At this point we have completed the development of the plugin, next we just need to package all the files in the directory into a zip file. The zip file should have a file structure similar to:
+
+- label.py
+- label_func.py
+- requirements.txt
+- model.tflite
+- install.sh
+- pyai.py
+- pyai.json
+- functions
+  - labelImage.json
+
+## Plugin Installation
+
+As the same with installing native plugins, we can also install Python plugins via the eKuiper manager UI or the REST API. To use the REST API, upload the zip file packaged above to the machine where eKuiper is located. Then use the following API to install it:
+
+```text
+### Install pyai plugin
+POST http://{{host}}/plugins/portables
+Content-Type: application/json
+
+{"name":"pyai", "file": "file:///tmp/pyai.zip"}
+```
+
+The installation process requires an Internet connection to download dependencies, including `tflite_runtime`, which may take a long time depending on the network conditions.
+
+## Run the plugin
+
+Once the plugin installed, we can use it in our rule. We will create a rule to receive image byte data from a mqtt topic and label the image by tflite model.
+
+### Define the stream
+
+Define the stream by eKuiper rest API. We create a stream named tfdemo whose format is binary and the topic is tfdemo.
+
+```shell
+POST http://{{host}}/streams
+Content-Type: application/json
+
+{"sql":"CREATE STREAM tfdemo () WITH (DATASOURCE=\"tfdemo\", FORMAT=\"BINARY\")"}
+```
+
+### 定义规则
+
+通过 eKuiper rest API 定义规则。 我们将创建一个名为 ruleTf 的规则。 我们只是从 tfdemo 流中读取图像，然后对其运行自定义函数 *labelImage*。 返回结果将是 AI 识别的图像的标签数组，包含按照置信度排名的标签。我们的规则取出其中第一个置信度最高的标签，并发送到 MQTT 主题 `ekuiper/labels`。
+
+### Define the rule
+
+Define the rule by eKuiper rest API.  We will create a rule named ruleTf. We just read the images from tfdemo stream and run the custom function *labelImage* against it. The returned result will be an array of labels of the images recognized by the AI, containing labels ranked by confidence. Our rule takes the first of these labels with the highest confidence and sends it to the MQTT topic `ekuiper/labels`.
+
+```shell
+POST http://{{host}}/rules
+Content-Type: application/json
+
+{
+  "id": "ruleTf",
+  "sql": "SELECT labelImage(self)[0]->label as label FROM tfdemo",
+  "actions": [
+   {
+      "mqtt":{
+        "server": "tcp://emqx.io:1883",
+        "sendSingle": true,
+        "topic": "ekuiper/labels"
+      }
+    }
+  ]
+}
+```
+
+### Feed the data
+
+Here we create a go program to send image data to the tfdemo topic to be processed by the rule.
+
+```go
+package main
+
+import (
+	"fmt"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"io/ioutil"
+	"time"
+)
+
+func main() {
+	const TOPIC = "tfdemo"
+
+	images := []string{
+		"peacock.png",
+		"frog.jpg",
+		// Other images
+	}
+	opts := mqtt.NewClientOptions().AddBroker("tcp://yourownhost:1883")
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		panic(token.Error())
+	}
+	for _, image := range images {
+		fmt.Println("Publishing " + image)
+		payload, err := ioutil.ReadFile(image)
+		if err != nil {
+			fmt.Println(err)
+			continue
+		}
+		if token := client.Publish(TOPIC, 0, false, payload); token.Wait() && token.Error() != nil {
+			fmt.Println(token.Error())
+		} else {
+			fmt.Println("Published " + image)
+		}
+		time.Sleep(1 * time.Second)
+	}
+	client.Disconnect(0)
+}
+
+```
+
+Run pub.go and it will start feeding images into the tfdemo theme. Please note that most MQTT brokers do not support transferring image files that are too large. In a practical scenario, we can adjust the size limit of the MQTT broker or use another source for the image stream input.
+
+### Check the result
+
+Since our rule definition has only one target: MQTT, the results will be written to the MQTT theme `ekuiper/labels`. Using an MQTT client subscribed to the theme, we input the tfdemo topic with two images *peacock.png* and *frog.png* and we will get two results.
+
+```shell
+{"label":"85:peacock"}
+{"label":"33:tailed frog, bell toad, ribbed toad, tailed toad, Ascaphus trui"}
+```
+
+The images are correctly labeled.
+
+## Conclusion
+
+In this tutorial, we guide you through building custom eKuiper Python plugins that implement a label feature for real-time image streams using pre-trained TensorFlow Lite models. Developers can replace the models with their own desired models to implement their own plug-ins.
\ No newline at end of file
diff --git a/docs/zh_CN/tutorials/ai/python_tensorflow_lite_tutorial.md b/docs/zh_CN/tutorials/ai/python_tensorflow_lite_tutorial.md
new file mode 100644
index 00000000..fb885751
--- /dev/null
+++ b/docs/zh_CN/tutorials/ai/python_tensorflow_lite_tutorial.md
@@ -0,0 +1,278 @@
+# 使用 Python 函数插件运行 AI 算法
+
+[LF Edge eKuiper](https://www.lfedge.org/projects/ekuiper/) 是一款边缘轻量级物联网数据分析/流软件，可在各种资源受限的物联网设备上运行。
+
+[TensorFlow Lite](https://www.tensorflow.org/lite/guide) 是一组帮助开发人员在移动端、嵌入式和物联网设备上运行 TensorFlow
+模型的工具，它使得设备上的机器学习预测具有低延迟和较小的二进制容量。
+
+通过集成 eKuiper 和 TensorFlow Lite，用户可以通过包含预先构建的 TensorFlow 模型分析流中的数据。 在本教程中，我们将引导您构建一个 eKuiper 插件，通过预先训练的图像识别 TensorFlow
+模型，标记边缘设备生成的流图片（二进制数据）。在早前的[教程](./tensorflow_lite_tutorial.md)，我们采用了 GO 语言原生插件的方式实现了模型调用。在本教程种，我们将使用 Python 插件实现类似的功能。
+
+教程完成的插件包可在[此处](https://github.com/lf-edge/ekuiper/blob/master/docs/resources/pyai.zip)下载，其中也包含了完整的源代码。
+
+## 先决条件
+
+如需运行 TensorFlow Lite 解释器，我们需要一个经过训练的模型。在本教程中，我们将不介绍如何训练和涵盖这个模型，您可以通过查看 [tflite converter](https://www.tensorflow.org/lite/convert) 了解如何做到这一点。我们既可以训练一个新的模型，也可以在线选择一个。在本教程中，我们将使用 [TensorFlow 图像分类示例](https://www.tensorflow.org/lite/examples/image_classification/overview) 中的图像分类模型。
+
+开始教程之前，请准备以下产品或环境：
+1. 安装 Python 3.x 环境。
+2. 通过 `pip install nng ekuiper tflite_runtime` 安装 nng，ekuiper 和 tensorflow lite 包。
+
+默认情况下，eKuiper 的 portable 插件运行时会通过 `python` 命令来运行插件。如果您的环境不支持 `python` 命令，请通过[配置文件](../../operation/config/configuration_file.md#portable-插件配置)更换为可用的 Python 命令，如 `python3`。
+
+若使用 Docker 进行开发，可使用 `lfedge/ekuiper:<tag>-slim-python` 版本。该版本包含 eKuiper 和 python 环境，无需再手动安装。
+
+## 开发插件
+
+为了集成 eKuiper 和 TensorFlow Lite，我们将开发一个定制的 eKuiper 函数插件，供 eKuiper 规则使用。例如，我们将创建 `labelImage`  函数，其输入是表示图像的二进制类型数据，输出是表示图像标签的字符串。所以，如果输入图像中有孔雀，`labelImage(col)` 将输出 `peacock`。
+
+要开发函数插件，我们需要：
+
+1. 通过 Python 实现业务逻辑并包装为 eKuiper 函数。
+2. 按照插件格式对相关文件进行打包。
+
+创建 Python 文件实现扩展接口（源，sink 或函数）。在本例中，我们要开发的是函数插件， 因此需要实现函数扩展接口。
+- 编写 Python 图像分类函数
+- 包装已有函数为 eKuiper 函数插件
+
+
+### 实现业务逻辑
+
+我们的目标函数希望接收一个图像的二进制数据作为输入参数，在函数中进行图像的预处理等操作，调用 TensorFlow Lite 模型进行推理，从推理结果中提取出可能性最大的分类结果并输出。我们需要使用 Python ，跟编写普通的 Python 函数相同的方式实现这个函数。
+
+1. 下载[图像分类模型](https://storage.googleapis.com/download.tensorflow.org/models/tflite/mobilenet_v1_1.0_224_quant_and_labels.zip)，解压后放置在插件项目中。里面包含一个模型文件 `mobilenet_v1_1.0_224.tflite` 和分类文本文件 `labels.txt`。
+2. 实现图像分类推理业务逻辑。创建一个 Python 文件 label.py, 在其中实现函数 `label(file_bytes)`。
+
+label 函数将接收由 eKuiper 规则传过来的 base64 编码的图像数据并进行推理分类。其实现的伪代码如下：
+
+```python
+def label(file_bytes):
+    # 载入模型文件
+    interpreter = tf.Interpreter(
+        model_path= cwd + 'mobilenet_v1_1.0_224.tflite')
+    
+    # 预处理输入图片，将其转换为输入的 tensor 格式，此处代码省略
+    
+    # 设置模型输入，调用推理，拿到结果 tensor
+    interpreter.set_tensor(input_details[0]['index'], input_data)
+    interpreter.invoke()
+    output_data = interpreter.get_tensor(output_details[0]['index'])
+    
+    # 对结果进行后处理，转换为输出格式，代码省略
+    return result
+```
+
+以上代码仅跟业务逻辑相关，无需调用 eKuiper 的 SDK 即可进行测试。我们只需要保证输入输出为可转为 JSON 格式的类型即可。例如，若返回值为 numpy array，需先转换成 list 类型。开发者可在自己的业务逻辑文件或另外的文件中添加 main 函数或单元测试进行测试。例如，可采用以下 main 函数测试以上业务逻辑。
+
+```python
+# To test the logic
+if __name__ == '__main__':
+    file_name = "peacock.jpg"
+    with open(file_name, "rb") as f:
+        cwd = './'
+        result = label(base64.b64encode(f.read()))
+        print(json.dumps(result))
+```
+
+该文件使用了 `peacock.jpg` 图像文件作为输入，调用了 label 函数进行测试，并将结果转换为 json 字符串并打印。从而可以观察函数运行结果是否符合预期。此处，我们应当得到一个 json 数组，按照置信度的高低对识别结果进行排序。
+
+```json
+[{"confidence": 0.9999935626983643, "label": "85:peacock"}, {"confidence": 2.156877371817245e-06, "label": "8:cock"}, {"confidence": 1.5930896779536852e-06, "label": "81:black grouse"}, {"confidence": 9.999589565268252e-07, "label": "92:coucal"}, {"confidence": 7.304166160793102e-07, "label": "96:jacamar"}]
+```
+
+详细信息请参见完整代码中的 `lable.py` 文件。
+
+### 插件实现
+
+与原生插件相同，Python 插件也需要实现对应的接口。Python 插件也支持 Source, Sink 和 Function 接口，[接口定义](../../extension/portable/python_sdk.md#插件开发)与原生插件类似。此处，我们需要实现的是函数接口。
+
+创建 `label_func.py` 函数对上节实现的函数进行封装。导入 eKuiper 的插件 SDK 中的 Function 类，创建对应的实现类。其中，validate 函数用于参数的验证；is_aggregate 用于定义函数是否为聚合函数。关键的实现都在 exec 函数中。此处，我们将 eKuiper 流中的数据作为参数，调用上文实现的逻辑，并将结果返回到 eKuiper 中。
+
+注意此处导入的 eKuiper python SDK 版本应与最终运行的 eKuiper 版本形同。
+
+```python
+from typing import List, Any
+
+from ekuiper import Function, Context
+
+from label import label
+
+# 继承 SDK 中的 Function 类，并实现其中的方法
+class LabelImageFunc(Function):
+
+    def __init__(self):
+        pass
+
+    def validate(self, args: List[Any]):
+        if len(args) != 1:
+            return "invalid arg length"
+        return ""
+
+    def exec(self, args: List[Any], ctx: Context):
+        return label(args[0])
+
+    def is_aggregate(self):
+        return False
+
+# 创建类的一个对象，供入口函数调用
+labelIns = LabelImageFunc()
+```
+
+代码实现完成后，我们还需要给每个函数添加一个描述文件，置于 functions 目录下。本插件只实现了一个函数，因此只需要创建 `labelImage.json` 文件。注意，文件名为函数的名字，而非插件的名字。该文件可以帮助 eKuiper manager 自动生产插件相关的 UI。
+
+### 插件打包
+
+至此，我们已经完成了主要功能的开发，接下来需要将这些文件打包成插件的格式。插件打包需要完成几个步骤：
+
+1. 如果插件有额外的依赖，例如本例中的 TensorFlow Lite, 需要创建依赖安装脚本 `install.sh`。插件安装时，eKuiper 会查找插件包中是否有安装脚本文件 `install.sh`，若有的话执行安装脚本。在本例中，我们创建一个 `requirements.txt` 文件列出所有的依赖包。在 `install.sh` 通过调用 `pip install -r $cur/requirements.txt` 完成依赖的安装。对于别的插件，若无特殊要求可重用该脚本，更新`requirements.txt`即可。
+2. 创建 Python 入口文件，用于暴露所有实现的接口。因为在单个插件中可以实现多个扩展，所以需要一个入口文件定义各个扩展的实现类。其内容为一个 main 函数，为插件运行时入口。它调用 SDK 里的方法定义插件，包括插件名，插件里实现的 source, sink, function 的键值列表。此处仅实现一个名为 `labelImage` 的函数插件，其对应的实现方法为 `labelIns`。之后调用 start 方法启动插件进程的运行。Python 插件进程独立于 eKuiper 主进程。
+    ```python
+    if __name__ == '__main__':
+        # 定义插件
+        c = PluginConfig("pyai", {}, {},
+                         {"labelImage": lambda: labelIns})
+        # 启动插件
+        plugin.start(c)
+    ```
+3. 创建 JSON 格式的插件描述文件，用于定义插件的元数据。文件名必须与插件名相同，即 `pyai.json`。其中定义的函数名与入口文件必须完全对应，文件内容如下。其中，executable 用于定义插件的可执行入口文件名。
+    ```json
+    {
+      "version": "v1.0.0",
+      "language": "python",
+      "executable": "pyai.py",
+      "sources": [
+      ],
+      "sinks": [
+      ],
+      "functions": [
+        "labelImage"
+      ]
+    }
+    ```
+
+至此我们已经完成了插件的开发，接下来只需要把目录中的所有文件打包成 zip 文件即可。 zip文件的文件结构应类似于：
+
+- label.py
+- label_func.py
+- requirements.txt
+- model.tflite
+- install.sh
+- pyai.py
+- pyai.json
+- functions
+  - labelImage.json
+
+## 插件安装
+
+与安装原生插件相同，我们也可以通过管理控制台或者 REST API 进行 Python 插件的安装。使用 REST API 的时候，把上文打包的 zip 文件上传到 eKuiper 所在的机器中。然后使用如下 API 进行安装。
+
+```text
+### Install pyai plugin
+POST http://{{host}}/plugins/portables
+Content-Type: application/json
+
+{"name":"pyai", "file": "file:///tmp/pyai.zip"}
+```
+
+安装过程中需要联网下载依赖，包括 `tflite_runtime`，视网络情况可能安装过程需要较长时间。
+
+## 运行插件
+
+插件安装后，我们就可以在规则中使用它了。 我们将创建一个规则用于接收来自 MQTT 主题的图像字节数据，并通过 tflite 模型标记该图像。
+
+### 定义流
+
+通过 eKuiper rest API 定义流。 我们创建一个名为 tfdemo 的流，其格式为二进制，主题为 tfdemo。
+
+```shell
+POST http://{{host}}/streams
+Content-Type: application/json
+
+{"sql":"CREATE STREAM tfdemo () WITH (DATASOURCE=\"tfdemo\", FORMAT=\"BINARY\")"}
+```
+
+### 定义规则
+
+通过 eKuiper rest API 定义规则。 我们将创建一个名为 ruleTf 的规则。 我们只是从 tfdemo 流中读取图像，然后对其运行自定义函数 *labelImage*。 返回结果将是 AI 识别的图像的标签数组，包含按照置信度排名的标签。我们的规则取出其中第一个置信度最高的标签，并发送到 MQTT 主题 `ekuiper/labels`。
+
+```shell
+POST http://{{host}}/rules
+Content-Type: application/json
+
+{
+  "id": "ruleTf",
+  "sql": "SELECT labelImage(self)[0]->label as label FROM tfdemo",
+  "actions": [
+   {
+      "mqtt":{
+        "server": "tcp://emqx.io:1883",
+        "sendSingle": true,
+        "topic": "ekuiper/labels"
+      }
+    }
+  ]
+}
+```
+
+### 输入数据
+
+在这里，我们创建了一个 go 程序，用于将图像数据发送到 tfdemo 主题以便由规则进行处理。
+
+```go
+package main
+
+import (
+	"fmt"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"io/ioutil"
+	"time"
+)
+
+func main() {
+	const TOPIC = "tfdemo"
+
+	images := []string{
+		"peacock.png",
+		"frog.jpg",
+		// 其他你需要的图像
+	}
+	opts := mqtt.NewClientOptions().AddBroker("tcp://yourownhost:1883")
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		panic(token.Error())
+	}
+	for _, image := range images {
+		fmt.Println("Publishing " + image)
+		payload, err := ioutil.ReadFile(image)
+		if err != nil {
+			fmt.Println(err)
+			continue
+		}
+		if token := client.Publish(TOPIC, 0, false, payload); token.Wait() && token.Error() != nil {
+			fmt.Println(token.Error())
+		} else {
+			fmt.Println("Published " + image)
+		}
+		time.Sleep(1 * time.Second)
+	}
+	client.Disconnect(0)
+}
+
+```
+
+运行 pub.go，它将开始将图像输入 tfdemo 主题。请注意，大部分的 MQTT broker 不支持传输太大的图像文件。实际场景中，我们可以调整 MQTT broker 的大小限制或者使用别的 source 进行图像流的输入。
+
+### 检查结果
+
+因为我们的规则定义只有一个目标：MQTT，所以结果将写入MQTT 主题 `ekuiper/labels`。使用 MQTT 客户端订阅该主题，我们用 *peacock.png* 和 *frog.png* 两个图像输入 tfdemo 主题，我们将得到两个结果。
+
+```shell
+{"label":"85:peacock"}
+{"label":"33:tailed frog, bell toad, ribbed toad, tailed toad, Ascaphus trui"}
+```
+
+图像标记正确。
+
+## 结论
+
+在本教程中，我们引导您构建自定义的 eKuiper Python 插件，以利用预先训练好的 TensorFlow Lite 模型，实现了实时图像流的分类标注功能。开发者可以将模型替换为自己所需的模型，实现自己的插件。
\ No newline at end of file
diff --git a/etc/ops/filter.json b/etc/ops/filter.json
new file mode 100644
index 00000000..dc0706ee
--- /dev/null
+++ b/etc/ops/filter.json
@@ -0,0 +1,52 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to filter out rows based on a filter condition.",
+      "zh_CN": "用于过滤数据流的操作"
+    }
+  },
+  "properties": [{
+    "name": "expr",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "filter condition expression",
+      "zh_CN": "过滤条件语句"
+    },
+    "label": {
+      "en_US": "Condition",
+      "zh_CN": "条件"
+    }
+  }],
+  "node": {
+    "display": true,
+    "category": "op",
+    "input": {
+      "type": "any",
+      "rowType": "any",
+      "collectionType": "any"
+    },
+    "output": {
+      "type": "same",
+      "strategy": "keep"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Filter",
+      "zh": "过滤"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/function.json b/etc/ops/function.json
new file mode 100644
index 00000000..e8dce08b
--- /dev/null
+++ b/etc/ops/function.json
@@ -0,0 +1,52 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to invoke a function",
+      "zh_CN": "用于调用函数的操作"
+    }
+  },
+  "properties": [{
+    "name": "expr",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "Function call expression",
+      "zh_CN": "函数调用语句"
+    },
+    "label": {
+      "en_US": "Expression",
+      "zh_CN": "函数语句"
+    }
+  }],
+  "node": {
+    "display": false,
+    "category": "op",
+    "input": {
+      "type": "any",
+      "rowType": "any",
+      "collectionType": "single"
+    },
+    "output": {
+      "type": "same",
+      "strategy": "append"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Function",
+      "zh": "函数"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/groupby.json b/etc/ops/groupby.json
new file mode 100644
index 00000000..d2e648d2
--- /dev/null
+++ b/etc/ops/groupby.json
@@ -0,0 +1,52 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to group the events by the condition.",
+      "zh_CN": "用于按条件对事件进行分组的操作。"
+    }
+  },
+  "properties": [{
+    "name": "dimensions",
+    "default": "[]",
+    "optional": false,
+    "control": "text",
+    "type": "list_string",
+    "hint": {
+      "en_US": "the dimension fields to group by",
+      "zh_CN": "分组的维度字段"
+    },
+    "label": {
+      "en_US": "Dimensions",
+      "zh_CN": "维度"
+    }
+  }],
+  "node": {
+    "display": true,
+    "category": "op",
+    "input": {
+      "type": "collection",
+      "rowType": "single",
+      "collectionType": "any"
+    },
+    "output": {
+      "type": "collection",
+      "strategy": "grouped"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Group By",
+      "zh": "分组"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/join.json b/etc/ops/join.json
new file mode 100644
index 00000000..b6322642
--- /dev/null
+++ b/etc/ops/join.json
@@ -0,0 +1,119 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to combine rows from two or more tables, based on a related column between them.",
+      "zh_CN": "用于基于两个或多个表之间的相关列连接行的操作。"
+    }
+  },
+  "properties": [{
+    "name": "from",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "string",
+    "hint": {
+      "en_US": "From table name",
+      "zh_CN": "来源表的名字"
+    },
+    "label": {
+      "en_US": "From",
+      "zh_CN": "来源表"
+    }
+  },{
+    "name": "joins",
+    "optional": false,
+    "control": "list",
+    "type": "list_object",
+    "hint": {
+      "en_US": "Join conditions",
+      "zh_CN": "连接条件"
+    },
+    "label": {
+      "en_US": "Joins",
+      "zh_CN": "连接"
+    },
+    "default": [
+      {
+        "name": "name",
+        "default": "",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "Join table name",
+          "zh_CN": "连接的表名"
+        },
+        "label": {
+          "en_US": "Table Name",
+          "zh_CN": "表名"
+        }
+      },
+      {
+        "name": "type",
+        "default": "",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "values": [
+          "inner",
+          "left",
+          "right",
+          "full",
+          "cross"
+        ],
+        "hint": {
+          "en_US": "Join type",
+          "zh_CN": "连接类型"
+        },
+        "label": {
+          "en_US": "Type",
+          "zh_CN": "类型"
+        }
+      },
+      {
+        "name": "on",
+        "default": "",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "join condition expression",
+          "zh_CN": "连接条件语句"
+        },
+        "label": {
+          "en_US": "Condition",
+          "zh_CN": "条件"
+        }
+      }
+    ]
+  }],
+  "node": {
+    "display": true,
+    "category": "op",
+    "input": {
+      "type": "collection",
+      "rowType": "single",
+      "collectionType": "single"
+    },
+    "output": {
+      "type": "collection",
+      "strategy": "append"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Join",
+      "zh": "连接"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/orderby.json b/etc/ops/orderby.json
new file mode 100644
index 00000000..9f8d57f1
--- /dev/null
+++ b/etc/ops/orderby.json
@@ -0,0 +1,83 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to order the rows.",
+      "zh_CN": "用于排序的操作"
+    }
+  },
+  "properties": [{
+    "name": "sorts",
+    "optional": false,
+    "control": "list",
+    "type": "list_object",
+    "hint": {
+      "en_US": "order expression",
+      "zh_CN": "排序语句"
+    },
+    "label": {
+      "en_US": "Order",
+      "zh_CN": "排序"
+    },
+    "default": [
+      {
+        "name": "field",
+        "default": "",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "Order by field",
+          "zh_CN": "用于排序的字段名"
+        },
+        "label": {
+          "en_US": "Field",
+          "zh_CN": "字段"
+        }
+      },
+      {
+        "name": "order",
+        "default": false,
+        "optional": false,
+        "control": "checkbox",
+        "type": "boolean",
+        "hint": {
+          "en_US": "Order by descending",
+          "zh_CN": "是否降序"
+        },
+        "label": {
+          "en_US": "Descending",
+          "zh_CN": "降序"
+        }
+      }
+    ]
+  }],
+  "node": {
+    "display": true,
+    "category": "op",
+    "input": {
+      "type": "any",
+      "rowType": "any",
+      "collectionType": "any"
+    },
+    "output": {
+      "type": "same",
+      "strategy": "keep"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Filter",
+      "zh": "过滤"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/pick.json b/etc/ops/pick.json
new file mode 100644
index 00000000..1ddd694f
--- /dev/null
+++ b/etc/ops/pick.json
@@ -0,0 +1,52 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to pick the selected fields.",
+      "zh_CN": "用于选取字段的操作"
+    }
+  },
+  "properties": [{
+    "name": "fields",
+    "default": "",
+    "optional": false,
+    "control": "text",
+    "type": "list_string",
+    "hint": {
+      "en_US": "select fields",
+      "zh_CN": "选取字段"
+    },
+    "label": {
+      "en_US": "Fields",
+      "zh_CN": "字段"
+    }
+  }],
+  "node": {
+    "display": true,
+    "category": "op",
+    "input": {
+      "type": "any",
+      "rowType": "any",
+      "collectionType": "any"
+    },
+    "output": {
+      "type": "same",
+      "strategy": "pick"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Pick",
+      "zh": "选择"
+    }
+  }
+}
\ No newline at end of file
diff --git a/etc/ops/window.json b/etc/ops/window.json
new file mode 100644
index 00000000..1e6c246c
--- /dev/null
+++ b/etc/ops/window.json
@@ -0,0 +1,110 @@
+{
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/query_language_elements.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/query_language_elements.md"
+    },
+    "description": {
+      "en_US": "An operation to create a streaming window",
+      "zh_CN": "用于创建窗口的操作"
+    }
+  },
+  "properties": [
+    {
+      "name": "type",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "values": [
+        "tumblingwindow",
+        "hoppingwindow",
+        "slidingwindow",
+        "sessionwindow",
+        "countwindow"
+      ],
+      "hint": {
+        "en_US": "window type",
+        "zh_CN": "窗口类型"
+      },
+      "label": {
+        "en_US": "Window Type",
+        "zh_CN": "窗口类型"
+      }
+    },{
+      "name": "unit",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "values": [
+        "ms",
+        "ss",
+        "mi",
+        "hh",
+        "dd"
+      ],
+      "hint": {
+        "en_US": "Time unit for the window",
+        "zh_CN": "窗口的时间单位"
+      },
+      "label": {
+        "en_US": "Time Unit",
+        "zh_CN": "时间单位"
+      }
+    },{
+      "name": "size",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "int",
+      "hint": {
+        "en_US": "Window Length",
+        "zh_CN": "窗口长度"
+      },
+      "label": {
+        "en_US": "Window Length",
+        "zh_CN": "窗口长度"
+      }
+    },{
+      "name": "interval",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "int",
+      "hint": {
+        "en_US": "Window trigger interval",
+        "zh_CN": "窗口触发周期"
+      },
+      "label": {
+        "en_US": "Window Trigger Interval",
+        "zh_CN": "窗口触发周期"
+      }
+  }],
+  "node": {
+    "display": false,
+    "category": "op",
+    "input": {
+      "type": "row",
+      "rowType": "any",
+      "collectionType": "any",
+      "allowMulti": true
+    },
+    "output": {
+      "type": "collection",
+      "strategy": "append"
+    },
+    "icon": "iconPath",
+    "label": {
+      "en": "Window",
+      "zh": "窗口"
+    }
+  }
+}
\ No newline at end of file
diff --git a/internal/topo/graph/node.go b/internal/topo/graph/node.go
index 68ae5810..308a7a1d 100644
--- a/internal/topo/graph/node.go
+++ b/internal/topo/graph/node.go
@@ -49,6 +49,6 @@ type Groupby struct {
 type Orderby struct {
 	Sorts []struct {
 		Field string `json:"field"`
-		Order string `json:"order"`
+		Desc  bool   `json:"desc"`
 	}
 }
diff --git a/internal/topo/planner/planner_graph.go b/internal/topo/planner/planner_graph.go
index dd09df1c..c3f52d5e 100644
--- a/internal/topo/planner/planner_graph.go
+++ b/internal/topo/planner/planner_graph.go
@@ -285,7 +285,10 @@ func parseOrderBy(props map[string]interface{}) (*operator.OrderOp, error) {
 	}
 	stmt := "SELECT * FROM unknown ORDER BY"
 	for _, s := range n.Sorts {
-		stmt += " " + s.Field + " " + s.Order
+		stmt += " " + s.Field + " "
+		if s.Desc {
+			stmt += "DESC"
+		}
 	}
 	p, err := xsql.NewParser(strings.NewReader(stmt)).Parse()
 	if err != nil {


Test patch:
diff --git a/test/graph_group_order_rule.jmx b/test/graph_group_order_rule.jmx
index 57d5c9e0..4c339ce1 100644
--- a/test/graph_group_order_rule.jmx
+++ b/test/graph_group_order_rule.jmx
@@ -190,7 +190,7 @@
         &quot;props&quot;: {&#xd;
           &quot;sorts&quot;: [{&#xd;
             &quot;field&quot;: &quot;count&quot;,&#xd;
-            &quot;order&quot;: &quot;desc&quot;&#xd;
+            &quot;desc&quot;: true&#xd;
           }]&#xd;
         }&#xd;
       },&#xd;