AI算法的快速应用落地
本文将为你介绍使用DTS调度AI算法服务的流程,实现AI算法的快速应用落地。首先需要先把AI算法进行服务化,然后在DTS的产品-服务中进行定义,接下来我们就可以在任务管理中快速进行调度和使用了。
包含以下几个流程:
网关添加服务地址、创建产品、定义产品Profile、创建任务、结果验证
示例通过从摄像头定时获取图片进行人数分析的流程演示整个过程。
准备事项
1.注册数字孪生服务DTService(以下简称DTS)账户
2.准备APIfox,用于模拟设备(或其他可以发送Http请求的软件,如postman、vs等)。实际使用时,设备需支持反向控制。
3.成功部署看门狗服务,连接实时摄像头或演示视频数据;
4.成功把AI算法进行服务化
示例
AI算法服务化
此步骤为简单可实现的AI算法服务化方法,可根据实际使用需要进行调整或修改。
-
在服务器或本地机器上安装Python 3.8+环境。然后使用pip安装核心库,包括Ultralytics(用于YOLO模型)、FastAPI(用于构建API)、Uvicorn(作为API服务器)、以及OpenCV或Pillow(用于图片处理)。
-
从Ultralytics官网下载预训练YOLO模型(如yolov11.pt),或使用自定义训练的模型。通过Ultralytics API加载模型(例如model = YOLO(‘yolov11n.pt’)),确保模型能处理图片输入并输出检测结果(如边界框、类别、置信度)。
-
定义图片输入格式(支持JPEG/PNG等),并准备输出格式(如JSON,包含检测结果)。
-
使用FastAPI创建一个简单的Web应用。定义一个POST端点(如/predict),允许用户上传资源文件。
-
在端点处理函数中,接收上传的资源(如图片),传入YOLO模型进行推理(例如调用model.predict(image)),提取结果(如对象位置和标签),并转换为JSON响应返回给调用者。
DTS 作为调用者且使用异步调用时,会发送回执参数以及输入参数到AI算法服务端。AI算法服务端需根据回执参数中的结果返回地址和方式将结果发送到对应目标地址。DTS收到分析结果,才能正确解析。结果返回的方式,DTS使用者和AI算法提供者双方可以协商确定。
同步调用时,AI算法可直接返回结果,DTS使用者根据自己需要决定结果返回方式。结果格式不满足BV JSON时,需要编写消息解析脚本进行解析。
-
使用Uvicorn启动API服务器(命令如uvicorn main:app –reload),在本地浏览器或Postman工具中测试:上传图片,调用API,检查返回结果是否正确(例如JSON中包含检测到的对象信息)。
-
使用Docker将应用打包成镜像(编写Dockerfile,包括安装依赖和复制代码),便于移植。构建镜像并测试容器运行。
本示例中采用YOLO11模型为基础搭建AI算法服务器,将结果以事件上报的方式返回到DTS中。
部分代码示例
获取回执参数和回调地址
# 记录请求内容
data = request.get_json()
...
callback_message = data.get('_bv_callback_message', {})
callback_url = data.get('_bv_callback_url', '')
发送结果时携带回调参数
# 如果提供了回调信息
if callback_message:
callback_result['_bv_callback_message'] = callback_message
....
# 如果提供了回调URL,发送POST请求
if callback_url:
try:
requests.post(callback_url, json=callback_result, timeout=5)
logger.info(f"[回调] 发送分析结果到 {callback_url} 成功")
except Exception as e:
logger.error(f"[回调] 发送分析结果到 {callback_url} 失败: {e}")
网关
添加服务地址
- 进入实体管理模块,选择网关。
- 选择“通用网关”,点击编辑进入通用网关的详情页面->选择服务列表 Tab页面。
- 在服务列表 Tab页面下,选择添加一个服务地址。
- 命名为“AICenterServer”,选择类型为Http方式,服务地址输入指向AI算法服务化后的地址。
这个服务地址需指向接收控制服务请求的目标地址。示例选择使用Http方式,还可以根据需要选择WebSocket、Redis等方式。

