[RFC] OpenViking 内容解析插件化架构

[MaojiaSheng]

[RFC] OpenViking 内容解析插件化架构

1. 概述

OpenViking 需要一个灵活的插件化架构，以支持：

• 更复杂的内容理解能力
• 领域特定的解析需求（如法律文档、医学报告等）
• 第三方开发者贡献的解析插件
• 远程解析服务集成

2. 当前架构分析

2.1 现有解析系统结构

openviking/parse/
├── base.py              # 基础定义：NodeType、ParseResult、ResourceNode
├── registry.py          # 解析器注册表（ParserRegistry）
├── parsers/             # 具体解析器实现
│   ├── base_parser.py   # 抽象基类 BaseParser
│   ├── pdf.py           # PDFParser
│   ├── markdown.py      # MarkdownParser
│   ├── word.py          # WordParser
│   ├── excel.py         # ExcelParser
│   └── ...              # 其他解析器
├── custom.py            # 自定义解析器支持
└── tree_builder.py      # 文档树构建工具

2.2 现有扩展机制

当前系统提供了两种扩展方式：

• CustomParserProtocol - 基于协议的自定义解析器
• CallbackParserWrapper - 回调函数包装器

这些机制允许在运行时动态添加解析器，但缺乏：

• 插件生命周期管理
• 配置管理
• 版本控制
• 依赖管理
• 远程服务集成

3. 插件化架构设计

3.1 核心概念

插件类型

类型	描述	使用场景
本地插件	Python 包形式的插件，安装在 OpenViking 的 venv 中	高性能、频繁使用的解析器
远程插件	通过 API 访问的外部服务	资源密集型、专有算法、第三方服务
内置插件	OpenViking 官方提供的解析器	核心功能，默认安装

插件元数据

每个插件需要提供以下元数据：

{
  "name": "medical-document-parser",
  "version": "1.0.2",
  "author": "OpenViking Team",
  "description": "解析医学文档格式，提取结构化数据",
  "type": "local",
  "entry_point": "medical_parser:MedicalDocumentParser",
  "supported_extensions": [".med", ".medical"],
  "dependencies": ["pydicom>=2.0", "nltk>=3.8"],
  "config_schema": {
    "ocr_enabled": {"type": "boolean", "default": true},
    "entity_extraction": {"type": "string", "default": "basic"}
  }
}

3.2 架构层次

┌─────────────────────────────────────────┐
│    OpenViking 应用层                    │
└─────────────────────────────────────────┘
                    │
┌─────────────────────────────────────────┐
│   解析器注册表 (ParserRegistry)         │
│  - 插件发现与加载                       │
│  - 解析器选择策略                       │
│  - 插件生命周期管理                     │
└─────────────────────────────────────────┘
                    │
┌─────────────────────────────────────────┐
│   插件适配器层 (Plugin Adapters)        │
│  - 本地插件适配器 (LocalPluginAdapter)  │
│  - 远程插件适配器 (RemotePluginAdapter)│
│  - 内置插件适配器 (BuiltinPluginAdapter)│
└─────────────────────────────────────────┘
                    │
┌──────────────────┬──────────────────┬────┐
│  本地插件        │  远程插件        │ 内置│
└──────────────────┴──────────────────┴────┘

4. 插件管理命令

4.1 核心命令

# 使用 pip 直接安装插件（推荐）
pip install openviking-medical-parser

# 初始化插件配置（为插件创建配置文件）
openviking-server plugins install medical-parser

# 列出所有可用插件
openviking-server plugins list
openviking system plugins list

# 配置插件
openviking-server plugins config medical-parser --set ocr_enabled=false

# 测试插件
openviking-server plugins test medical-parser --file test_document.med

# 更新插件
openviking-server plugins update medical-parser

# 注销插件
openviking-server plugins uninstall medical-parser

4.2 配置目录结构

~/.openviking/
├── config/
│   ├── plugins.toml          # 插件全局配置
│   └── plugins/              # 插件配置子目录
│       ├── medical_parser.toml
│       └── legal_parser.toml
└── data/
    └── plugins/              # 插件数据目录
        └── medical_parser/

5. 插件接口规范

5.1 解析器接口（本地插件）

所有本地插件必须继承 BaseParser 并实现以下接口：

from openviking.parse import BaseParser
from openviking.parse.base import ParseResult

