TDgpt 算法开发指南：在 TDengine 中扩展自定义时序分析与机器学习算法

数据库时序数据库物联网大数据实时分析云原生【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址https://gitcode.com/taosdata/tdengine点击查看免费下载导读本文面向希望在 TDengine 的 AI 分析组件 TDgpt 中扩展自定义算法的开发者。TDgpt 是一个可扩展的高级时间序列数据分析 Agent本文以官方算法开发指南为主体结合 TDgpt 源码 与 TDgpt 目录说明完整讲解算法目录规划、类命名与继承规则、execute方法实现、anode 启动时的半动态加载机制以及如何通过 SQL 语句调用新算法。读完本文你将掌握从零编写并上线一个可被ANOMALY_WINDOW、FORECAST等 SQL 语法直接调用的自定义时序算法。一、TDgpt 的算法扩展机制TDgpt 是 TDengine 生态中独立于taosd服务进程的分析平台安装后以taosanoded服务运行。它的核心设计目标之一是可扩展你可以用 Python 开发任意统计、机器学习或基础模型算法将其加入 TDgpt随后在 SQL 语句中像调用内置算法一样调用它们。这种扩展性来自 anode 的半动态加载机制anode 启动时扫描指定目录将满足要求的 Python 文件注册到平台。整体流程只有三步按照 TDgpt 的约定开发一个分析算法必须使用 Python。将源码文件放入指定目录重启 anode。执行CREATE ANODE语句把 anode 加入你的 TDengine 集群。完成以上步骤后新算法即可通过 SQL 使用。由于 TDgpt 与 TDengine 是解耦部署的在 anode 上新增或升级算法不会影响TDengine 服务端taosd应用侧也只需更新 SQL 语句即可切换到新算法无需改动应用逻辑。这种按需扩展的能力使 TDgpt 可以覆盖非常广泛的使用场景。二、准备开发环境算法开发的第一步是获取 TDgpt 源码。克隆 TDengine 社区版仓库后TDgpt 的源码位于 tools/tdgpt 目录。开发过程中可以重点参考以下路径算法基类定义tools/tdgpt/taosanalytics/base.py服务注册与扫描逻辑tools/tdgpt/taosanalytics/service_registry.py内置异常检测算法tools/tdgpt/taosanalytics/algo/ad/如 ksigma.py、grubbs.py、iqr.py、lof.py内置预测算法tools/tdgpt/taosanalytics/algo/fc/如 holtwinters.py、arima.py、prophet.py、theta.py、ces.py等此外tools/tdgpt/README.md 给出了 taosanode 的安装说明安装后 Python 虚拟环境默认建立在/var/lib/taos/taosanode/venv服务通过systemctl start taosanoded启动REST 服务由 uWSGI 承载默认监听127.0.0.1:6035。三、anode 的目录结构与算法放置位置安装后的 anode 目录结构如下. ├── bin ├── cfg ├── lib │ └── taosanalytics │ ├── algo │ │ ├── ad │ │ └── fc │ ├── misc │ └── test ├── log - /var/log/taos/taosanode ├── model - /var/lib/taos/taosanode/model └── venv - /var/lib/taos/taosanode/venv各目录用途目录说明taosanalytics源码主目录其中algo子目录存放算法实现test子目录存放单元与集成测试misc子目录存放其他文件algo下ad子目录放异常检测算法fc子目录放预测算法venvPython 虚拟环境model各数据集对应的已训练模型cfg配置文件对照当前仓库源码tools/tdgpt/taosanalytics 的algo目录结构比安装目录更加丰富除ad异常检测与fc预测外还有correl相关性分析、imputat数据填充、custom用户自定义算法的推荐放置位置以及dynamic动态模型。其中custom目录同样分ad与fc两个子目录是官方为用户自定义算法预留的位置。算法放置限制异常检测算法的 Python 源码放在./taosanalytics/algo/ad预测算法的 Python 源码放在./taosanalytics/algo/fc。从注册逻辑看这个限制是强制的service_registry.py 的register_all_services()会以必需的requiredTrue方式加载algo/ad、algo/fc、algo/imputat、algo/correl四个目录目录不存在时直接抛出FileNotFoundError快速失败而algo/custom/ad、algo/custom/fc两个目录以可选requiredFalse方式加载目录缺失仅记录日志、不阻断启动。四、算法类的开发规则anode 自动注册算法因此算法文件命名与类定义必须严格遵守以下约定。4.1 文件命名规则算法文件必须以下划线_开头、以Service结尾。例如_KSigmaService是 k-sigma 异常检测算法的类名文件为ksigma.py见 ksigma.py。对应地Holt-Winters 预测算法类名为_HoltWintersService见 holtwinters.py。从扫描逻辑可以印证这一约定service_registry.py 遍历模块中的全部类时会跳过基类名并且跳过不以_开头的类if not class_name.startswith(_)则跳过。因此不以_开头的类不会被注册而以_开头是让注册器识别你自定义算法的关键标记。4.2 类继承规则所有异常检测算法必须继承AbstractAnomalyDetectionService并实现execute方法所有预测算法必须继承AbstractForecastService并实现execute方法。这两类基类都定义在 tools/tdgpt/taosanalytics/base.py 中。基类继承体系如下AnalyticsService最顶层抽象基类定义了execute()抽象方法以及get_desc()、get_params()、get_status()等默认实现内部用READY 0x01、UNAVAILABLE 0x02表示服务状态。AbstractAnalyticsService引入name、desc、status、_builtins等类属性提供set_input_list()与set_params()的默认实现。AbstractAnomalyDetectionServicebase.pytype固定为anomaly-detection自带valid_code 1作为正常值标记set_input_list()会判断输入是一维还是二维列表多维时要求各维度长度一致。AbstractForecastServicebase.pytype固定为forecastset_params()强制要求传入start_ts、time_step、rows三个参数并支持可选的period周期、conf置信度默认 0.95、return_conf、precision、tz等参数。除了这两类base.py 中还定义了AbstractImputationService填充、AbstractCorrelationService相关性、AbstractRegressionService回归、AbstractClassificationService分类等基类对应仓库中algo/imputat、algo/correl、algo/dynamic等目录下的算法类型说明 TDgpt 的扩展面已经超出文档所述的异常检测与预测两类。4.3 类属性初始化算法类必须初始化以下两个属性name算法的标识符只允许小写字母。该标识符会在你使用SHOW语句查看可用算法时显示同时也是 SQL 语句中algo参数对应的取值。desc算法的基本描述。例如内置 k-sigma 算法ksigma.py的写法class _KSigmaService(AbstractAnomalyDetectionService): KSigma algorithm is to check the anomaly data in the input list name ksigma desc the k-sigma algorithm (or 3σ rule) expresses a conventional heuristic that nearly all values are taken to lie within k (usually three) standard deviations of the mean, and thus it is empirically useful to treat 99.7% probability as near certainty _builtins TrueSQL 侧的对应关系如下--- 这里的 algo 取值就是类中定义的 name 值 SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algoname)五、动手实现一个自定义算法下面以两个完整示例演示如何编写自定义算法。5.1 自定义异常检测算法将下面的文件放入taosanalytics/algo/ad/目录例如_MySigmaService所在文件。一个合格的异常检测算法需要继承AbstractAnomalyDetectionService、初始化name与desc、实现execute返回逐点标记列表。参考内置 k-sigma 的实现ksigma.py它通过set_params接收k参数取值范围[1, 3]execute中计算均值与标准差得到阈值区间落在区间外的点标记为-1异常区间内标记为1正常即valid_codeimport numpy as np from taosanalytics.base import AbstractAnomalyDetectionService class _MySigmaService(AbstractAnomalyDetectionService): customized k-sigma anomaly detection algorithm name mysigma # 小写字母标识SQL 中 algomysigma 即调用本算法 desc custom anomaly detection based on sigma range _builtins False # 自定义算法标记为非内置 def __init__(self, k_val3): super().__init__() self.k_val k_val def set_params(self, params): super().set_params(params) if k in params: k int(params[k]) if k 1 or k 3: raise ValueError(k value out of range, valid range [1, 3]) self.k_val k def get_params(self): return {k: self.k_val} def execute(self): if self.input_is_empty(): return [] avg np.mean(self.list) std np.std(self.list) upper avg self.k_val * std lower avg - self.k_val * std # 异常点返回 -1不等于 valid_code正常点返回 1 return [-1 if x lower or x upper else 1 for x in self.list]注意set_params内部对参数做校验当参数非法时抛出ValueError注册器与调用链路会将异常信息返回给请求方。输入为空时execute应返回空列表input_is_empty()是基类提供的辅助方法。5.2 自定义预测算法将下面的文件放入taosanalytics/algo/fc/目录。预测算法需继承AbstractForecastService。基类的set_params已保证start_ts、time_step、rows等字段就绪你只需关注execute中的模型计算与结果格式。参考内置 Holt-Winters 实现holtwinters.pyexecute校验输入长度与预测行数后调用平滑模型得到预测序列再通过insert_ts_list在结果最前面插入时间戳列表最终返回{mse: ..., res: [ts_list, forecast, lower, upper]}from taosanalytics.base import AbstractForecastService from taosanalytics.algo.forecast import insert_ts_list class _MySmaService(AbstractForecastService): customized simple moving average forecast algorithm name mysma desc forecast by simple moving average _builtins False def __init__(self): super().__init__() self.window 5 def set_params(self, params): super().set_params(params) if window in params: self.window int(params[window]) def execute(self): if self.list is None or len(self.list) self.window: raise ValueError(number of input data is less than the window size) if self.rows 0: raise ValueError(fc rows is not specified yet) # 以最后 window 个点的均值作为后续每个预测点的基础值 base sum(self.list[-self.window:]) / self.window forecast [base] * self.rows # 返回格式第一项必须是时间戳列表 res [forecast] insert_ts_list(res, self.start_ts, self.time_step, self.rows) return {mse: 0.0, res: res}预测结果必须满足 forecast.py 中check_forecast_results的约束res第一项为时间戳列表后续各项预测值、可选的下置信界、上置信界长度必须与时间戳数量一致若开启置信区间return_conf结果中还应包含上下界两行。5.3 动态模型与配置文件扩展进阶除手写 Python 文件外TDgpt 还支持通过 JSON 配置文件注册动态模型。service_registry.py 的register_service_from_file()会解析包含algo字段的 JSON将模型按类型包装为DynamicForecastService、DynamicAnomalyService、DynamicRegressionService或DynamicClassificationServicesync_dynamic_services()会在每次获取服务时同步动态模型目录配置文件被删除时自动将对应模型从内存中移除。动态模型目录由配置项dynamic_model_dir指定Linux 默认/usr/local/taos/taosanode/model/dynamic/。这类能力更接近模型托管而非算法源码扩展本文不再展开。六、半动态加载原理anode 如何发现你的算法理解加载原理有助于排查为什么算法没生效。核心代码在 service_registry.py 的_register_services_in_dir()加载流程如下目录扫描列出算法目录下所有文件跳过__init__.py、__pycache__以及非.py文件子目录不会递归加载。模块导入以lib_prefix 文件名构造模块名通过importlib.import_module导入导入失败仅记录错误日志并跳过不影响其他算法。类筛选遍历模块中所有类inspect.getmembers跳过基类名和不以_开头的类类必须定义在当前模块内algo_cls.__module__ module.__name__否则视为从别的模块导入而跳过。实例化注册对通过筛选的类执行algo_cls()实例化并以类属性name作为 key 注册进服务字典实例化或注册失败同样只记日志、跳过。去重保护_register_service()发现同名服务已存在时抛出RuntimeError并拒绝注册。注册完成后app.py 中的_init_app()会在服务启动时调用loader.register_all_services()对外暴露的 REST 接口包括/list列出全部算法、/anomaly-detect异常检测、/forecast预测、/imputation、/correlation、/regression等见 app.py。TDengine 服务端正是通过这些接口与 anode 通信因此算法增减只影响 anode不影响taosd。从代码还可以推断一个兼容性细节注册器对 Python 3.12 环境会跳过_SHESDService因 pandas 兼容问题见 service_registry.py说明自定义算法也应关注所依赖第三方库与运行环境 Python 版本的兼容性。七、将算法上线并在 SQL 中调用完成开发与本地验证后按以下步骤上线把算法 Python 文件复制到 anode 安装目录下对应的taosanalytics/algo/ad异常检测或taosanalytics/algo/fc预测目录。重启 anode 服务systemctl restart taosanoded安装与启停说明见 tools/tdgpt/README.md。启动日志位于/var/log/taos/taosanode可在其中看到load algorithm:xxx、register dynamic model:xxx等注册信息。在 TDengine 集群中执行CREATE ANODE语句将 anode 注册进集群anode 管理相关操作参见 anode 管理文档。使用SHOW类语句查看可用算法列表确认name已出现。在 SQL 中通过algoname方式调用例如-- 调用自定义异常检测算法 mysigmak 参数取 2 SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algomysigma,k2); -- 调用自定义预测算法 mysma预测 10 行窗口 7 SELECT _wstart, _wend, forecast FROM foo FORECAST(col_name, algomysma,window7,rows10);八、开发要点与常见问题小结命名是硬约束文件内类名必须_开头、Service结尾不以_开头的类会被注册器直接忽略。name必须小写它是 SQL 侧的算法标识大小写不匹配将导致算法未找到。务必实现execute基类中的execute是抽象方法execute的返回值格式异常检测的逐点标记、预测的[ts, forecast, ...]结构必须与框架约定一致否则结果校验会失败。参数校验放在set_params非法参数应抛出ValueError框架会捕获并返回错误信息避免算法运行到一半才报错。目录不可缺失algo/ad、algo/fc、algo/imputat、algo/correl四个内置目录必须存在缺失会导致 anode 启动失败自定义算法建议放在algo/custom/ad、algo/custom/fc可选目录缺失不影响启动。解耦部署算法更新只需重启 anode无需重启taosd应用侧改 SQL 即可切换算法。以上约定均可对照仓库源码逐一验证基类与参数约束见 base.py扫描注册逻辑见 service_registry.py内置算法示例见 ad 目录 与 fc 目录。按照本文的步骤与规则你即可为 TDgpt 交付可被 SQL 直接调用的自定义时序分析能力。赞分享数据库时序数据库物联网大数据实时分析云原生【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址https://gitcode.com/taosdata/tdengine点击查看免费下载相关推荐TDgpt 算法扩展开发指南为 TDengine 时序分析平台编写自定义分析与预测算法TDgpt 算法扩展开发指南为 TDengine 时序分析平台编写自定义分析与预测算法 TDgpt 是 TDengine 生态中可扩展的高级时序数据分析 Ag数据库时序数据库大数据物联网云原生TDgpt 自定义算法开发指南为 TDengine 扩展时序预测与异常检测模型TDgpt 自定义算法开发指南为 TDengine 扩展时序预测与异常检测模型 导读 TDgpt 是 TDengine 内置的时序数据智能分析组件即 tao数据库时序数据库大数据物联网云原生TDgpt 分析算法开发指南为 TDengine 添加自定义时序预测与异常检测算法TDgpt 分析算法开发指南为 TDengine 添加自定义时序预测与异常检测算法 TDgpt 是 TDengine 内置的时序数据高级分析智能体Anode数据库时序数据库大数据物联网云原生上一篇Avalonia Canvas4 种图形 14 行 XAML 画出跨平台速度仪表盘下一篇【特别发布】 Ant Design Blazor 1.1.2版本发布增强覆盖层与表格功能创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考