add: workers/ 进程插件目录 + demo 示例(external 模式经 Zenoh 注册路由)

This commit is contained in:
2026-08-14 17:44:19 +08:00
parent b9498750ff
commit 1f727bcdd7
12 changed files with 183 additions and 4 deletions
+4 -1
View File
@@ -53,11 +53,13 @@ chongming-zenoh/
│ │ ├── api/generated/ # 自动生成 API 客户端 │ │ ├── api/generated/ # 自动生成 API 客户端
│ │ ├── views/ # 页面组件 │ │ ├── views/ # 页面组件
│ │ └── router/ # Vue Router 配置 │ │ └── router/ # Vue Router 配置
│ └── plugins/ # Worker 插件源码 │ └── plugins/ # Worker 插件源码(inline 模式)
│ ├── auth/ # JWT 鉴权 │ ├── auth/ # JWT 鉴权
│ ├── doubao/ # 豆包 AI(支持 AICC 安全通道) │ ├── doubao/ # 豆包 AI(支持 AICC 安全通道)
│ ├── aibox_sophon_adapter/ # 设备心跳 + 告警 │ ├── aibox_sophon_adapter/ # 设备心跳 + 告警
│ └── example/ # 演示示例 │ └── example/ # 演示示例
├── workers/ # 进程插件(external 模式,独立进程经 Zenoh 注册路由)
│ └── demo/ # 进程插件示例
├── utils/ # 独立工具包 ├── utils/ # 独立工具包
│ ├── bytedance/ # 字节跳动安全通信 SDK │ ├── bytedance/ # 字节跳动安全通信 SDK
│ ├── config/ # TOML 配置加载 │ ├── config/ # TOML 配置加载
@@ -354,6 +356,7 @@ uv pip install -e utils/<name> # 安装/更新工具包
| **doubao** | `src/plugins/doubao/` | 豆包视觉理解,支持 AICC 安全通信通道 | | **doubao** | `src/plugins/doubao/` | 豆包视觉理解,支持 AICC 安全通信通道 |
| **aibox_sophon_adapter** | `src/plugins/aibox_sophon_adapter/` | AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传 | | **aibox_sophon_adapter** | `src/plugins/aibox_sophon_adapter/` | AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传 |
| **example** | `src/plugins/example/` | 四则运算 + 心跳演示(含 MQTT 端点示例) | | **example** | `src/plugins/example/` | 四则运算 + 心跳演示(含 MQTT 端点示例) |
| **demo**(进程插件示例) | `workers/demo/` | 独立进程运行(load_mode=external),经 Zenoh 注册路由 |
--- ---
+6
View File
@@ -54,6 +54,12 @@ def worker_serve():
parser.add_argument("--config", default="config.toml", help="全局配置文件路径") parser.add_argument("--config", default="config.toml", help="全局配置文件路径")
args = parser.parse_args() args = parser.parse_args()
# 控制台脚本启动时 sys.path 不含项目根目录,显式加入以保证
# plugins.<name>(inline 插件)与 workers.<name>(进程插件)均可导入
project_root = os.path.abspath(os.getcwd())
if project_root not in sys.path:
sys.path.insert(0, project_root)
config = load_config() config = load_config()
logger = get_logger("app") logger = get_logger("app")
+5 -1
View File
@@ -46,6 +46,7 @@ def _load_plugin_models(worker_type: str) -> Optional[Set[str]]:
for mod_name in ( for mod_name in (
f"plugins.{worker_type}.models.pydantic", f"plugins.{worker_type}.models.pydantic",
f"src.plugins.{worker_type}.models.pydantic", f"src.plugins.{worker_type}.models.pydantic",
f"workers.{worker_type}.models.pydantic",
): ):
try: try:
mod = importlib.import_module(mod_name) mod = importlib.import_module(mod_name)
@@ -62,7 +63,10 @@ def _load_plugin_models(worker_type: str) -> Optional[Set[str]]:
and id(obj) not in seen_objs and id(obj) not in seen_objs
): ):
seen_objs.add(id(obj)) seen_objs.add(id(obj))
names.add(name) # 用 obj.__name__ 而非 dir() 名称:gen_models 生成的
# <Plugin>Response 别名与真实模型同一对象,别名保留原类名,
# 按对象去重后自动收敛到真实模型名(避免别名抢占排序位)
names.add(obj.__name__)
return names return names
return None return None
+19 -1
View File
@@ -101,6 +101,23 @@ def generate_pydantic_code(
routes: list[dict[str, Any]], routes: list[dict[str, Any]],
) -> str: ) -> str:
"""生成 pydantic.py 文件内容""" """生成 pydantic.py 文件内容"""
# 收集全部字段类型,按需生成 typing 导入(避免未使用导入触发 lint)
type_strings: list[str] = []
has_optional_field = False
for route in routes:
for p in route.get("parsed_params", []):
type_strings.append(p.get("type", "str"))
for f in route.get("parsed_response", []):
type_strings.append(f.get("type", "str"))
# 非必填字段渲染为 Optional[...]
if not f.get("required", True):
has_optional_field = True
typing_imports = []
if has_optional_field or any("Optional" in t for t in type_strings):
typing_imports.append("Optional")
if any("Any" in t for t in type_strings):
typing_imports.append("Any")
lines = [] lines = []
lines.append('"""') lines.append('"""')
lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}") lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}")
@@ -109,7 +126,8 @@ def generate_pydantic_code(
lines.append("") lines.append("")
lines.append("from __future__ import annotations") lines.append("from __future__ import annotations")
lines.append("") lines.append("")
lines.append("from typing import Any, Optional") if typing_imports:
lines.append(f"from typing import {', '.join(typing_imports)}")
lines.append("from pydantic import BaseModel, Field") lines.append("from pydantic import BaseModel, Field")
lines.append("") lines.append("")
Generated
+1 -1
View File
@@ -319,7 +319,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git
[[package]] [[package]]
name = "chongming-worker" name = "chongming-worker"
version = "0.1.0" version = "0.1.0"
source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#a58e92c21e5bf4b3ada55b2abc894e8ead7d17bc" } source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#4908378e597eb200e255c71df92d16e272b4b12e" }
dependencies = [ dependencies = [
{ name = "eclipse-zenoh" }, { name = "eclipse-zenoh" },
{ name = "tomli" }, { name = "tomli" },
+8
View File
@@ -0,0 +1,8 @@
"""
进程插件目录(load_mode = "external" 的插件)
与 src/plugins/(inline 插件)结构一致,区别仅在运行方式:
- 由 Docker/K8s 编排拉起独立进程(容器入口 uv run worker_serve --type <name>)
- 进程自读 config/workers/<name>/ 配置、进程内契约校验,经 Zenoh
(chongming/routes/**)注册路由,网关动态挂载 HTTP 端点
"""
+41
View File
@@ -0,0 +1,41 @@
"""
Demo 进程插件 — 独立进程运行示例
与 inline 插件(src/plugins/)结构完全一致,区别仅在运行方式:
- config/workers/demo/config.toml 声明 load_mode = "external"、module = "workers.demo"
- 由 Docker/K8s 编排拉起独立进程: uv run worker_serve --type demo
- 启动流程: 自读 TOML → 进程内契约校验 → create_workers →
经 Zenoh 发布路由描述符(网关 RouteRegistry 动态挂载 HTTP 路由)→
每 10s 周期重发布自愈 → SIGTERM 时摘除路由并 cleanup
"""
import logging
from chongming_worker.plugin import Plugin
from .router import * # noqa: F401,F403 触发 @rpc_method 注册
logger = logging.getLogger(__name__)
class DemoPlugin(Plugin):
"""Demo 插件 — 无数据库的进程插件最小示例。"""
type = "demo"
db_name = None # 无数据库
auto_bind = True # 自动绑定 models/pydantic.py(支持 workers/ 路径)
async def on_init(self):
"""首次 create_workers 时调用一次:读取插件私有配置。"""
prefix = self.config.get("greeting_prefix", "Hello")
self.set_var("greeting_prefix", prefix)
logger.info("[demo] 初始化完成 (greeting_prefix=%s)", prefix)
async def on_cleanup(self):
"""cleanup 时调用(无外部资源需要释放)。"""
logger.info("[demo] 清理完成")
# 暴露给 worker_serve 的模块级入口(签名与 inline 插件一致)
plugin = DemoPlugin()
create_workers = plugin.create_workers
cleanup = plugin.cleanup
+32
View File
@@ -0,0 +1,32 @@
"""
Demo 进程插件全局运行时状态 — 与 inline 插件约定一致
约定:
- 模块只做"声明默认值",不产生任何导入期副作用
- 所有值统一在 on_init 中通过 set_global_var 显式写入
- set_global_var 仅允许写入 _ALLOWED 白名单,拼写错误会立即抛错
"""
from typing import Any
# 允许写入的全局变量白名单(防止拼写错误静默创建无用的新变量)
_ALLOWED = frozenset({"greeting_prefix"})
# 显式声明所有运行时状态(默认值,运行时由 on_init 写入)
greeting_prefix: str = "Hello"
def set_global_var(name: str, value: Any) -> None:
"""白名单校验版 setter — 设置模块级运行时状态。"""
if name not in _ALLOWED:
raise KeyError(f"global_var 未知变量: {name}(允许: {sorted(_ALLOWED)})")
globals()[name] = value
def require(name: str) -> Any:
"""取"必须已存在"的值,未初始化时抛清晰错误。"""
if name not in _ALLOWED:
raise KeyError(f"global_var 未知变量: {name}")
return globals()[name]
__all__ = ["greeting_prefix", "set_global_var", "require"]
+14
View File
@@ -0,0 +1,14 @@
"""
模型导出 — demo
自动生成,请勿手动修改
"""
from .pydantic import DemoResponse
from .pydantic import HelloGetRequest
from .pydantic import HelloGetResponse
__all__ = [
"DemoResponse",
"HelloGetRequest",
"HelloGetResponse",
]
+23
View File
@@ -0,0 +1,23 @@
"""
自动生成的 Pydantic 模型 — demo
来源: config/workers/demo/
"""
from __future__ import annotations
from typing import Optional
from pydantic import BaseModel, Field
class HelloGetRequest(BaseModel):
"""进程插件示例:问候 — 请求体"""
name: str = Field(...)
class HelloGetResponse(BaseModel):
"""进程插件示例:问候 — 响应体"""
greeting: str = Field(...)
worker_id: Optional[str] = Field(default='')
timestamp: float = Field(...)
# 所有响应模型的联合类型
DemoResponse = HelloGetResponse
+5
View File
@@ -0,0 +1,5 @@
"""Demo 进程插件路由端点"""
from . import hello # noqa: F401 触发 @rpc_method 注册
__all__ = ["hello"]
+25
View File
@@ -0,0 +1,25 @@
"""问候端点 — 演示框架参数注入(worker_id)与插件私有配置读取"""
import time
from chongming_worker import rpc_method
from .. import global_var as gv
from ..models.pydantic import HelloGetRequest, HelloGetResponse
@rpc_method(
subject="demo.hello",
method="GET",
summary="进程插件示例:问候",
response_model=HelloGetResponse,
)
async def hello(
hello_get_request: HelloGetRequest,
worker_id: str = "",
) -> HelloGetResponse:
"""返回问候语 + 处理请求的 worker 副本 ID(证明请求经独立进程处理)"""
return HelloGetResponse(
greeting=f"{gv.greeting_prefix}, {hello_get_request.name}!",
worker_id=worker_id,
timestamp=time.time(),
)