PandaAI QuantFlow 插件系统架构解析从节点注册到工作流执行的完整链路【免费下载链接】panda_quantflow项目地址: https://gitcode.com/gh_mirrors/pa/panda_quantflowPandaAI QuantFlow 是一个面向量化研究场景的可视化工作流平台而插件系统正是它最核心的骨架——用户拖拽画布上的一个个节点本质上就是在调用一套精心设计的插件架构。本文将以节点注册 → 插件加载 → 工作流持久化 → 分层调度执行为主线用通俗的语言为你拆解工作流执行的完整链路帮助新手快速理解这套插件系统架构的设计精髓也为想开发自定义插件的开发者指明入口。为什么插件系统是 QuantFlow 的灵魂在传统的量化平台里每个功能往往写成死代码想加一个因子、换一个模型都要改源码重新部署。QuantFlow 的做法完全不同一切皆节点。读取数据是节点计算因子是节点训练模型是节点回测是节点……每个节点都是一个独立的插件通过画布连线自由组合成工作流。这种设计的三大优势非常直观优势说明 即插即用插件写好放入指定目录重启即被自动发现无需改动主程序 灵活编排节点可任意连线形成串行、并行、分支等复杂量化流程 专注业务开发者只需关心节点自身的输入输出调度与日志由引擎代劳第一环节点注册——一个装饰器搞定一切工作流里的每个节点其身份信息都由work_node()装饰器登记。它位于 work_node_registery.py核心逻辑只有两件事校验类型确保被装饰的类继承了BaseWorkNode不符合直接报错从源头杜绝非法插件。写入注册表把节点名、显示名、分组、类型、配色等元信息作为类属性挂到节点类上并登记进全局字典ALL_WORK_NODES。比如内置的Python 代码输入节点只需要几行声明就能完成注册work_node(namePython代码输入, group01-基础工具, typecode, box_colorgreen) class CodeControl(BaseWorkNode): ...这里的group还支持用/分割自动生成多层目录结构方便前端把几十个节点分类展示。前端表单怎么来的秘密在ui()装饰器节点有输入参数前端就要渲染表单。QuantFlow 的方案非常巧妙用 ui_control.py 中的ui()装饰器给 Pydantic 输入模型打补丁把 UI 偏好输入框类型、行数、占位符等直接注入到 JSON Schema 中。前端读 Schema 就能自动生成对应的表单控件插件作者完全不用写一行前端代码。第二环BaseWorkNode——插件的统一契约所有插件都必须继承 base_work_node.py 中的BaseWorkNode抽象基类并实现三个关键方法方法作用input_model()声明节点接收什么输入Pydantic 模型output_model()声明节点输出什么结果Pydantic 模型run()节点真正的业务逻辑接收输入模型、返回输出模型输入输出都用 Pydantic 模型定义意味着引擎可以在执行前做字段级校验字段缺失、类型不对都能提前拦截。内置的贴心日志系统插件作者在run()里调用self.log_info(开始处理数据)即可记录日志。这个日志系统做了两层设计节点内部先把日志放进内存队列执行结束后再由引擎统一异步落库到 MongoDB。这样既避免了在同步代码里调用异步方法的尴尬也保证日志与节点执行状态能一一对应。第三环动态加载——插件即插即用的秘密注册表有了插件文件什么时候加载答案在 work_node_loader.py 的load_all_nodes()中它在服务启动时执行一次内部插件目录panda_plugins/internal/存放官方内置节点因子构建、LightGBM、XGBoost、LSTM、特征工程等 40 个节点自定义插件目录panda_plugins/custom/留给用户自己开发的节点加载器会递归遍历这两个目录用 Python 的importlib动态导入每个.py模块执行文件时work_node()装饰器便会自动把节点注册进ALL_WORK_NODES。单个模块加载失败不影响其他插件容错性很强。更值得关注的是代码中还预留了load_work_node_from_db()函数——支持把用户自定义节点的 Python 源码存进数据库运行时按对象 ID 动态编译加载。这意味着未来用户可以在网页上直接编写并发布插件实现真正的云端即写即用。第四环工作流保存——节点与连线如何持久化用户在画布上搭好的流程图会通过 workflow_save_logic.py 保存到 MongoDB 的workflow集合中。数据模型由两个核心类承载WorkNodeModelwork_node_model.py记录节点类型、画布坐标、宽高以及两类关键数据——static_input_data用户手动填写的静态参数和output_db_id节点运行结果的数据库引用LinkModellink_model.py记录一条连线从哪个节点的哪个输出字段流向哪个节点的哪个输入字段还有运行状态禁用/启用/运行中/成功/失败有意思的是这套模型正在逐步去 Litegraph 化早期版本依赖 Litegraph 库保存画布数据现在节点数据已独立建模静态输入和运行结果也能完整落库为后续功能演进铺路。第五环工作流执行引擎——从入队到分层调度这是整条链路的高潮部分。点击运行按钮后依次发生以下事情① 运行入口鉴权与入队workflow_run_logic.py 负责接收运行请求先做权限校验防止调用他人工作流再用 MongoDB事务同时创建workflow_run运行记录并更新工作流的last_run_id保证数据一致性。随后根据运行模式分发任务CLOUD 模式把任务 JSON 发布到 RabbitMQ 消息队列由独立的工作进程消费执行LOCAL 模式通过 FastAPI 的BackgroundTasks直接在本地后台线程执行② 拓扑排序确定执行顺序真正干活的是 run_workflow_utils.py 中的run_workflow_in_background()。引擎拿到工作流定义后第一件事是调用determine_workflow_execution_order()做拓扑排序。算法思路很经典统计每个节点的入度依赖的前置节点数量入度为 0 的节点构成第一层执行完一层后把后继节点的入度减一又入度为 0 的节点组成下一层……依此类推最终得到分层执行序列例如第1层: [读取CSV, 读取行情] ← 无依赖可并行 第2层: [因子计算] ← 依赖第1层 第3层: [LightGBM训练, 回测] ← 依赖第2层可并行如果发现存在循环依赖或无法到达的节点引擎会直接报错拒绝执行——这等于在运行前就帮用户排查掉了流程图中的死锁。③ 分层执行线程池并行 输入注入引擎按层执行节点每一层的节点互不依赖通过run_in_threadpool放进线程池并行运行。每个节点的执行过程是从ALL_WORK_NODES注册表取出节点类若节点名带:前缀则走数据库动态加载路径实例化注入静态输入用户在画布填的参数 动态输入从前置节点输出结果中按连线字段映射取值调用节点的run()方法执行业务逻辑把输出结果保存到 GridFSMongoDB 的文件存储返回output_db_id更新运行状态、成功节点列表、已通过连线列表④ 全流程状态机与友好报错整个运行过程的状态变化清晰可见PENDING排队中→RUNNING运行中带百分比进度→SUCCESS成功或FAILED失败也支持MANUAL_STOP手动终止。特别值得一提的是引擎的友好报错机制当节点执行失败时generate_friendly_error_message()会分析异常类型比如发现缺少df_factor字段会提示该字段通常由公式节点或因子构建节点输出并给出连接修复建议和调试信息。新手面对报错不再一头雾水。快速上手开发你的第一个自定义节点理解了整条链路开发插件就水到渠成。参考panda_plugins/custom/examples/下的示例只需三步在custom目录新建一个.py文件继承BaseWorkNode实现input_model()、output_model()并用ui()美化输入表单实现run()写入核心逻辑用work_node()声明节点名称与分组重启服务后你的节点就会自动出现在画布的节点面板里成为工作流的一等公民。总结从work_node()装饰器的轻量注册到BaseWorkNode的统一契约再到动态加载、持久化建模与分层调度执行PandaAI QuantFlow 的插件系统架构形成了一条清晰完整的链路。它的设计哲学值得借鉴用最小的约定换最大的自由——插件作者只需关注输入是什么、输出是什么、逻辑怎么做其余的事务、调度、日志、错误处理全部交给引擎。这套架构让量化研究从改代码进化到搭积木无论是新手学习还是专业研究都能高效地把想法变成可运行的工作流。【免费下载链接】panda_quantflow项目地址: https://gitcode.com/gh_mirrors/pa/panda_quantflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
PandaAI QuantFlow 插件系统架构解析：从节点注册到工作流执行的完整链路
PandaAI QuantFlow 插件系统架构解析从节点注册到工作流执行的完整链路【免费下载链接】panda_quantflow项目地址: https://gitcode.com/gh_mirrors/pa/panda_quantflowPandaAI QuantFlow 是一个面向量化研究场景的可视化工作流平台而插件系统正是它最核心的骨架——用户拖拽画布上的一个个节点本质上就是在调用一套精心设计的插件架构。本文将以节点注册 → 插件加载 → 工作流持久化 → 分层调度执行为主线用通俗的语言为你拆解工作流执行的完整链路帮助新手快速理解这套插件系统架构的设计精髓也为想开发自定义插件的开发者指明入口。为什么插件系统是 QuantFlow 的灵魂在传统的量化平台里每个功能往往写成死代码想加一个因子、换一个模型都要改源码重新部署。QuantFlow 的做法完全不同一切皆节点。读取数据是节点计算因子是节点训练模型是节点回测是节点……每个节点都是一个独立的插件通过画布连线自由组合成工作流。这种设计的三大优势非常直观优势说明 即插即用插件写好放入指定目录重启即被自动发现无需改动主程序 灵活编排节点可任意连线形成串行、并行、分支等复杂量化流程 专注业务开发者只需关心节点自身的输入输出调度与日志由引擎代劳第一环节点注册——一个装饰器搞定一切工作流里的每个节点其身份信息都由work_node()装饰器登记。它位于 work_node_registery.py核心逻辑只有两件事校验类型确保被装饰的类继承了BaseWorkNode不符合直接报错从源头杜绝非法插件。写入注册表把节点名、显示名、分组、类型、配色等元信息作为类属性挂到节点类上并登记进全局字典ALL_WORK_NODES。比如内置的Python 代码输入节点只需要几行声明就能完成注册work_node(namePython代码输入, group01-基础工具, typecode, box_colorgreen) class CodeControl(BaseWorkNode): ...这里的group还支持用/分割自动生成多层目录结构方便前端把几十个节点分类展示。前端表单怎么来的秘密在ui()装饰器节点有输入参数前端就要渲染表单。QuantFlow 的方案非常巧妙用 ui_control.py 中的ui()装饰器给 Pydantic 输入模型打补丁把 UI 偏好输入框类型、行数、占位符等直接注入到 JSON Schema 中。前端读 Schema 就能自动生成对应的表单控件插件作者完全不用写一行前端代码。第二环BaseWorkNode——插件的统一契约所有插件都必须继承 base_work_node.py 中的BaseWorkNode抽象基类并实现三个关键方法方法作用input_model()声明节点接收什么输入Pydantic 模型output_model()声明节点输出什么结果Pydantic 模型run()节点真正的业务逻辑接收输入模型、返回输出模型输入输出都用 Pydantic 模型定义意味着引擎可以在执行前做字段级校验字段缺失、类型不对都能提前拦截。内置的贴心日志系统插件作者在run()里调用self.log_info(开始处理数据)即可记录日志。这个日志系统做了两层设计节点内部先把日志放进内存队列执行结束后再由引擎统一异步落库到 MongoDB。这样既避免了在同步代码里调用异步方法的尴尬也保证日志与节点执行状态能一一对应。第三环动态加载——插件即插即用的秘密注册表有了插件文件什么时候加载答案在 work_node_loader.py 的load_all_nodes()中它在服务启动时执行一次内部插件目录panda_plugins/internal/存放官方内置节点因子构建、LightGBM、XGBoost、LSTM、特征工程等 40 个节点自定义插件目录panda_plugins/custom/留给用户自己开发的节点加载器会递归遍历这两个目录用 Python 的importlib动态导入每个.py模块执行文件时work_node()装饰器便会自动把节点注册进ALL_WORK_NODES。单个模块加载失败不影响其他插件容错性很强。更值得关注的是代码中还预留了load_work_node_from_db()函数——支持把用户自定义节点的 Python 源码存进数据库运行时按对象 ID 动态编译加载。这意味着未来用户可以在网页上直接编写并发布插件实现真正的云端即写即用。第四环工作流保存——节点与连线如何持久化用户在画布上搭好的流程图会通过 workflow_save_logic.py 保存到 MongoDB 的workflow集合中。数据模型由两个核心类承载WorkNodeModelwork_node_model.py记录节点类型、画布坐标、宽高以及两类关键数据——static_input_data用户手动填写的静态参数和output_db_id节点运行结果的数据库引用LinkModellink_model.py记录一条连线从哪个节点的哪个输出字段流向哪个节点的哪个输入字段还有运行状态禁用/启用/运行中/成功/失败有意思的是这套模型正在逐步去 Litegraph 化早期版本依赖 Litegraph 库保存画布数据现在节点数据已独立建模静态输入和运行结果也能完整落库为后续功能演进铺路。第五环工作流执行引擎——从入队到分层调度这是整条链路的高潮部分。点击运行按钮后依次发生以下事情① 运行入口鉴权与入队workflow_run_logic.py 负责接收运行请求先做权限校验防止调用他人工作流再用 MongoDB事务同时创建workflow_run运行记录并更新工作流的last_run_id保证数据一致性。随后根据运行模式分发任务CLOUD 模式把任务 JSON 发布到 RabbitMQ 消息队列由独立的工作进程消费执行LOCAL 模式通过 FastAPI 的BackgroundTasks直接在本地后台线程执行② 拓扑排序确定执行顺序真正干活的是 run_workflow_utils.py 中的run_workflow_in_background()。引擎拿到工作流定义后第一件事是调用determine_workflow_execution_order()做拓扑排序。算法思路很经典统计每个节点的入度依赖的前置节点数量入度为 0 的节点构成第一层执行完一层后把后继节点的入度减一又入度为 0 的节点组成下一层……依此类推最终得到分层执行序列例如第1层: [读取CSV, 读取行情] ← 无依赖可并行 第2层: [因子计算] ← 依赖第1层 第3层: [LightGBM训练, 回测] ← 依赖第2层可并行如果发现存在循环依赖或无法到达的节点引擎会直接报错拒绝执行——这等于在运行前就帮用户排查掉了流程图中的死锁。③ 分层执行线程池并行 输入注入引擎按层执行节点每一层的节点互不依赖通过run_in_threadpool放进线程池并行运行。每个节点的执行过程是从ALL_WORK_NODES注册表取出节点类若节点名带:前缀则走数据库动态加载路径实例化注入静态输入用户在画布填的参数 动态输入从前置节点输出结果中按连线字段映射取值调用节点的run()方法执行业务逻辑把输出结果保存到 GridFSMongoDB 的文件存储返回output_db_id更新运行状态、成功节点列表、已通过连线列表④ 全流程状态机与友好报错整个运行过程的状态变化清晰可见PENDING排队中→RUNNING运行中带百分比进度→SUCCESS成功或FAILED失败也支持MANUAL_STOP手动终止。特别值得一提的是引擎的友好报错机制当节点执行失败时generate_friendly_error_message()会分析异常类型比如发现缺少df_factor字段会提示该字段通常由公式节点或因子构建节点输出并给出连接修复建议和调试信息。新手面对报错不再一头雾水。快速上手开发你的第一个自定义节点理解了整条链路开发插件就水到渠成。参考panda_plugins/custom/examples/下的示例只需三步在custom目录新建一个.py文件继承BaseWorkNode实现input_model()、output_model()并用ui()美化输入表单实现run()写入核心逻辑用work_node()声明节点名称与分组重启服务后你的节点就会自动出现在画布的节点面板里成为工作流的一等公民。总结从work_node()装饰器的轻量注册到BaseWorkNode的统一契约再到动态加载、持久化建模与分层调度执行PandaAI QuantFlow 的插件系统架构形成了一条清晰完整的链路。它的设计哲学值得借鉴用最小的约定换最大的自由——插件作者只需关注输入是什么、输出是什么、逻辑怎么做其余的事务、调度、日志、错误处理全部交给引擎。这套架构让量化研究从改代码进化到搭积木无论是新手学习还是专业研究都能高效地把想法变成可运行的工作流。【免费下载链接】panda_quantflow项目地址: https://gitcode.com/gh_mirrors/pa/panda_quantflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
