lzm-plugin 0.1.1

通用插件化框架 — 将连接器/阅读器/解析器抽象为pip可安装的独立插件，与项目逻辑完全解耦

lzm-plugin — 通用插件化框架

将连接器、阅读器、解析器抽象为 pip 可安装的独立插件，与项目逻辑完全解耦。

---

目录

• 背景
• 安装
• 快速开始
• 核心概念
• 使用指南
  • PluginRegistry — 插件注册表
  • PluginEngine — 执行引擎
  • PluginEventBus — 事件总线
  • CompositionReader — 阅读器编排
  • ParserFactory — 解析器工厂
• 编写自定义插件
• API 参考
• 目录结构
• 相关文档

---

背景

在实际项目中，连接器（Connector）、网页阅读器（Web Reader）、文档解析器（Document Parser）等能力常常深度耦合在项目代码中，导致：

• 难以扩展：新增一种能力需要修改基类
• 依赖膨胀：所有依赖捆绑安装，无论是否使用
• 复用困难：相同能力无法在不同项目之间共享

本包将这些通用能力按最有价值、无数据库依赖的原则抽离为独立 pip 包，通过标准接口定义实现与项目逻辑的完全解耦。

设计目标

目标	实现方式
零依赖核心	核心框架不依赖任何第三方库
按需加载	插件使用 try/except ImportError 懒加载，缺失依赖的插件自动降级
类型安全	全部接口使用类型注解
异步优先	IO 操作均使用 async/await
插件隔离	插件通过 PluginContext 访问外部世界，不直接 import 项目模块

---

安装

# 安装核心框架（零依赖）
pip install lzm-plugin

# 安装全部阅读器（网页内容抓取）
pip install "lzm-plugin[readers]"

# 安装全部解析器（文档转换）
pip install "lzm-plugin[parsers]"

# 安装全部功能
pip install "lzm-plugin[all]"

可选依赖说明

依赖组	包含功能	依赖包
[readers]	网页阅读器：Jina AI、Firecrawl、Trafilatura、HtmlClean 等	httpx, trafilatura, beautifulsoup4, lxml, readability-lxml, markdownify
[parsers]	文档解析器：Markitdown、图片解析	markitdown, Pillow
[all]	全部插件	上述所有

核心包零外部依赖，即使不安装任何可选组，PluginRegistry、PluginEngine、PluginEventBus、MarkdownCleanerPlugin 仍可正常使用。

---

快速开始

5 分钟快速集成

import asyncio
import logging

from lzm.plugin import plugin_registry, PluginEngine, PluginContext

async def main():
    # 1. 创建执行引擎
    engine = PluginEngine(plugin_registry)
    await engine.start()

    # 2. 注册自定义阅读器（见「编写自定义插件」章节）
    # ...

    # 3. 链式编排 — 自动按优先级调用阅读器
    ctx = PluginContext(logger=logging.getLogger("demo"))
    result = await engine.compose_readers("https://example.com/article", ctx)

    if result.success:
        print(f"内容: {result.data}")
    else:
        print(f"失败: {result.error_message}")

    await engine.stop()

asyncio.run(main())

使用内置阅读器一键爬网页

import asyncio
import logging

from lzm.plugin import plugin_registry, PluginContext
from lzm.plugin.readers.composition_reader import read_url

async def main():
    ctx = PluginContext(logger=logging.getLogger("demo"))
    content = await read_url("https://example.com/article", ctx, plugin_registry)
    if content:
        print(content[:500])  # 打印前 500 字符

asyncio.run(main())

---

核心概念

┌──────────────────────────────────────────────────────────────┐
│                     lzm-plugin 架构总览                       │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────────────────────────────────────────┐        │
│  │                  PluginEngine                     │        │
│  │   执行引擎（编排 + 执行 + 状态追踪）                │        │
│  └──────────────┬────────────────────────┬──────────┘        │
│                 │                        │                    │
│         ┌───────▼────────┐      ┌────────▼───────┐           │
│         │ PluginRegistry  │      │ PluginEventBus │           │
│         │ 注册表（索引）    │      │ 事件总线（解耦） │           │
│         └───────┬────────┘      └────────────────┘           │
│                 │                                            │
│    ┌────────────┼────────────┬──────────────────┐            │
│    ▼            ▼            ▼                  ▼            │
│ DataSource  WebReader  ExternalService  DocumentParser      │
│  连接器       阅读器       外部服务           解析器          │
│                                                              │
└──────────────────────────────────────────────────────────────┘