class MedicalDocumentParser(BaseParser):
    """解析医学文档的插件"""

    @property
    def supported_extensions(self):
        return [".med", ".medical"]

    async def parse(self, source, instruction="", **kwargs) -> ParseResult:
        # 解析逻辑
        pass

    async def parse_content(self, content, source_path=None, instruction="", **kwargs):
        # 直接解析内容
        pass

6. 解析器匹配钩子机制

6.1 设计目标

结合 ov add-resource 的输入参数（path, reason, instruction）设计一个表现良好的钩子机制，支持：

1. 简单的正则表达式匹配 - 快速识别常见资源类型
2. 复杂的语义判断 - 基于大模型理解内容特征
3. 灵活的配置 - 允许用户和插件开发者定义匹配规则
4. 组合匹配 - 支持多个钩子按优先级组合使用

6.2 钩子机制架构

class ParserMatchingHook:
    """解析器匹配钩子的基础接口"""

    @property
    def priority(self) -> int:
        """钩子优先级，数值越小优先级越高（范围：0-100）"""
        return 50

    async def match(self, context: ParserMatchingContext) -> Optional[ParserMatchResult]:
        """
        判断是否匹配该钩子对应的解析器

        Args:
            context: 匹配上下文，包含 path, reason, instruction 等信息

        Returns:
            匹配结果，包含解析器名称、置信度、匹配依据等
        """
        raise NotImplementedError()


class ParserMatchingContext:
    """解析器匹配上下文信息"""

    def __init__(
        self,
        path: str,
        reason: str = "",
        instruction: str = "",
        meta: Optional[dict] = None,
    ):
        self.path = path                    # 资源路径（URL 或本地文件）
        self.reason = reason                # 添加资源的原因
        self.instruction = instruction      # 处理指令
        self.meta = meta or {}              # 额外元数据
        self.effective_instruction = instruction or reason


class ParserMatchResult:
    """解析器匹配结果"""

    def __init__(
        self,
        parser_name: str,
        confidence: float = 1.0,
        match_criteria: str = "",
        explanation: str = "",
    ):
        self.parser_name = parser_name      # 匹配的解析器名称
        self.confidence = confidence        # 匹配置信度（0-1）
        self.match_criteria = match_criteria # 匹配标准（如 "url_pattern", "semantic"）
        self.explanation = explanation      # 匹配解释（用于调试）

6.3 内置钩子实现

6.3.1 URL 模式匹配钩子

class UrlPatternMatchingHook(ParserMatchingHook):
    """基于 URL 模式匹配的钩子"""

    priority = 10  # 高优先级

    def __init__(self, parser_name: str, patterns: List[str]):
        self.parser_name = parser_name
        self.patterns = patterns

    async def match(self, context: ParserMatchingContext) -> Optional[ParserMatchResult]:
        if context.path.startswith("http://") or context.path.startswith("https://"):
            for pattern in self.patterns:
                if re.search(pattern, context.path):
                    return ParserMatchResult(
                        parser_name=self.parser_name,
                        confidence=0.9,
                        match_criteria="url_pattern",
                        explanation=f"URL matches pattern: {pattern}"
                    )
        return None

6.3.2 语义匹配钩子（基于大模型）

class SemanticMatchingHook(ParserMatchingHook):
    """基于大模型语义理解的钩子"""

    priority = 80  # 较低优先级，作为 fallback

    def __init__(self, parser_name: str, description: str):
        self.parser_name = parser_name
        self.description = description

    async def match(self, context: ParserMatchingContext) -> Optional[ParserMatchResult]:
        # 使用大模型判断是否需要使用该解析器
        if context.effective_instruction:
            prompt = (
                f"根据以下资源添加的原因和处理指令，判断是否需要使用 {self.parser_name} "
                f"解析器来解析资源 '{context.path}'。\n\n"
                f"解析器描述：{self.description}\n"
                f"添加原因：{context.reason}\n"
                f"处理指令：{context.instruction}\n\n"
                "回答：是 或 否"
            )

            try:
                response = await llm.query(prompt)  # 调用大模型

                if "是" in response:
                    return ParserMatchResult(
                        parser_name=self.parser_name,
                        confidence=0.8,
                        match_criteria="semantic",
                        explanation=f"LLM 判断需要使用 {self.parser_name}"
                    )
            except Exception as e:
                logger.warning(f"Semantic matching failed: {e}")

        return None

6.3.3 文件类型匹配钩子

