文档类型：开发指南 适用对象：插件开发者 创建时间：2026-02-22 文档版本：v1.0 当前支持：仅支持数据源插件开发
重要说明：当前版本仅支持数据源插件开发，其他插件类型（AI模型、搜索引擎等）为架构预留，暂未实现。
小遥搜索采用微内核架构，核心系统最小化，扩展功能以插件形式实现：
┌─────────────────────────────────────────────────────────────┐
│ API层 │
└────────────────────────────┬────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────────┐
│ 微内核层 (Core) │
│ • 插件生命周期管理 • 配置管理 • 数据库 • 搜索协调 │
└────────────────────────────┬────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────────┐
│ 插件层 (Plugins) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 数据源插件 │ │ AI模型插件 │ │ 搜索引擎插件 │ │
│ │ 语雀/飞书 │ │ OpenAI/Claude│ │ Faiss/Whoosh│ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘
| 原则 | 说明 |
|---|---|
| 约定优于配置 | 插件放到 data/plugins/ 目录自动发现 |
| 接口隔离 | 通过抽象基类定义接口，插件实现接口 |
| 异步优先 | 遵循 asyncio 异步架构 |
| 故障隔离 | 插件故障不影响核心功能 |
data/plugins/
├── datasource/ # 数据源插件目录
│ └── your-plugin/ # 你的插件
│ ├── plugin.py # 插件实现（必需）
│ ├── config.yaml # 插件配置（必需）
│ ├── data/ # 数据存储目录
│ └── README.md # 说明文档
步骤1：创建插件目录和文件
mkdir -p data/plugins/datasource/myplugin
cd data/plugins/datasource/myplugin
touch plugin.py config.yaml步骤2：编写插件代码（见第3章） 步骤3：启用插件并重启后端
所有数据源插件必须继承 DataSourcePlugin 并实现以下方法：
from abc import ABC, abstractmethod
from typing import Dict, Any
class DataSourcePlugin(ABC):
"""数据源插件基类"""
@classmethod
@abstractmethod
def get_metadata(cls) -> Dict[str, Any]:
"""返回插件元数据"""
pass
@abstractmethod
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化插件"""
pass
@abstractmethod
async def sync(self) -> bool:
"""同步数据到本地（核心方法）"""
pass
@abstractmethod
async def cleanup(self):
"""清理资源"""
pass
def get_file_source_info(self, file_path: str, content: str) -> Dict[str, Any]:
"""获取数据源信息（可选实现）"""
return {"source_type": None, "source_url": None}# config.yaml
plugin:
id: myplugin # 插件唯一标识
name: 我的插件 # 显示名称
version: "1.0.0" # 版本号
type: datasource # 插件类型
enabled: true # 是否启用
datasource:
# 数据源特定配置
api_url: "https://api.example.com"
download_dir: "./data/downloaded"外部数据源
│
│ plugin.sync() 下载文件
▼
本地文件系统 (data/plugins/datasource/myplugin/data/*.md)
│
│ FileScanner 扫描
▼
索引和搜索 (自动处理)
# plugin.py
import asyncio
from pathlib import Path
from typing import Dict, Any
from app.plugins.interface.datasource import DataSourcePlugin
from app.core.logging_config import logger
class MyDataSource(DataSourcePlugin):
"""我的数据源插件"""
def __init__(self):
self._config = {}
self._plugin_dir = None
@classmethod
def get_metadata(cls) -> Dict[str, Any]:
return {
"id": "myplugin",
"name": "我的数据源",
"version": "1.0.0",
"type": "datasource",
"author": "Your Name",
"description": "插件描述"
}
async def initialize(self, config: Dict[str, Any]) -> bool:
"""初始化：准备目录、验证配置"""
try:
self._config = config
self._plugin_dir = Path(__file__).parent
# 创建下载目录
download_dir = self._plugin_dir / "data" / "downloaded"
download_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"插件初始化成功: {download_dir}")
return True
except Exception as e:
logger.error(f"插件初始化失败: {e}")
return False
async def sync(self) -> bool:
"""同步：下载数据到本地"""
try:
download_dir = self._plugin_dir / "data" / "downloaded"
# TODO: 实现你的下载逻辑
# 例如：调用API、下载文件等
# 示例：创建测试文件
test_file = download_dir / "test.md"
test_file.write_text("# 测试文档\n内容...")
logger.info(f"同步完成: {test_file}")
return True
except Exception as e:
logger.error(f"同步失败: {e}")
return False
def get_file_source_info(self, file_path: str, content: str) -> Dict[str, Any]:
"""返回数据源信息（索引时调用）"""
return {
"source_type": "myplugin",
"source_url": None
}
async def cleanup(self):
"""清理资源"""
logger.info("插件清理完成")# config.yaml
plugin:
id: myplugin
name: 我的数据源
version: "1.0.0"
type: datasource
enabled: true
datasource:
# 在这里添加你的配置项
api_url: "https://api.example.com"完整的语雀插件实现使用 yuque-dl CLI 工具：
# 核心同步逻辑
async def sync(self) -> bool:
"""调用 yuque-dl 下载数据"""
cmd = ["npx", "yuque-dl", self.repo_url, "-d", self.download_dir]
process = await asyncio.create_subprocess_exec(*cmd)
await process.communicate()
return process.returncode == 0详见：yuque-dl使用文档
┌─────────────────────────────────────────────────────────────┐
│ 应用启动时 │
├─────────────────────────────────────────────────────────────┤
│ 1. 扫描 data/plugins/ 目录 │
│ 2. 读取各插件的 config.yaml │
│ 3. 加载并初始化插件（调用 initialize()） │
│ 4. 自动执行同步（调用 sync()） │
│ 5. 同步完成后，文件保存到 data/ 目录 │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 构建索引（手动触发） │
├─────────────────────────────────────────────────────────────┤
│ 用户手动调用索引 API，指定插件下载的目录： │
│ │
│ POST /api/index/create │
│ { │
│ "folder_path": "data/plugins/datasource/myplugin/data", │
│ "recursive": true │
│ } │
└─────────────────────────────────────────────────────────────┘
启动后端时，插件加载和同步日志会输出到控制台：
INFO: 开始扫描插件目录: data/plugins
INFO: 加载插件: myplugin
INFO: 插件初始化成功: myplugin
INFO: 自动执行插件同步: myplugin
INFO: 同步完成，共下载 10 个文件
INFO: 插件加载完成，共加载 1 个插件
步骤1：将插件放到正确目录
# 确保插件目录结构正确
data/plugins/datasource/myplugin/
├── plugin.py
├── config.yaml
└── data/步骤2：启动后端，查看加载日志
cd backend
python main.py步骤3：检查同步的文件
# 查看插件下载的文件
ls data/plugins/datasource/myplugin/data/步骤4：手动构建索引
# 使用 curl 或 API 工具
curl -X POST "http://127.0.0.1:8000/api/index/create" \
-H "Content-Type: application/json" \
-d '{"folder_path": "data/plugins/datasource/myplugin/data", "recursive": true}'步骤5：测试搜索
curl -X POST "http://127.0.0.1:8000/api/search" \
-H "Content-Type: application/json" \
-d '{"query": "测试关键词"}'| 问题 | 解决方案 |
|---|---|
| 插件未被加载 | 检查 config.yaml 中 enabled: true |
| 初始化失败 | 查看控制台日志的详细错误信息 |
| 同步无文件 | 检查插件 sync() 方法的实现逻辑 |
| 索引失败 | 确认同步后文件确实存在于 data/ 目录 |
| 搜索无结果 | 确认已构建索引，且搜索关键词匹配文件内容 |
| 技术 | 用途 |
|---|---|
| Python ABC | 插件接口定义 |
| importlib | 插件动态加载 |
| asyncio | 异步处理 |
| YAML | 配置文件格式 |
文档版本：v1.0 创建时间：2026-02-22 维护者：XiaoyaoSearch Team