插件类型

类型	用途	方法	示例
DataSourcePlugin	数据源连接器	execute()	飞书文档同步、数据库拉取
WebReaderPlugin	网页阅读器	read()	Jina AI 阅读、Firecrawl 抓取
ExternalServicePlugin	外部服务	execute()	AI 网关、消息推送
DocumentParserPlugin	文档解析器	parse()	Markitdown 转换、图片 OCR

三大核心组件

组件	职责	核心方法
PluginRegistry	插件注册与发现	register()、get()、get_by_type()、discover()
PluginEngine	插件执行与编排	execute_plugin()、compose_readers()、execute_all()
PluginEventBus	解耦通信	publish()、subscribe()、on()

---

使用指南

PluginRegistry — 插件注册表

注册表是插件的"户口本"，负责插件的注册、查询、发现和注销。

from lzm.plugin import plugin_registry
from lzm.plugin.core.base import PluginType

# 注册插件
plugin_registry.register(my_reader)

# 查询插件
by_name = plugin_registry.get("jina-reader")       # 按名称
by_type = plugin_registry.get_by_type(PluginType.WEB_READER)  # 按类型

# 检查存在
exists = plugin_registry.has("jina-reader")

# 列出所有插件
all_plugins = plugin_registry.list_plugins()
# 返回 [{"name": "...", "version": "...", "type": "...", "description": "..."}]

# 注销插件
plugin_registry.unregister("trafilatura-reader")

# 插件总数
count = plugin_registry.count()

# 从目录发现插件（扫描指定目录的 .py 文件）
count = plugin_registry.discover("/path/to/plugins")

注册规则：

• 插件名称必须全局唯一，重复注册会抛出 PluginRegisterError
• 插件必须继承 BasePlugin 并定义 plugin_type 类属性
• 读取不存在的插件会抛出 PluginNotFoundError

PluginEngine — 执行引擎

执行引擎是插件的"大脑"，负责插件的生命周期管理（加载→执行→结果处理）。

import logging
from lzm.plugin import plugin_registry, PluginEngine, PluginContext

engine = PluginEngine(plugin_registry)

# 启动引擎
await engine.start()

ctx = PluginContext(logger=logging.getLogger("app"))

# 1. 执行单个插件
result = await engine.execute_plugin(
    "jina-reader",
    context=ctx,
    url="https://example.com"
)
print(result.success)     # True / False
print(result.data)        # 返回内容
print(result.error_message)  # 错误信息

# 2. 链式编排阅读器（按优先级降序调用）
#    高优先级返回 None → 自动降级到低优先级
result = await engine.compose_readers("https://example.com", ctx)

# 3. 批量执行所有插件
results = await engine.execute_all(context=ctx)
# 返回 {"plugin-name": PluginResult, ...}

# 停止引擎
await engine.stop()

编排策略（compose_readers）：

1. 获取所有 WebReaderPlugin 实例
2. 按 get_priority() 降序排序（优先级 0-100，越高越优先）
3. 依次调用 can_handle(url) → 通过则调用 read(url, ctx)
4. 第一个返回非空内容的即作为最终结果
5. 全部返回空 → 返回错误结果

PluginEventBus — 事件总线

事件总线提供解耦的模块间通信能力，支持通配符匹配和装饰器订阅。

from lzm.plugin.core.event_bus import PluginEventBus, PluginEvent

bus = PluginEventBus()

# ── 普通订阅 ──

async def on_started(event: PluginEvent):
    print(f"[{event.name}] {event.payload}")

bus.subscribe("plugin.started", on_started)

# ── 通配符订阅 ──

@bus.on("data_source.*")
async def on_data_source(event: PluginEvent):
    print(f"数据源事件: {event.name}")

# ── 发布事件 ──

await bus.publish("plugin.started", {"name": "jina-reader"})
await bus.publish("data_source.sync.completed", {"source": "feishu", "count": 100})

# ── 解订阅 ──

bus.unsubscribe("plugin.started", on_started)

# ── 清空所有 ──

await bus.clear()

print(f"当前处理器数: {bus.get_handler_count()}")

事件匹配规则：

• 支持标准 fnmatch 通配符：* 匹配任意字符，? 匹配单个字符
• 示例："plugin.*" 匹配 "plugin.started"、"plugin.stopped" 等
• 一个事件可触发多个匹配的处理器
• 单个处理器异常不会影响其他处理器（异常隔离）