class FileExtensionMatchingHook(ParserMatchingHook):
    """基于文件扩展名匹配的钩子"""

    priority = 20

    def __init__(self, parser_name: str, extensions: List[str]):
        self.parser_name = parser_name
        self.extensions = extensions

    async def match(self, context: ParserMatchingContext) -> Optional[ParserMatchResult]:
        # 对于本地文件
        if context.path and len(context.path) <= 1024 and "\n" not in context.path:
            try:
                path_obj = Path(context.path)
                if path_obj.exists() and path_obj.suffix.lower() in self.extensions:
                    return ParserMatchResult(
                        parser_name=self.parser_name,
                        confidence=0.95,
                        match_criteria="file_extension",
                        explanation=f"File extension matches: {path_obj.suffix}"
                    )
            except Exception:
                pass

        return None

6.3.4 关键词匹配钩子（匹配 reason 和 instruction）

class KeywordMatchingHook(ParserMatchingHook):
    """基于 reason 和 instruction 关键词匹配的钩子"""

    priority = 50

    def __init__(self, parser_name: str, keywords: List[str], match_reason: bool = True, match_instruction: bool = True):
        self.parser_name = parser_name
        self.keywords = [kw.lower() for kw in keywords]
        self.match_reason = match_reason
        self.match_instruction = match_instruction

    async def match(self, context: ParserMatchingContext) -> Optional[ParserMatchResult]:
        """匹配 reason 或 instruction 中的关键词"""
        matched_keywords = []

        if self.match_reason and context.reason:
            for keyword in self.keywords:
                if keyword in context.reason.lower():
                    matched_keywords.append(keyword)

        if self.match_instruction and context.instruction:
            for keyword in self.keywords:
                if keyword in context.instruction.lower():
                    matched_keywords.append(keyword)

        if matched_keywords:
            # 计算匹配度：匹配的关键词数量 / 总关键词数量
            confidence = min(len(set(matched_keywords)) / len(self.keywords), 0.9)

            return ParserMatchResult(
                parser_name=self.parser_name,
                confidence=confidence,
                match_criteria="keywords",
                explanation=f"Matched keywords: {', '.join(set(matched_keywords))}"
            )

        return None

6.4 钩子配置系统

# ~/.openviking/config/plugins.toml

[parse.hooks]
enabled = true

[parse.hooks.url_patterns]
wechat = { parser = "wechat-parser", patterns = ["mp\\.weixin\\.qq\\.com/s/", "mp\\.weixin\\.qq\\.com/mp/"], priority = 10 }
feishu = { parser = "feishu-parser", patterns = ["\\.feishu\\.cn/", "\\.larksuite\\.com/"], priority = 15 }
github = { parser = "github-parser", patterns = ["github\\.com/[^/]+/[^/]+"], priority = 20 }

[parse.hooks.semantic]
medical_documents = { parser = "medical-parser", description = "专门用于解析医学文档、病例报告、医疗文献的解析器", priority = 80 }
legal_documents = { parser = "legal-parser", description = "专门用于解析法律文档、合同、法规的解析器", priority = 85 }

[parse.hooks.file_extensions]
pdf = { parser = "pdf-parser", extensions = [".pdf"], priority = 20 }
markdown = { parser = "markdown-parser", extensions = [".md", ".markdown"], priority = 25 }

[parse.hooks.keywords]
medical = { parser = "medical-parser", keywords = ["医学", "病例", "诊断", "药品", "治疗"], match_reason = true, match_instruction = true, priority = 50 }
legal = { parser = "legal-parser", keywords = ["法律", "合同", "法规", "诉讼", "权利"], match_reason = true, match_instruction = true, priority = 55 }
technical = { parser = "technical-parser", keywords = ["架构", "设计", "代码", "技术", "系统"], match_reason = true, match_instruction = true, priority = 60 }

6.5 钩子管理命令

# 列出所有钩子
ov plugins hooks list

# 添加 URL 模式匹配钩子
ov plugins hooks add url-pattern \
  --name wechat-parser \
  --patterns "mp\.weixin\.qq\.com/s/" "mp\.weixin\.qq\.com/mp/" \
  --priority 10

# 添加语义匹配钩子
ov plugins hooks add semantic \
  --name medical-parser \
  --description "专门用于解析医学文档的解析器" \
  --priority 80

# 添加关键词匹配钩子
ov plugins hooks add keyword \
  --name medical-parser \
  --keywords "医学" "病例" "诊断" "药品" "治疗" \
  --match-reason true \
  --match-instruction true \
  --priority 50