创建产品-AI产品
- 进入实体管理模块,选择产品。
- 选择创建一个新的产品。
- 命名为“AI产品”,设定接入网关协议选择为HTTP协议,数据格式为透传/自定义。
- 设置完成后点击立即创建。创建成功的产品,会自动创建品类中的产品Profile内容。
加入网关
- 进入实体管理模块,选择网关。
- 选择“通用网关”,点击编辑进入网关的详情页面->选择网关信息 Tab页面。
- 点击编辑,进入网关信息编辑页面。
- 修改关联产品,勾选中“AI产品”产品。将新创建的产品加入到通用网关 下。
通用网关为系统内置的网关,可支持加入HTTP协议接入的实体。
定义产品Profile(事件)
- 进入实体管理模块,选择产品。
- 选择“AI产品”产品,点击编辑进入产品的详情页面->选择产品Profile Tab页面。
- 点击编辑草稿,选择添加功能,选择功能类型为事件 。
- 输入功能名称为”分析结果“,设置标识符为analysis_count,事件类型选择为告警。
- 根据AI算法服务化的算法返回的分析结果设置输出参数。此处示例定义输出参数
objectNumberobjectBoundingBoxboundingBoxImageboundingBoxImageUrltimeheatMapImageUrlheatMapImage。 - 根据需要选择是否需要持久化存储,此处示例勾选需要。
- 点击确定保存事件。
需根据具体使用的AI算法服务化的算法进行相关定义。

