kernel2 2.6.7

可插拔 FastAPI 模块内核：懒加载 · ServiceRegistry · 工作流引擎 · Python Shell · 日志监控 · 动态模块运行时

kernel2

Kernel V2.0 — 面向 FastAPI 的可插拔模块系统

最小核心启动 · 按需懒加载 · Redis Pub/Sub IPC · entry_points 插件发现

---

为什么需要 kernel2？

传统 FastAPI 项目随着业务增长，main.py 会积累大量顶层导入和 lifespan 初始化代码：数据库连接池、消息队列、定时任务……每次冷启动都要全量初始化，即使当前请求根本用不到这些服务。

kernel2 将应用拆分为相互隔离的 KernelModule：

启动时（毫秒级）:  注册路由 + 建立 Redis 连接
首次请求时（按需）: 触发对应模块的重量级初始化

	传统方式	kernel2
冷启动	全量初始化	仅核心基础设施
服务隔离	共享 import	模块边界明确
模块间通信	直接函数调用	Redis Pub/Sub
插件扩展	手动注册	entry_points 自动发现

---

安装

pip install fastapi uvicorn

# 可选：Redis IPC（不安装或 Redis 不可达时自动降级，不影响运行）
pip install "redis[asyncio]>=5.0"

---

5 分钟上手

第一步：定义模块

# modules/hello.py
from kernel2 import KernelModule

class HelloModule(KernelModule):
    name = "hello"
    version = "1.0.0"
    description = "演示模块"

    # 哪些路径属于本模块（用于触发懒加载）
    PATH_PREFIXES = ["/api/hello"]

    def register_routes(self, app) -> None:
        """启动时调用，仅做路由注册，禁止 IO 操作"""
        from fastapi import APIRouter
        router = APIRouter(prefix="/api/hello")

        @router.get("")
        async def hello():
            return {"message": "Hello from kernel2!"}

        @router.get("/status")
        async def status():
            # 重量级服务在 on_load 后才可用
            from kernel2 import registry
            svc = registry.get("hello_service", None)
            return {"service_ready": svc is not None}

        app.include_router(router)

    async def on_load(self) -> None:
        """首次命中 PATH_PREFIXES 时触发，执行重量级初始化"""
        import asyncio
        print("[HelloModule] 初始化中...")
        await asyncio.sleep(0.1)  # 模拟耗时操作（DB 连接、加载模型等）

        from kernel2 import registry
        registry.register("hello_service", {"ready": True})
        print("[HelloModule] 就绪")

    async def on_unload(self) -> None:
        """应用关闭时调用"""
        print("[HelloModule] 已卸载")

第二步：创建应用

# main.py
from kernel2 import create_kernel_app
from modules.hello import HelloModule

app = create_kernel_app(
    modules=[HelloModule()],
    title="My App",
)

第三步：运行

uvicorn main:app --reload

# 启动日志（毫秒级）
INFO  [Kernel] started — 1 modules registered

# 首次请求 /api/hello 时触发懒加载
[HelloModule] 初始化中...
[HelloModule] 就绪
INFO  GET /api/hello  200

查看内核状态：

curl http://localhost:8000/_kernel/status

{
  "kernel": "2.0",
  "redis": false,
  "modules": {
    "hello": {
      "loaded": true,
      "version": "1.0.0",
      "description": "演示模块",
      "prefixes": ["/api/hello"]
    }
  }
}

---

核心概念

KernelModule 生命周期

应用启动
  └─ register_routes(app)    ← 同步，仅注册路由，零 IO
  └─ register_services(reg)  ← 同步，注册轻量占位/工厂

首次 HTTP 请求命中 PATH_PREFIXES
  └─ on_load()               ← 异步，重量级初始化（只执行一次）

应用关闭
  └─ on_unload()             ← 异步，释放资源

两阶段路由注册

FastAPI 要求路由必须在 ASGI 生命周期开始前注册。create_kernel_app() 自动处理此时序：

① app = FastAPI(...)
② for mod in modules: mod.register_routes(app)   ← 路由在此注册（正确）
③ ASGI 启动 → lifespan → ModuleLoader 初始化
④ 请求到达 → 懒加载中间件触发 on_load()

⚠️ 不要在 on_load() 内部调用 app.include_router()，那时路由匹配表已锁定。

ServiceRegistry

全局单例，模块间共享服务实例：

from kernel2 import registry