# 测试钩子匹配
ov plugins hooks test wechat-parser --path "https://mp.weixin.qq.com/s/E4yeEnX459XbpWuP4Wmkig"
ov plugins hooks test medical-parser --reason "这是一篇关于医学诊断的文章" --instruction "请解析这篇医学文档"

# 删除钩子
ov plugins hooks remove wechat-parser

6.6 匹配流程优化

async def match_parser(context: ParserMatchingContext, registry: ParserRegistry) -> Optional[BaseParser]:
    """
    优化后的解析器匹配流程

    1. 按优先级从低到高收集所有钩子（相同优先级按注册顺序）
    2. 异步并行执行所有钩子的 match 方法
    3. 过滤有效匹配（confidence > 0.5）
    4. 按优先级和置信度排序
    5. 返回最佳匹配
    """

    # 获取所有配置的钩子
    all_hooks = get_all_configured_hooks()

    # 按优先级排序（数值越小优先级越高）
    sorted_hooks = sorted(all_hooks, key=lambda h: h.priority)

    # 异步执行所有匹配
    tasks = []
    for hook in sorted_hooks:
        task = asyncio.create_task(hook.match(context))
        task.hook = hook
        tasks.append(task)

    # 收集匹配结果
    matches = []
    done, pending = await asyncio.wait(tasks, timeout=5.0)  # 5秒超时

    for task in done:
        result = task.result()
        if result and result.confidence > 0.5:
            matches.append(result)

    # 按优先级和置信度排序
    matches.sort(key=lambda m: (-m.confidence, all_hooks.index(m.hook)))

    if matches:
        best_match = matches[0]
        parser = registry.get_parser(best_match.parser_name)

        if parser:
            logger.debug(
                f"Selected parser '{best_match.parser_name}' "
                f"with confidence {best_match.confidence:.2f} "
                f"({best_match.match_criteria})"
            )
            return parser

    return None

6.7 使用示例

6.7.1 微信公众号文章的匹配过程

# 微信公众号文章的匹配过程
context = ParserMatchingContext(
    path="https://mp.weixin.qq.com/s/E4yeEnX459XbpWuP4Wmkig",
    reason="添加一篇关于 AI 架构设计的文章",
    instruction="请详细解析这篇文章的架构设计思路"
)

# 1. URL 模式匹配钩子（priority=10）快速匹配到 wechat-parser
# 2. 无需继续执行其他钩子（由于优先级高）
parser = await match_parser(context, registry)

# parser 为 WeChatParser 实例
result = await parser.parse(context.path)

6.7.2 医学文档的匹配过程（基于关键词匹配）

# 医学文档的匹配过程（基于关键词匹配）
context = ParserMatchingContext(
    path="/path/to/medical_report.pdf",
    reason="这是一份关于糖尿病患者的诊断报告",
    instruction="请解析这份医学报告，重点关注患者的血糖水平和治疗方案"
)

# 1. 首先执行 URL 模式匹配钩子（没有匹配）
# 2. 然后执行文件扩展名匹配钩子（匹配 .pdf → pdf-parser, priority=20）
# 3. 接着执行关键词匹配钩子（匹配 "医学"、"诊断"、"治疗" → medical-parser, priority=50）
# 4. 最后执行语义匹配钩子（优先级最低）

parser = await match_parser(context, registry)

# 最终选择 pdf-parser（优先级更高）
# 如果我们想让医学文档专门使用 medical-parser，可以调整优先级配置
result = await parser.parse(context.path)

6.7.3 技术文档的匹配过程（综合匹配）

# 技术文档的匹配过程（综合匹配）
context = ParserMatchingContext(
    path="https://github.com/openviking/openviking",
    reason="OpenViking 的官方代码仓库",
    instruction="请解析这个代码仓库的架构设计"
)

# 1. URL 模式匹配钩子（匹配 github.com/[^/]+/[^/]+ → github-parser, priority=20）
# 2. 同时关键词匹配钩子也匹配到了 "架构"、"代码" → technical-parser, priority=60）
# 3. URL 模式匹配钩子优先级更高，所以选择 github-parser

parser = await match_parser(context, registry)
result = await parser.parse(context.path)

7. 真实用例：微信公众号解析插件

7.1 插件实现

from openviking.parse import BaseParser
from openviking.parse.parsers.html import HTMLParser
import re
from bs4 import BeautifulSoup
import json


