# Worker 插件开发指南 每个 `src/plugins//` 子目录是一个独立的 Worker 插件,封装特定领域的业务逻辑。插件通过 Zenoh RPC 对外提供服务,HTTP 路由由框架根据 TOML 配置自动生成。 --- ## 目录结构约定 ```text / ├── __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 转发 + 后台任务 | 高级模式 | --- ## 开发新插件:完整流程 ### 第一步:创建目录结构 ```bash mkdir -p src/plugins//router mkdir -p src/plugins//models touch src/plugins//__init__.py touch src/plugins//global_var.py touch src/plugins//router/__init__.py ``` ### 第二步:注册插件 在 `config/workers/config.toml` 中添加: ```toml [[workers]] type = "" module = "plugins." factory = "create_workers" cleanup = "cleanup" count = 3 [workers.config] type = "" data_dir = "./data" ``` ### 第三步:编写端点 TOML 配置 创建 `config/workers//config.toml`(分组配置): ```toml [worker] defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = false } [worker.registration] router_prefix = "/" tags = [""] ``` 创建每个操作的 `.toml` 文件,如 `config/workers//my_action.toml`: ```toml [POST] summary = "我的操作" auth_required = true [POST.response_model] result = { type = "str", required = true } message = { type = "str", default = "" } ``` ### 第四步:生成 Pydantic 模型 ```bash uv run python -m chongming.scripts.gen_models ``` 自动在 `src/plugins//models/` 下生成 `pydantic.py` 和 `__init__.py`。 ### 第五步:实现 RPC 端点 `src/plugins//router/my_action.py`: ```python from chongming_worker.decorators import rpc_method from ..models.pydantic import MyActionPostRequest, MyActionPostResponse @rpc_method( subject=".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//router/__init__.py`: ```python from . import my_action __all__ = ["my_action"] ``` ### 第七步:实现工厂函数 `src/plugins//__init__.py`: ```python 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("") # 自动绑定生成的 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")) / ".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//routes.py`: ```python from fastapi import APIRouter, UploadFile, File, Form router = APIRouter(prefix="/", tags=[""]) @router.post("/upload") async def upload( file: UploadFile = File(...), description: str = Form(...), ): # 处理文件上传 return {"status": "ok"} ``` 在 `__init__.py` 中暴露 `router`: ```python from .routes import router # 网关通过 getattr 自动发现 ``` 网关在 `load_workers()` 阶段自动检测并注册该 router,无需修改网关代码。 --- ## 为现有插件添加新端点 1. 创建 `config/workers//.toml` 2. 运行 `uv run python -m chongming.scripts.gen_models` 3. 创建 `router/.py`,引用生成的模型 4. 更新 `router/__init__.py` 5. 重启服务 --- ## 全局变量模式 每个插件通过 `global_var.py` 管理模块级共享资源: ```python # global_var.py from chongming_database import Database db: Database | None = None def set_db(database: Database): global db db = database ``` 在端点中直接引用: ```python from ..global_var import db @rpc_method(subject=".action", method="GET", ...) async def handler(request): async with db.get_session() as session: ... ``` --- ## @rpc_method 装饰器 ```python @rpc_method( subject=".", # 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](auth/README.md) | 用户注册、登录、JWT 鉴权、用户信息管理 | 含 SQLite 数据库 | | [doubao](doubao/README.md) | 火山引擎豆包视觉理解模型集成 | 第三方 API 调用 | | [example](example/README.md) | 四则运算示例 + 访问计数器 | 入门参考 |