CompositionReader — 阅读器编排

CompositionReader 是 PluginEngine.compose_readers() 的独立封装，提供直接使用的编排能力。

import logging
from lzm.plugin import plugin_registry, PluginContext
from lzm.plugin.readers.composition_reader import CompositionReader, read_url

# 方式一：使用编排器对象
reader = CompositionReader(plugin_registry)
ctx = PluginContext(logger=logging.getLogger("app"))
content = await reader.read("https://example.com", ctx)

# 方式二：使用快捷函数（推荐）
content = await read_url("https://example.com", ctx, plugin_registry)

ParserFactory — 解析器工厂

统一管理文档解析器，支持根据文件扩展名自动选择解析器。

import logging
from lzm.plugin import PluginContext
from lzm.plugin.parsers.factory import ParserFactory

factory = ParserFactory()
ctx = PluginContext(logger=logging.getLogger("app"))

# 1. 从注册表发现解析器
from lzm.plugin import plugin_registry
count = factory.discover_from_registry(plugin_registry)

# 2. 检查是否支持某文件
if factory.can_parse("document.pdf"):
    print("支持 PDF 解析")

# 3. 解析文件
with open("report.pdf", "rb") as f:
    result = await factory.parse_file(f.read(), "report.pdf", context=ctx)
    if result.success:
        print(result.data)

# 4. 列出可用解析器
parsers = factory.list_parsers()
# [{"name": "markitdown-parser", "formats": ["pdf", "docx", ...]}, ...]

# 5. 所有支持的格式
formats = factory.list_supported_formats()
# {"pdf": "markitdown", "docx": "markitdown", ...}

内置解析器及支持格式：

解析器	格式	依赖
MarkitdownParserPlugin	pdf, docx, pptx, xlsx, html, csv, json, xml, rtf, epub, md	markitdown
ImageParserPlugin	jpg, jpeg, png, webp	Pillow

---

编写自定义插件

最简单的阅读器插件（零依赖）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自定义阅读器插件示例