# 注册（在 on_load 中）
registry.register("my_db", db_pool)

# 读取（带默认值，避免未加载时崩溃）
db = registry.get("my_db", None)
if db is None:
    raise HTTPException(503, "服务初始化中")

# 检查是否已注册
if registry.is_registered("my_db"):
    ...

RedisBus（可选）

模块间通过 Redis Pub/Sub 通信，无需直接依赖彼此的 Python 对象：

from kernel2 import registry

bus = registry.get("redis_bus", None)

# 发布事件
if bus:
    await bus.publish("order.completed", {"order_id": "123"})

# 订阅事件（在 on_load 中注册）
if bus:
    bus.subscribe("order.completed", self._handle_order)

Redis 不可用时自动降级：bus.publish() 和 bus.subscribe() 静默跳过，不影响核心业务。

---

手动集成（不用 create_kernel_app）

如果已有 FastAPI 应用，可手动接入：

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from kernel2 import ModuleLoader, RedisBus, registry
from modules.hello import HelloModule

MODULES = [HelloModule()]

@asynccontextmanager
async def lifespan(app):
    bus = RedisBus(os.getenv("REDIS_URL", "redis://localhost:6379/0"))
    await bus.connect()
    registry.register("redis_bus", bus)

    loader = ModuleLoader(app, registry, bus)
    loader.register_all(MODULES, skip_routes=True)  # ← skip_routes 必须为 True
    loader.discover_plugins()
    registry.register("module_loader", loader)

    yield

    await loader.unload_all()
    await bus.disconnect()

app = FastAPI(lifespan=lifespan)

# ① 路由在 ASGI 生命周期前注册（关键）
for mod in MODULES:
    mod.register_routes(app)

# ② 懒加载中间件
@app.middleware("http")
async def lazy_loader(request: Request, call_next):
    loader = registry.get("module_loader", None)
    if loader:
        name = loader.detect_module(request.url.path)
        if name:
            await loader.load(name)
    return await call_next(request)

---

插件发现

第三方包通过 pyproject.toml 声明插件，无需修改主应用代码：

# 插件包的 pyproject.toml
[project.entry-points."kernel.modules"]
my_plugin = "my_package:MyPluginModule"

主应用调用 loader.discover_plugins() 自动扫描已安装的插件并注册。

---

项目结构

kernel2/
├── __init__.py        # 公开导出：KernelModule, registry, ModuleLoader, RedisBus, create_kernel_app
├── module.py          # KernelModule 基类
├── registry.py        # ServiceRegistry 单例
├── loader.py          # ModuleLoader（懒加载 + 插件发现）
├── bus.py             # RedisBus（Pub/Sub IPC）
├── factory.py         # create_kernel_app() 工厂
├── requirements.txt
└── BEST_PRACTICES.md  # 详细规范与常见陷阱

---

API 速查

create_kernel_app(modules, redis_url, reg, **fastapi_kwargs) → FastAPI

参数	类型	说明
modules	list[KernelModule]	内置模块实例列表
redis_url	str	Redis URL，默认 redis://localhost:6379/0
reg	ServiceRegistry | None	自定义 Registry，默认全局单例
**fastapi_kwargs	—	直接传给 FastAPI(...)

自动提供：懒加载中间件 · /_kernel/status 端点 · 优雅关闭

---

KernelModule 属性与方法

成员	类型	说明
name	str	模块唯一标识（必填）
version	str	版本号，默认 "1.0.0"
description	str	模块描述
PATH_PREFIXES	list[str]	触发懒加载的路径前缀
is_loaded	bool	是否已完成 on_load()
register_routes(app)	同步	注册路由，零 IO
register_services(reg)	同步	注册轻量服务占位
on_load()	异步	重量级初始化（仅执行一次）
on_unload()	异步	资源释放

---

ModuleLoader 常用方法

loader = registry.get("module_loader")

await loader.load("my_module")          # 手动触发加载（幂等）
await loader.ensure_loaded("a", "b")   # 并发加载多个模块（预热）
loader.status()                         # 返回所有模块状态 dict
await loader.unload_all()               # 优雅关闭所有已加载模块

---

环境变量

变量	默认值	说明
REDIS_URL	redis://localhost:6379/0	Redis 连接地址（手动集成时使用）

---

进一步阅读

• BEST_PRACTICES.md — 模块设计规范、常见陷阱、测试指南、插件开发