当前位置：首页 > news >正文

PipesHub AI自定义开发：如何扩展新的数据连接器和AI工具

news 2026/5/5 19:25:08

PipesHub AI自定义开发：如何扩展新的数据连接器和AI工具

【免费下载链接】pipeshub-aiPipesHub is a fully extensible and explainable workplace AI platform for enterprise search and workflow automation项目地址: https://gitcode.com/gh_mirrors/pi/pipeshub-ai

PipesHub AI是一款功能强大的企业级搜索和工作流自动化平台，支持通过自定义开发扩展新的数据连接器和AI工具。本文将详细介绍如何为PipesHub AI平台开发自定义数据连接器，帮助企业轻松集成各类外部服务和AI能力。

数据连接器开发基础

什么是数据连接器

数据连接器是PipesHub AI平台的核心组件，它们作为外部服务（如OneDrive、Gmail、Slack、Jira等）与平台之间的桥梁，负责认证、数据获取、转换和同步。每个连接器都继承自BaseConnector类，并实现标准的同步、内容流处理和清理方法。

PipesHub AI平台登录界面，展示了系统的用户友好设计

连接器主要功能

认证：支持OAuth、API令牌、服务账户等多种认证方式
数据获取：从外部API获取文件、邮件、文档、工单等数据
数据转换：将外部数据转换为标准化的Record模型
权限提取：维护访问控制权限
增量同步：使用增量链接或时间戳实现增量同步
事件发布：触发索引和嵌入生成

开发环境准备

必备知识

✅ Python 3.x（async/await、类型提示）
✅ OAuth 2.0流程
✅ REST API集成
✅ Git & GitHub工作流

环境设置

在开始连接器开发前，需要确保本地开发环境已正确配置：

克隆仓库：

git clone https://gitcode.com/gh_mirrors/pi/pipeshub-ai

安装依赖：请参考项目中的CONTRIBUTING.md文件完成开发环境的设置，确保所有服务都在运行且依赖项已正确安装。
必要工具：
- Python 3.12
- Node.js 22.15
- Docker
- 图形数据库、Kafka、Redis、MongoDB、ETCD Server、Qdrant（通过Docker本地运行）
- VS Code（推荐）

连接器开发步骤

1. 了解代码结构

连接器相关代码主要位于以下目录：

backend/ ├── python/ │ └── app/ │ ├── connectors/ │ │ ├── api/ │ │ │ └── router.py # FastAPI端点 │ │ ├── core/ │ │ │ ├── base/ │ │ │ │ ├── connector/ │ │ │ │ │ └── connector_service.py # BaseConnector类 │ │ │ │ ├── data_processor/ │ │ │ │ │ └── data_source_entities_processor.py │ │ │ │ └── data_store/ │ │ │ │ └── data_store.py │ │ │ ├── factory/ │ │ │ │ └── connector_factory.py # 连接器注册 │ │ │ └── registry/ │ │ │ └── connector_builder.py # @ConnectorBuilder装饰器 │ │ └── sources/ │ │ ├── microsoft/ │ │ │ ├── outlook/ │ │ │ │ └── connector.py # ⭐ 示例连接器 │ │ │ ├── onedrive/ │ │ │ └── common/ │ │ │ └── apps.py # 应用定义 │ │ ├── atlassian/ │ │ │ ├── confluence_cloud/ │ │ │ └── jira_cloud/ │ │ └── YOUR_CONNECTOR/ # 🆕 创建你的连接器 │ │ └── connector.py

2. 创建应用定义

文件路径：backend/python/app/connectors/sources/{vendor}/common/apps.py

应用定义用于标识连接器所属的应用组和名称：

from app.config.constants.arangodb import Connectors from app.connectors.core.interfaces.connector.apps import App, AppGroup class YourConnectorApp(App): """YourConnector的应用定义。""" def __init__(self) -> None: super().__init__( app_name=Connectors.YOUR_CONNECTOR, # 连接器名称 app_group=AppGroup.YOUR_VENDOR # 例如：AppGroup.MICROSOFT_365 )

3. 创建连接器类

文件路径：backend/python/app/connectors/sources/{vendor}/{connector_name}/connector.py

连接器类是实现同步逻辑的核心，需要继承BaseConnector并实现必要的方法：

from app.connectors.core.base.connector.connector_service import BaseConnector from app.connectors.core.registry.connector_builder import ConnectorBuilder @ConnectorBuilder("YourConnector") .in_group("Your Vendor") # 例如："Microsoft 365", "Atlassian" .with_auth_type("OAUTH_ADMIN_CONSENT") # 或 "OAUTH", "API_TOKEN" .with_description("从YourConnector同步数据") .with_categories(["Category1", "Category2"]) # 例如：["Email"], ["Storage"] .configure(lambda builder: builder .with_icon("/assets/icons/connectors/yourconnector.svg") # 添加认证字段和配置 ) .build_decorator() class YourConnector(BaseConnector): """用于从YourService同步数据的连接器。""" async def init(self) -> bool: """使用凭据和API客户端初始化连接器。""" # 实现初始化逻辑 return True async def run_sync(self) -> None: """运行完全同步。""" # 实现同步逻辑 async def run_incremental_sync(self) -> None: """使用增量链接运行增量同步。""" # 实现增量同步逻辑 async def stream_record(self, record: Record) -> StreamingResponse: """流记录内容（下载文件/邮件/文档）。""" # 实现内容流处理

4. 实现核心同步逻辑

连接器的核心功能是同步外部数据，主要包括：

获取外部数据：调用外部API获取数据
数据转换：将外部数据转换为平台的Record模型
权限提取：从外部数据中提取访问权限信息
数据存储：将处理后的数据存储到图数据库
同步点管理：维护同步状态，支持增量同步

详细实现可参考CONNECTOR_INTEGRATION_PLAYBOOK.md中的示例代码。

5. 注册连接器

文件路径：backend/python/app/connectors/core/factory/connector_factory.py

将新开发的连接器注册到工厂类中，以便系统能够识别和使用：

from app.connectors.sources.your_vendor.your_connector.connector import YourConnector class ConnectorFactory: _connectors = { # ... 现有连接器 Connectors.YOUR_CONNECTOR: YourConnector, } # ...