Author: Lzm
Date: 2026-07-15
"""

from lzm.plugin.core.base import WebReaderPlugin, PluginMeta, PluginType, PluginContext

class MyCustomReader(WebReaderPlugin):
    """自定义阅读器，继承 WebReaderPlugin"""

    plugin_type = PluginType.WEB_READER

    def get_meta(self) -> PluginMeta:
        return PluginMeta(
            name="my-custom-reader",        # 全局唯一标识
            version="1.0.0",                # 语义化版本号
            plugin_type=PluginType.WEB_READER,
            description="我的自定义阅读器",
            author="Your Name",
            created_date="2026-07-15",
        )

    def get_priority(self) -> int:
        """优先级：0-100，越高越优先被调用"""
        return 50

    def can_handle(self, url: str) -> bool:
        """是否可以处理此 URL"""
        return url.startswith("https://example.com")

    async def read(self, url: str, context: PluginContext) -> str | None:
        """执行读取，返回 Markdown 内容"""
        # context.logger.info(f"正在读取: {url}")
        # context.extra 可获取自定义配置
        return f"# Content from {url}\n\nHello from my custom reader!"

    async def fallback(self, url: str, context: PluginContext) -> str | None:
        """降级方案（可选）：当 read() 失败时调用"""
        return None

注册并使用自定义插件

import asyncio
import logging

from lzm.plugin import plugin_registry, PluginEngine, PluginContext

async def main():
    # 注册自定义阅读器
    plugin_registry.register(MyCustomReader())

    # 注册内置阅读器（需安装 [readers]）
    from lzm.plugin.readers import HtmlCleanReaderPlugin
    plugin_registry.register(HtmlCleanReaderPlugin())

    # 执行编排
    engine = PluginEngine(plugin_registry)
    await engine.start()

    ctx = PluginContext(logger=logging.getLogger("demo"))
    result = await engine.compose_readers("https://example.com/article", ctx)

    if result.success:
        print(f"读取成功，{len(result.data)} 字符")
    else:
        # 所有阅读器都失败时尝试降级
        print(f"所有阅读器失败: {result.error_message}")

    await engine.stop()

asyncio.run(main())

编写解析器插件

from typing import Any
from lzm.plugin.core.base import (
    DocumentParserPlugin, PluginMeta, PluginType, PluginContext, PluginResult,
)

class MyParser(DocumentParserPlugin):
    plugin_type = PluginType.DOCUMENT_PARSER

    def get_meta(self) -> PluginMeta:
        return PluginMeta(
            name="my-parser",
            version="1.0.0",
            plugin_type=PluginType.DOCUMENT_PARSER,
            description="自定义解析器",
            author="Your Name",
            created_date="2026-07-15",
        )

    def supported_formats(self) -> set[str]:
        """声明支持的扩展名"""
        return {"txt", "md"}

    async def parse(
        self,
        content: bytes,
        filename: str,
        context: PluginContext,
        **kwargs: Any,
    ) -> PluginResult:
        text = content.decode("utf-8")
        return PluginResult.ok(
            data={"text": text, "filename": filename},
            source="my-parser",
        )

---

API 参考

核心数据类

类	字段	说明
PluginType	DATA_SOURCE, WEB_READER, EXTERNAL_SERVICE, DOCUMENT_PARSER	插件类型枚举（StrEnum）
PluginMeta	name, version, plugin_type, description, author, created_date, dependencies	插件元信息
PluginContext	logger, config_loader, storage, event_bus, http_client, cache, extra	运行时上下文
PluginResult	success, data, error_message, metadata	执行结果

PluginResult 提供了两个工厂方法：

# 成功
PluginResult.ok(data={"key": "value"}, source="plugin-name")

# 失败
PluginResult.fail("错误描述", error_code=500, retryable=True)

异常体系

异常	触发场景	属性
PluginNotFoundError	查询/执行不存在的插件	plugin_name
PluginRegisterError	重复注册或不合法的插件	plugin_name, message
PluginExecutionError	插件执行失败	plugin_name, message, original
PluginConfigError	插件配置校验失败	plugin_name, field, message

全部继承自 PluginError 基类。

全局导出

from lzm.plugin import (
    # 数据类
    PluginType, PluginMeta, PluginContext, PluginResult,
    # 基类
    BasePlugin, DataSourcePlugin, WebReaderPlugin,
    ExternalServicePlugin, DocumentParserPlugin,
    # 核心组件
    PluginRegistry, PluginEngine,
    # 全局单例
    plugin_registry,
    # 异常
    PluginError, PluginNotFoundError, PluginRegisterError,
    PluginExecutionError, PluginConfigError,
)

---

目录结构

lzm-plugin/
├── src/lzm/plugin/
│   ├── __init__.py                 # 统一导出
│   ├── core/                       # 核心框架（零依赖）
│   │   ├── base.py                 # PluginType, PluginMeta, PluginContext, PluginResult, BasePlugin
│   │   ├── registry.py            # PluginRegistry（注册/发现/索引）
│   │   ├── engine.py              # PluginEngine（执行/编排/生命周期）
│   │   ├── event_bus.py           # PluginEventBus（发布订阅/通配符）
│   │   └── exceptions.py          # 5 种异常类
│   ├── readers/                    # 网页阅读器（可选依赖）
│   │   ├── __init__.py            # 懒加载（try/except ImportError）
│   │   ├── composition_reader.py  # 编排引擎 + read_url() 快捷函数
│   │   ├── html_clean_reader.py   # readability + markdownify 本地清洗
│   │   ├── jina_reader.py         # Jina AI Reader API
│   │   ├── unifuncs_reader.py     # 中文内容优化
│   │   ├── markdown_cleaner.py    # 零依赖纯文本清洗
│   │   ├── trafilatura_reader.py  # 降级模式支持
│   │   └── firecrawl_reader.py    # JS 渲染支持
│   └── parsers/                    # 文档解析器（可选依赖）
│       ├── __init__.py            # 懒加载
│       ├── base.py                # BaseParserAdapter 兼容层
│       ├── markitdown_parser.py   # 11 种格式
│       ├── image_parser.py        # jpg/png/webp
│       └── factory.py             # ParserFactory 自动发现
├── tests/                          # 102 个测试用例
├── docs/                           # 详细文档
└── pyproject.toml

---

测试

# 安装 dev 依赖
pip install "lzm-plugin[all]" pytest pytest-asyncio

# 运行全部测试
cd lzm-plugin
py -3.12 -m pytest -v

# 运行特定套件
py -3.12 -m pytest tests/test_functional_core.py -v
py -3.12 -m pytest tests/test_functional_plugins.py -v

当前测试覆盖：102 个测试用例，9 大场景，100% 通过。

---

相关文档

• 开发进度时间线
• 常见错误归档
• 编码规范
• TODO 管理
• 逻辑记录

---

许可证

MIT

Copyright (c) 2026 Lzm