e9e029deaf83344acaf22329c3b804ccbb58367e
Worker 插件开发指南
每个 src/plugins/<name>/ 子目录是一个独立的 Worker 插件,封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务,HTTP 路由由框架根据 TOML 配置自动生成。
目录结构约定
<plugin_name>/
├── __init__.py # 入口:create_workers() 工厂 + cleanup() + auto_bind_models()
├── global_var.py # 模块级全局状态(Database、API client 等)
├── router/ # RPC 端点实现
│ ├── __init__.py # 导入所有端点模块(`from . import *` 或显式导入)
│ └── <操作名>.py # 每个端点一个文件,使用 @rpc_method 装饰器
├── models/ # 数据模型
│ ├── __init__.py # 模型导出(由 gen_models 自动生成,勿手动修改)
│ ├── pydantic.py # Pydantic 请求/响应模型(由 gen_models 自动生成)
│ └── <表名>.py # SQLModel ORM 模型(如需要数据库)
├── routes.py # 自定义 FastAPI 路由(可选,用于特殊端点)
└── README.md # 插件说明文档(可选)
已有插件概览
| 插件 | 特点 | 适合参考 |
|---|---|---|
| example | 最简示例,auto_bind_models + Database + cleanup |
入门学习 |
| auth | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 |
| doubao | 外部 API 客户端(Ark/OpenAI)+ AICC 安全通道 | 外部集成 |
| aibox_sophon_adapter | 自定义 APIRouter + MQTT 转发 + 后台任务 |
高级模式 |
开发新插件:完整流程
第一步:创建目录结构
mkdir -p src/plugins/<name>/router
mkdir -p src/plugins/<name>/models
touch src/plugins/<name>/__init__.py
touch src/plugins/<name>/global_var.py
touch src/plugins/<name>/router/__init__.py
第二步:注册插件
在 config/workers/config.toml 中添加:
[[workers]]
type = "<name>"
module = "plugins.<name>"
factory = "create_workers"
cleanup = "cleanup"
count = 3
[workers.config]
type = "<name>"
data_dir = "./data"
第三步:编写端点 TOML 配置
创建 config/workers/<name>/config.toml(分组配置):
[worker]
defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = false }
[worker.registration]
router_prefix = "/<prefix>"
tags = ["<name>"]
创建每个操作的 .toml 文件,如 config/workers/<name>/my_action.toml:
[POST]
summary = "我的操作"
auth_required = true
[POST.response_model]
result = { type = "str", required = true }
message = { type = "str", default = "" }
第四步:生成 Pydantic 模型
uv run python -m chongming.scripts.gen_models
自动在 src/plugins/<name>/models/ 下生成 pydantic.py 和 __init__.py。
第五步:实现 RPC 端点
src/plugins/<name>/router/my_action.py:
from chongming_worker.decorators import rpc_method
from ..models.pydantic import MyActionPostRequest, MyActionPostResponse
@rpc_method(
subject="<name>.my_action",
method="POST",
response_model=MyActionPostResponse,
)
async def my_action(request: MyActionPostRequest) -> MyActionPostResponse:
"""业务逻辑"""
result = await do_something(request)
return MyActionPostResponse(result="success", message="done")
第六步:更新 router 导出
src/plugins/<name>/router/__init__.py:
from . import my_action
__all__ = ["my_action"]
第七步:实现工厂函数
src/plugins/<name>/__init__.py:
from pathlib import Path
from chongming_worker import auto_bind_models
from chongming_worker.runtime import create_workers as _create_workers
from .router import * # 触发 @rpc_method 注册
auto_bind_models("<name>") # 自动绑定生成的 Pydantic 模型
async def create_workers(session, config, count, worker_manager):
"""插件入口:初始化资源,启动 Worker"""
from chongming_database import Database
from .global_var import set_db
db_path = Path(config.get("data_dir", "./data")) / "<name>.db"
db = Database(db_path)
await db.initialize()
set_db(db)
return _create_workers(session, config, count, worker_manager)
async def cleanup(worker_manager):
"""清理资源"""
from .global_var import db
if db:
await db.dispose()
注意:
create_workers必须是async函数,因为Database.initialize()是异步的。框架在asyncio.gather中并发调用所有插件的工厂函数。
自定义 FastAPI 路由(高级)
当端点无法用 TOML 描述时(如文件上传),插件可暴露自定义 APIRouter。
src/plugins/<name>/routes.py:
from fastapi import APIRouter, UploadFile, File, Form
router = APIRouter(prefix="/<prefix>", tags=["<name>"])
@router.post("/upload")
async def upload(
file: UploadFile = File(...),
description: str = Form(...),
):
# 处理文件上传
return {"status": "ok"}
在 __init__.py 中暴露 router:
from .routes import router # 网关通过 getattr 自动发现
网关在 load_workers() 阶段自动检测并注册该 router,无需修改网关代码。
为现有插件添加新端点
- 创建
config/workers/<name>/<new_op>.toml - 运行
uv run python -m chongming.scripts.gen_models - 创建
router/<new_op>.py,引用生成的模型 - 更新
router/__init__.py - 重启服务
全局变量模式
每个插件通过 global_var.py 管理模块级共享资源:
# global_var.py
from chongming_database import Database
db: Database | None = None
def set_db(database: Database):
global db
db = database
在端点中直接引用:
from ..global_var import db
@rpc_method(subject="<name>.action", method="GET", ...)
async def handler(request):
async with db.get_session() as session:
...
@rpc_method 装饰器
@rpc_method(
subject="<plugin>.<operation>", # Zenoh 主题(必需)
method="GET", # HTTP 方法:GET / POST / MQTT
response_model=ResponseCls, # Pydantic 响应模型类
)
async def handler(request: RequestCls) -> ResponseCls:
...
| method | 参数来源 | 请求绑定 |
|---|---|---|
GET |
URL query string | TOML params 定义 |
POST |
JSON body | 自动绑定到 Pydantic 请求模型 |
MQTT |
MQTT payload JSON | 同 POST |
- 设置
internal = true后,端点不注册 HTTP 路由,仅可通过 Zenoh RPC 或 MQTT 内部调用 - MQTT 方法返回
None时,Bridge 自动跳过响应发布
命名约定
| 概念 | 约定 | 示例 |
|---|---|---|
| 插件名 | 小写下划线 | aibox_sophon_adapter |
| 操作名 | 小写下划线 | board_ping, list_devices |
| Zenoh subject | <插件>.<操作> |
example.add |
| TOML 文件名 | 与操作名一致 | add.toml |
| Python 文件名 | 与操作名一致 | add.py |
| 模型类名 | <操作><方法>Request/Response |
AddGetRequest |
| 概念 | 约定 | 示例 |
|---|---|---|
| 插件名 | 小写下划线 | doubao, auth, example |
| 操作名 | 小写下划线 | analyze, add, change_password |
| Zenoh subject | <插件>.<操作> |
example.add, doubao.analyze |
| TOML 文件名 | 与操作名一致 | analyze.toml, add.toml |
| Python 路由文件名 | 与操作名一致 | analyze.py, add.py |
现有插件
| 插件 | 说明 | 文档 |
|---|---|---|
| auth | 用户注册、登录、JWT 鉴权、用户信息管理 | 含 SQLite 数据库 |
| doubao | 火山引擎豆包视觉理解模型集成 | 第三方 API 调用 |
| example | 四则运算示例 + 访问计数器 | 入门参考 |
Languages
Python
100%