class WeChatParser(BaseParser):
    """
    微信公众号文章解析插件

    专门针对微信公众号文章的结构特点进行优化：
    - 提取公众号信息（作者、发布时间、阅读量等）
    - 优化内容提取和转换
    - 处理微信特有的富媒体内容
    """

    @property
    def supported_extensions(self):
        return []  # 不直接关联文件扩展名，通过钩子机制进行 URL 模式匹配

    async def parse(self, source, instruction="", **kwargs):
        """解析微信公众号文章"""
        # 不再在解析器内部进行 URL 模式匹配
        # 完全依赖钩子机制进行解析器选择

        html_parser = HTMLParser()
        html_result = await html_parser.parse(source, instruction, **kwargs)
        enhanced_result = self._enhance_with_wechat_specifics(html_result, source)
        return enhanced_result

    def _enhance_with_wechat_specifics(self, result, url):
        """增强解析结果，添加微信特有的元数据"""
        result.meta["source_type"] = "wechat_article"
        result.meta["url_pattern"] = "wechat"
        result.parser_name = "WeChatParser"
        return result

7.2 使用流程

7.2.1 安装和配置

# 1. 安装插件
pip install openviking-wechat-parser

# 2. 初始化配置（确保插件被正确配置）
openviking-server plugins install wechat-parser

# 3. 验证钩子配置（可选，检查是否已正确配置 URL 模式匹配）
ov plugins hooks list

7.2.2 钩子配置示例

在 ~/.openviking/config/plugins.toml 中应该自动添加以下配置：

[parse.hooks]
enabled = true

[parse.hooks.url_patterns]
wechat = { parser = "wechat-parser", patterns = ["mp\\.weixin\\.qq\\.com/s/", "mp\\.weixin\\.qq\\.com/mp/"], priority = 10 }

7.2.3 使用插件

# 使用插件（钩子机制自动识别）
ov add-resource https://mp.weixin.qq.com/s/E4yeEnX459XbpWuP4Wmkig

# 添加资源时包含处理指令（钩子机制可以使用这些信息进行语义匹配）
ov add-resource https://mp.weixin.qq.com/s/E4yeEnX459XbpWuP4Wmkig \
  --reason "添加一篇关于 OpenViking 架构设计的微信文章" \
  --instruction "请详细解析这篇文章的内容，重点关注架构设计思路"

7.2.4 测试钩子匹配

# 测试钩子是否能正确匹配微信公众号 URL
ov plugins hooks test wechat-parser --path "https://mp.weixin.qq.com/s/E4yeEnX459XbpWuP4Wmkig"

系统行为：

• 钩子机制自动识别微信公众号 URL 模式
• 选择 WeChatParser 进行解析
• 优化内容提取（标题、作者、发布时间、阅读量）
• 处理微信特有的富媒体内容
• 完整的元数据提取

匹配流程：

1. URL 模式匹配钩子（priority=10）快速识别微信公众号 URL
2. 返回匹配结果，包含解析器名称和置信度（0.9）
3. ParserRegistry 获取 WeChatParser 实例
4. 执行解析过程

8. 远程插件 API 示例

8.1 请求格式

POST /api/v1/parse HTTP/1.1
Host: parser.example.com
Content-Type: application/json

{
  "source_type": "file_path",
  "source": "/path/to/document.med",
  "instruction": "提取患者信息和诊断结果",
  "config": {
    "ocr_enabled": true
  }
}

8.2 响应格式

{
  "success": true,
  "result": {
    "root": {
      "type": "root",
      "title": "患者诊断报告",
      "children": [...]
    },
    "parser_name": "medical-document-parser",
    "parser_version": "1.0.2",
    "parse_time": 2.5
  },
  "warnings": []
}

8. 安全性考虑

8.1 本地插件安全

• 沙箱执行：使用 Python 安全机制隔离插件
• 权限控制：插件只能访问配置允许的资源
• 签名验证：官方插件需要数字签名验证
• 依赖扫描：检查插件依赖的安全性

8.2 远程插件安全

• 网络隔离：可配置代理和防火墙规则
• API 密钥管理：加密存储和传输
• 请求验证：检查请求来源和内容完整性
• 响应验证：验证返回结果的格式和安全性

9. 结论

OpenViking 内容解析插件化架构将提供：

1. 高度可扩展性：支持无限扩展的解析能力
2. 灵活的部署选项：本地、远程和混合部署
3. 丰富的生态系统：开发者友好的 API 和文档
4. 企业级特性：安全、监控和管理功能
5. 向后兼容性：保持与现有代码的兼容

这个架构将使 OpenViking 能够适应不断增长的内容解析需求，并成为一个强大的内容理解平台。