定义产品Profile(服务)
- 进入实体管理模块,选择产品。
- 选择“AI产品”产品,点击编辑进入产品的详情页面->选择产品Profile Tab页面。
- 点击编辑草稿,选择添加功能,选择功能类型为服务 。
- 输入功能名称为”图片分析“,设置标识符为picture_analysis_count。
- 服务地址选择已定义的服务地址”AICenterServer“。
- 根据AI算法服务化的算法所需参数 ,服务设置输入参数
analysisTyperesultTypeanalysisResource,输出参数analysisTime。 - 选择结果返回方式为“事件上报”,反控成功后,通过目标事件分析结果的事件返回。
设备反控执行后,根据使用场景选择结果返回方式,是否需要将服务执行结果反馈到DTS中或其他第三方服务中。
在产品Profile所有编辑完成后,需点击发布 。发布后的产品Profile才能被DTS识别。
结果返回方式选择“属性上报”时,需定义相应的属性。可参考定义产品Profile(事件)
编辑消息解析脚本
- 进入实体管理模块,选择产品。
- 选择“AI产品”产品,点击编辑进入产品的详情页面->选择消息解析 Tab页面。
- 选择实体映射(上行)Tab页面,编辑上行解析脚本;
/** * 将第三方服务的自定义数据格式转换为数字孪生服务协议的数据格式,设备、属性上报、事件上报到DTS时调用。 * [@param](https://github.com/param "@param") {object} body - 入参 * [@param](https://github.com/param "@param") {object} body._bv_callback_message - 服务回调参数,回调时存在 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.productId - 产品编号 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.deviceId - 设备编号 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.identifier - 事件的唯一标识符,回调方式为事件上报时存在 * [@param](https://github.com/param "@param") {object} body._bv_callback_message._bv_service_callback - 服务调用参数 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.requestId - 用于标记请求 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.identifier - 服务唯一标识 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.version - 消息格式版本 * [@param](https://github.com/param "@param") {number} body._bv_callback_message._bv_service_callback.timestamp - 服务调用时间 * [@param](https://github.com/param "@param") {number} flag - 入参,0 设备上报 1 属性上报 2 事件上报 * [@returns](https://github.com/returns "@returns") {object} 出参 */ function rawDataToProtocol(body, flag) { var jsonMap = {}; var inputData = body; if (flag === 0) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceType - 设备类型 0:动态资源 1:静态资源,必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,产品属性,可用于更新设备属性值(该数据根据产品Profile中定义的属性进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ return body; } if (flag === 1) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceSecret - 设备签名(不传时将系统匹配),非必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {number} return.timestamp - 上行数据传输时间。时间戳格式为数值,值为自GMT 1970年1月1日0时0分到当前时间点所经过的毫秒数,非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,产品属性,可用于更新设备属性值(该数据根据产品Profile中定义的属性进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ return body; } if (flag === 2) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceSecret - 设备签名(不传时将系统匹配),非必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {string} return.identifier - 事件的唯一标识符,必传 * [@returns](https://github.com/returns "@returns") {number} return.timestamp - 上行数据传输时间。时间戳格式为数值,值为自GMT 1970年1月1日0时0分到当前时间点所经过的毫秒数,非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,事件的输出参数(该数据根据产品Profile中定义的事件输出参数进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ if (inputData._bv_callback_message) { jsonMap['deviceId'] = inputData._bv_callback_message.deviceId; jsonMap['identifier'] = inputData._bv_callback_message.identifier; // 存在文件资源时将存储至网关服务下 if(inputData.heatMapImage){ inputData.heatMapImageUrl = FileStorage.disk(inputData.heatMapImage, "jpg"); } if(inputData.boundingBoxImage){ inputData.boundingBoxImageUrl = FileStorage.disk(inputData.boundingBoxImage, "jpg"); } jsonMap['payload'] = { "analysisTime": inputData.analysisTime, "objectNumber": inputData.objectNumber, "objectBoundingBox": inputData.objectBoundingBox, "boundingBoxImageUrl": inputData.boundingBoxImageUrl, "heatMapImageUrl": inputData.heatMapImageUrl } } return jsonMap; } }
创建虚拟实体
- 进入实体管理模块,选择实体。
- 选择“添加实体“,为AI产品创建一个虚拟实体。
- 在添加实体页面,选择产品为”AI产品“,DeviceID选择自动生成一个DeviceID,输入实体名称为”AI虚拟实体“。
- 输入完成点击确定。DTS创建一个新的实体。

创建产品-BV相机(方式1)
- 进入实体管理模块,选择产品。
- 选择创建一个新的产品。
- 命名为“BV 相机”,从标准品类 中选择产品名称为“BV相机”的产品,设定接入网关协议选择为HTTP协议,数据格式为透传/自定义。
- 设置完成后点击立即创建。创建成功的产品,会自动创建品类中的产品Profile内容。
标准品类中BV相机已经预先定义完成产品Profile,其中包括bv相机的属性、事件以及所能支持的视频操作服务。
创建产品-BV相机(方式2)
- 进入实体管理模块,选择产品。
- 选择创建一个新的产品。
- 命名为“BV 相机”,选择自定义品类,设定接入网关协议选择为HTTP协议,数据格式为透传/自定义。
- 设置完成后点击立即创建。
导入产品Profile
- 进入实体管理模块,选择产品。
- 选择“BV 相机”产品,点击编辑进入产品的详情页面->选择产品Profile Tab页面。
- 点击编辑草稿,选择导入产品Profile功能;
- 下载产品Profile文件保存到本地,并将后缀修改为“.json”。
- 导入文件选择预先存在本地的产品Profile文件,选择导入。
- 提示导入成功,可查看到导入的属性、事件、服务的定义。
产品Profile内容可能会有因为版本的不同存在定义上的差异。

编辑消息解析脚本
- 进入实体管理模块,选择产品。
- 选择“BV 相机”产品,点击编辑进入产品的详情页面->选择消息解析 Tab页面。
- 选择实体映射(上行)Tab页面,编辑上行解析脚本;
/** * 将第三方服务的自定义数据格式转换为数字孪生服务协议的数据格式,设备、属性上报、事件上报到DTS时调用。 * [@param](https://github.com/param "@param") {object} body - 入参 * [@param](https://github.com/param "@param") {object} body._bv_callback_message - 服务回调参数,回调时存在 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.productId - 产品编号 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.deviceId - 设备编号 * [@param](https://github.com/param "@param") {string} body._bv_callback_message.identifier - 事件的唯一标识符,回调方式为事件上报时存在 * [@param](https://github.com/param "@param") {object} body._bv_callback_message._bv_service_callback - 服务调用参数 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.requestId - 用于标记请求 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.identifier - 服务唯一标识 * [@param](https://github.com/param "@param") {string} body._bv_callback_message._bv_service_callback.version - 消息格式版本 * [@param](https://github.com/param "@param") {number} body._bv_callback_message._bv_service_callback.timestamp - 服务调用时间 * [@param](https://github.com/param "@param") {number} flag - 入参,0 设备上报 1 属性上报 2 事件上报 * [@returns](https://github.com/returns "@returns") {object} 出参 */ function rawDataToProtocol(body, flag) { var jsonMap = {}; var inputData = body; if (flag === 0) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceType - 设备类型 0:动态资源 1:静态资源,必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,产品属性,可用于更新设备属性值(该数据根据产品Profile中定义的属性进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ return body; } if (flag === 1) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceSecret - 设备签名(不传时将系统匹配),非必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {number} return.timestamp - 上行数据传输时间。时间戳格式为数值,值为自GMT 1970年1月1日0时0分到当前时间点所经过的毫秒数,非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,产品属性,可用于更新设备属性值(该数据根据产品Profile中定义的属性进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ if (inputData._bv_callback_message) { var bv_callback_message = inputData._bv_callback_message; var bv_service_callback = bv_callback_message._bv_service_callback; var payload = inputData.payload ? inputData.payload : {}; if (bv_service_callback.identifier == "captureServerImage" || bv_service_callback.identifier == "captureServerVideo") { if (inputData.result != 0) { return; } if (bv_service_callback.identifier == "captureServerImage") { var resourceUrl = FileStorage.cache(inputData.file, inputData.suffix, bv_callback_message.deviceId); payload.deviceImage = resourceUrl; } if (bv_service_callback.identifier == "captureServerVideo") { var resourceUrl = FileStorage.disk(inputData.file, inputData.suffix); payload.deviceVideoUrl = resourceUrl; } } if (bv_service_callback.identifier == "video_attractive_score") { payload.score = inputData.score; } if (bv_service_callback.identifier == "captureImage") { //payload.deviceImage = inputData.payload.deviceImage.replace("172.27.240.1","10.132.25.2"); } jsonMap.deviceId = bv_callback_message.deviceId; jsonMap.payload = payload; return jsonMap; } else { return body; } } if (flag === 2) { /** * [@returns](https://github.com/returns "@returns") {string} return.deviceId - 设备编号(保证唯一),必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceSecret - 设备签名(不传时将系统匹配),非必传 * [@returns](https://github.com/returns "@returns") {string} return.deviceName - 设备名称(最大长度255),非必传 * [@returns](https://github.com/returns "@returns") {number} return.deviceStatus - 设备状态 0:禁用 1:启用(默认),非必传 * [@returns](https://github.com/returns "@returns") {string} return.location - 设备位置纬经度(默认0 0 0 = latitude longitude elevation),非必传 * [@returns](https://github.com/returns "@returns") {string} return.identifier - 事件的唯一标识符,必传 * [@returns](https://github.com/returns "@returns") {number} return.timestamp - 上行数据传输时间。时间戳格式为数值,值为自GMT 1970年1月1日0时0分到当前时间点所经过的毫秒数,非必传 * [@returns](https://github.com/returns "@returns") {string} return.version - 版本号,默认default,非必传 * [@returns](https://github.com/returns "@returns") {Object} return.payload - 扩展数据,事件的输出参数(该数据根据产品Profile中定义的事件输出参数进行过滤,已存在定义的数据会保留,其余数据则弃用),非必传 */ if (inputData._bv_callback_message) { jsonMap['deviceId'] = inputData._bv_callback_message.deviceId; jsonMap['identifier'] = inputData._bv_callback_message.identifier; // 存在文件资源时将存储至网关服务下 if(inputData.heatMapImage){ inputData.heatMapImageUrl = FileStorage.disk(inputData.heatMapImage, "jpg"); } if(inputData.boundingBoxImage){ inputData.boundingBoxImageUrl = FileStorage.disk(inputData.boundingBoxImage, "jpg"); } jsonMap['payload'] = { "analysisTime": inputData.analysisTime, "objectNumber": inputData.objectNumber, "objectBoundingBox": inputData.objectBoundingBox, "boundingBoxImageUrl": inputData.boundingBoxImageUrl, "heatMapImageUrl": inputData.heatMapImageUrl } } return jsonMap; } }
加入网关
- 进入实体管理模块,选择网关。
- 选择“BVMedia网关”,点击编辑进入网关的详情页面->选择网关信息 Tab页面。
- 点击编辑,进入网关信息编辑页面。
- 修改关联产品,勾选中BV 相机产品。将新创建的产品加入到BVMedia网关 下。
BVMedia网关为DTS内置视频网关,用于接入来自看门狗服务下的实体,无需开发和创建。BVMedia网关会定时更新看门狗服务中的视频设备到网关关联的产品下。更新之后,BV 相机产品就会有对应的实体。此过程需耐心等待。如想快速加入实体,可使用从网关发现实体 功能。
从网关发现实体
接口“获取相机设备集合”为系统预设置的接口,用于从看门狗服务下请求设备。2025年4月10日之后的DTS版本支持。
- 进入实体管理模块,选择实体。
- 选择功能批量添加,进入批量添加实体页面;
- 选择产品为”BV相机“,使用”从网关发现实体“的添加方式,选择请求接口”获取相机设备集合“。
- 设置完成,选择下一步。
- 系统会请求接口获取接口下的设备实体,若”提示发现X个实体“表示接口请求成功,点击确定即可添加对应的实体到产品下。若”提示发现0个实体“表示接口未能获取到实体,请检查接口。

创建任务
- 进入任务管理模块,选择任务。
- 在任务管理页面,创建一个任务名为“人数统计“的任务。
任务逻辑
每半小时 触发一次分析
执行任务 看门狗下的视频设备截取一张图用来做AI分析
AI分析获取到截图执行分析
- 创建完成后,选择“人数统计“任务点击编辑,进入编辑页面。
- 根据任务逻辑编辑对应的规则。设定触发条件为每小时,添加执行任务类型为”服务调用“,选择BV 相机产品下的某一设备或全部设备作为执行对象,选择服务为”异步抓图服务“,并点击参数配置,设置服务调用需要的输入参数。设置”设备出厂编号“的参数值为当前实体的出厂编号属性值。
异步抓图服务调用参数配置

- 可在下一次任务调用前设置一定时间的延迟,添加一个”延迟“任务。
- 添加执行任务类型为”服务调用“,选择“AI产品”产品下的某一设备或全部设备作为执行对象,选择服务为”图片分析“并点击参数配置,设置服务调用需要的输入参数。
图片分析服务调用参数配置
支持的分析类型以及编码 {0: ‘person’, 1: ‘bicycle’, 2: ‘car’, 3: ‘motorcycle’, 4: ‘airplane’, 5: ‘bus’, 6: ‘train’, 7: ‘truck’, 8: ‘boat’, 9: ‘traffic light’, 10: ‘fire hydrant’, 11: ‘stop sign’, 12: ‘parking meter’}
返回结果字段及其编码 {1:’objectNumber’,2:’objectBoundingBox’,4:’heatMapImage’,8:’boundingBoxImage’,16:’boundingBoxImageUrl’,32:’heatMapImageUrl’}
示例中分析图片来源为执行截图任务的实体的当前截图地址;分析类型为person,返回结果字段为objectNumber、objectBoundingBox、boundingBoxImage

- 配置完成,保存并启动任务。

结果验证
前往任务-执行日志,查看任务的执行情况。

数据监控
经过订阅管理将向DTS订阅设备的属性变化和事件发生的数据,订阅端收到数据之后进行可视化处理便能对设备进行实时数据监控。DTS数据孪生架构提供快速接入查看DTS设备数据的平台功能低代码设计器(无需订阅),详细可视化功能可查看《BigView低代码设计器教程》
附件
最后编辑:北京-产品-yantings 更新时间:2026-07-16 17:09