WeKnora 数据源连接器（Connector）实现指南：从零扩展飞书、Notion 等外部平台同步-育师

WeKnora 数据源连接器（Connector）实现指南：从零扩展飞书、Notion 等外部平台同步

【免费下载链接】WeKnoraOpen-source LLM knowledge platform: turn raw documents into a queryable RAG, an autonomous reasoning agent, and a self-maintaining Wiki.项目地址: https://gitcode.com/GitHub_Trending/we/WeKnora

本篇技术指南围绕 WeKnora 开源 LLM 知识平台的数据源同步框架展开，完整讲解如何为internal/datasource框架新增一个外部平台连接器（Connector），覆盖从包结构搭建、平台类型定义、API 客户端封装、Connector接口实现、容器注册、类型常量与元数据声明、单元测试，到 OAuth / 分页 / 增量同步 / 删除跟踪等通用模式的全部环节。读完本文，你将掌握 WeKnora 数据源框架的扩展契约，并能参照仓库内飞书（Feishu）与 Notion 的真实实现，独立为任意外部文档平台编写、注册并验证一个可用的 Connector。

一、Connector 是什么：适配器模式的落地

在 WeKnora 中，Connector 是把“外部平台的 API 形态”翻译成“WeKnora 数据模型”的适配器。它屏蔽了不同平台之间 API 风格、认证方式、资源组织的差异，让上层的数据源服务（DataSourceService）、同步调度器（Scheduler）和知识库落库流程完全感知不到具体平台的存在。

一个 Connector 需要负责四类核心职责：

连接校验（Connection validation）：验证凭据是否有效、网络是否可达；
资源列举（Resource listing）：列出用户可以选择的文档、空间、文件夹等资源；
全量同步（Full sync）：抓取所选资源下的全部条目；
增量同步（Incremental sync）：只抓取上次同步以来发生变化的条目。

在仓库中，这个契约被抽象为Connector接口，定义在 internal/datasource/connector.go，所有外部数据源连接器（Feishu、Lark、Notion、Yuque、RSS、GitLab、IMA 等）都是它的实现。整个数据源同步框架的架构可以参见 internal/datasource/README.md，其分层为：外部数据源 → Connector 注册表与适配器 → DataSourceService 业务逻辑 → HTTP Handler 与 API 路由（/api/v1/datasource）→ 数据库（data_sources、sync_logs表）。

二、Step 1：创建 Connector 包结构

首先为你的平台类型创建一个独立的 Go 包目录：

mkdir -p internal/datasource/connector/yourtype/

包内约定包含三个文件，职责清晰分离：

client.go—— API 客户端封装（HTTP 调用、认证、重试、分页）；
connector.go—— 实现Connector接口（校验、列举、全量/增量抓取）；
types.go—— 平台特有的数据结构（配置、资源、条目、游标）。

这一约定与仓库现有实现完全一致。例如 internal/datasource/connector/notion/ 目录下就是client.go、connector.go、types.go（另有markdown.go负责 Notion block 到 Markdown 的转换）；飞书则按 Wiki 与云盘两种模式拆成了 internal/datasource/connector/feishu/core/（共享的 Client、Region、导出逻辑）与wiki/、drive/三个子包。

三、Step 2：定义平台类型（types.go）

types.go存放该平台独有的数据结构，典型的四类结构如下：

package yourtype import "time" // Platform-specific configuration type Config struct { BaseURL string `json:"base_url"` APIToken string `json:"api_token"` // Or OAuth fields: AccessToken string `json:"access_token"` RefreshToken string `json:"refresh_token"` ExpiresAt time.Time `json:"expires_at"` } // Platform-specific resource representation type YourResource struct { ID string Name string Type string // "document", "folder", "space", etc. ModifiedAt time.Time URL string } // Platform-specific item representation type YourItem struct { ID string Title string Content string ContentHTML string ModifiedAt time.Time URL string CreatedBy string } // Platform-specific pagination/cursor type YourCursor struct { Offset int `json:"offset,omitempty"` LastModified time.Time `json:"last_modified,omitempty"` PageToken string `json:"page_token,omitempty"` }

源码印证：真实 Connector 的Config结构与上述模板一一对应。例如 Notion 的Config（见 internal/datasource/connector/notion/types.go）只包含一个APIKey string \json:"api_key"`字段，配合parseNotionConfig函数从DataSourceConfig.Credentials中提取并校验api_key（缺失或非空字符串会分别返回datasource.ErrInvalidCredentials包装的错误）。飞书 OAuth 场景则如模板所示，需要存储access_token、refresh_token、expires_at` 三个字段用于令牌生命周期管理。

四、Step 3：实现 API 客户端（client.go）

client.go是平台 API 的薄封装层，负责真实的 HTTP 通信。模板骨架：

package yourtype import ( "context" "fmt" "net/http" "encoding/json" ) type Client struct { baseURL string apiToken string httpClient *http.Client } // NewClient creates a new API client func NewClient(config *Config) *Client { return &Client{ baseURL: config.BaseURL, apiToken: config.APIToken, httpClient: &http.Client{Timeout: 30 * time.Second}, } } // Example methods func (c *Client) GetResources(ctx context.Context) ([]YourResource, error) { // Call platform API // Parse response // Return resources } func (c *Client) GetDocument(ctx context.Context, docID string) (*YourItem, error) { // Fetch single document } func (c *Client) GetDocumentsModifiedSince(ctx context.Context, since time.Time) ([]YourItem, error) { // Fetch documents modified since timestamp }

源码印证：仓库中真实客户端的实现比模板更完善。Notion 的newClient会固定使用NotionAPIVersion = "2026-03-11"与DefaultBaseURL = "https://api.notion.com"（见 internal/datasource/connector/notion/types.go），并在请求头携带Notion-Version与Authorization: Bearer <token>；飞书则在 internal/datasource/connector/feishu/core/client.go 中实现了更复杂的逻辑，并且仓库为它专门编写了重试与错误归类测试（client_retry_test.go、connector_error_reason_test.go），建议新 Connector 在客户端层就做好超时控制、分页循环和错误标准化，避免把平台错误原样透传给上层。

五、Step 4：实现 Connector 接口（connector.go）

这是整个扩展的核心步骤。请以 internal/datasource/connector.go 中定义的接口为准（注意：它比早期文档版本新增了两个方法，详见下文“接口的演进”小节）：

type Connector interface { Type() string Validate(ctx context.Context, config *types.DataSourceConfig) error ListResources(ctx context.Context, config *types.DataSourceConfig, parentID string) ([]types.Resource, error) ResolveResourceAncestors(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]string, error) FetchAll(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]types.FetchedItem, error) FetchIncremental(ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor) ([]types.FetchedItem, *types.SyncCursor, error) }

完整实现模板如下：

package yourtype import ( "context" "fmt" "github.com/Tencent/WeKnora/internal/types" ) type YourConnector struct { client *Client } // NewConnector creates a new connector func NewConnector() *YourConnector { return &YourConnector{} } // Type returns the connector type identifier func (c *YourConnector) Type() string { return types.ConnectorTypeYourType // Must match constant in types/datasource.go } // Validate verifies that the configuration is valid func (c *YourConnector) Validate(ctx context.Context, config *types.DataSourceConfig) error { if config == nil { return fmt.Errorf("config is nil") } // Parse your type-specific config yourConfig := &Config{} if err := parseConfig(config, yourConfig); err != nil { return fmt.Errorf("invalid config: %w", err) } // Create client client := NewClient(yourConfig) // Test connection _, err := client.GetResources(ctx) if err != nil { return fmt.Errorf("connection failed: %w", err) } return nil } // ListResources lists available resources (documents, spaces, folders) func (c *YourConnector) ListResources(ctx context.Context, config *types.DataSourceConfig, parentID string) ([]types.Resource, error) { yourConfig := &Config{} if err := parseConfig(config, yourConfig); err != nil { return nil, err } client := NewClient(yourConfig) yourResources, err := client.GetResources(ctx) if err != nil { return nil, err } // Convert to WeKnora Resource format resources := make([]types.Resource, len(yourResources)) for i, yr := range yourResources { resources[i] = types.Resource{ ExternalID: yr.ID, Name: yr.Name, Type: yr.Type, URL: yr.URL, ModifiedAt: yr.ModifiedAt, } } return resources, nil } // FetchAll performs a full sync func (c *YourConnector) FetchAll(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]types.FetchedItem, error) { yourConfig := &Config{} if err := parseConfig(config, yourConfig); err != nil { return nil, err } client := NewClient(yourConfig) var allItems []types.FetchedItem // Fetch all documents from specified resources for _, resourceID := range resourceIDs { // Get documents from this resource (implementation depends on platform) yourItems, err := client.GetDocumentsFromResource(ctx, resourceID) if err != nil { return nil, fmt.Errorf("failed to fetch resource %s: %w", resourceID, err) } // Convert to FetchedItem format for _, yi := range yourItems { item := types.FetchedItem{ ExternalID: yi.ID, Title: yi.Title, Content: []byte(yi.Content), ContentType: "text/markdown", FileName: fmt.Sprintf("%s.md", yi.Title), URL: yi.URL, UpdatedAt: yi.ModifiedAt, SourceResourceID: resourceID, Metadata: map[string]string{ "created_by": yi.CreatedBy, "platform": "yourtype", }, } allItems = append(allItems, item) } } return allItems, nil } // FetchIncremental performs an incremental sync func (c *YourConnector) FetchIncremental(ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor) ([]types.FetchedItem, *types.SyncCursor, error) { yourConfig := &Config{} if err := parseConfig(config, yourConfig); err != nil { return nil, nil, err } client := NewClient(yourConfig) // Determine start time for incremental fetch var sinceTime time.Time if cursor != nil && !cursor.LastSyncTime.IsZero() { sinceTime = cursor.LastSyncTime } else { sinceTime = time.Now().AddDate(0, 0, -7) // Default: last 7 days } // Fetch changed items yourItems, err := client.GetDocumentsModifiedSince(ctx, sinceTime) if err != nil { return nil, nil, fmt.Errorf("incremental fetch failed: %w", err) } // Convert to FetchedItem format items := make([]types.FetchedItem, len(yourItems)) for i, yi := range yourItems { items[i] = types.FetchedItem{ ExternalID: yi.ID, Title: yi.Title, Content: []byte(yi.Content), ContentType: "text/markdown", FileName: fmt.Sprintf("%s.md", yi.Title), URL: yi.URL, UpdatedAt: yi.ModifiedAt, Metadata: map[string]string{ "created_by": yi.CreatedBy, "platform": "yourtype", }, } } // Create new cursor for next sync nextCursor := &types.SyncCursor{ LastSyncTime: time.Now(), ConnectorCursor: map[string]interface{}{ "last_modified": time.Now(), }, } return items, nextCursor, nil } // Helper function to parse config func parseConfig(config *types.DataSourceConfig, target interface{}) error { data, err := json.Marshal(config.Credentials) if err != nil { return err } return json.Unmarshal(data, target) }

接口的演进：新方法说明

从当前源码看，Connector接口相比早期版本发生了两处演进，新 Connector 必须一并实现：

ListResources增加了parentID参数（懒加载）：parentID == ""时返回顶层资源（如飞书 Wiki 的空间列表）；parentID != ""时只返回该资源的直接子级。层级列举本身就是扁平或一次返回整棵树的 Connector（如 Notion）可以忽略 root 调用时的parentID，并对任何非空parentID返回空切片。
新增ResolveResourceAncestors：用于在懒加载选择器中还原预先存在的深层选择——对每个给定资源 ID，返回其所有祖先的ExternalID（去重、无序）。Notion 这类一次返回全量树、Yuque 这类扁平列表的 Connector 无需此能力，直接返回空切片即可（参见 internal/datasource/connector/notion/connector.go 中的ResolveResourceAncestors实现）。

可选进阶：StreamingConnector

对于文档体量大的平台（如飞书 Wiki 全量同步可能涉及数千节点），仓库还提供了可选接口StreamingConnector（同样定义在 internal/datasource/connector.go）。实现它的 Connector 通过FetchStream(ctx, config, cursor, h)方法配合StreamHandler的Emit（逐条摄入）与Checkpoint（分页边界持久化游标）接口，让服务端把“抓取 → 入库 → 打点”交错执行：大同步可以增量持久化、超时后从检查点续跑，而不是把所有条目驻留内存、重试时全部重来。Checkpoint收到的游标必须是可完整续跑的快照而非增量；未实现该接口的 Connector 自动回退到FetchAll/FetchIncremental不变。

六、Step 5：注册到容器

在 WeKnora 中，所有 Connector 通过**注册表（Registry）**统一管理，而不是直接塞进 dig 容器。推荐方式是在服务初始化段创建注册表并逐个注册：

// In the service initialization section connectorRegistry := datasource.NewConnectorRegistry() connectorRegistry.Register(yourconnector.NewConnector()) connectorRegistry.Register(feishuconnector.NewConnector()) // ... etc container.Provide(func() *datasource.ConnectorRegistry { return connectorRegistry })

源码印证：真实注册逻辑集中在 internal/container/container.go 的initConnectorRegistry()（第 1669 行起）。它创建注册表后依次注册：wiki.NewConnector(core.RegionFeishu)（飞书）、wiki.NewConnector(core.RegionLark)（Lark，飞书国际版，同一 Connector 仅 API host 与租户不同）、drive.NewDriveConnector(core.RegionFeishuDrive)与drive.NewDriveConnector(core.RegionLarkDrive)（飞书/Lark 云盘模式）、notionConnector.NewConnector()、yuqueConnector.NewConnector()、imaConnector.NewConnector()、rssConnector.NewConnector()、gitlabConnector.NewConnector()。值得注意的工程细节：注册错误通过errors.Join聚合，任何一个 Connector 配置错误或类型重复都会让容器初始化响亮地失败，而不是在运行时静默禁用该功能。

注册表本身（ConnectorRegistry）在 internal/datasource/connector.go 中实现：Register对 nil Connector 与空类型分别返回ErrConnectorNil、ErrConnectorTypeEmpty；Get在找不到类型时返回ErrConnectorNotFound；List返回所有已注册类型。相关错误统一定义在 internal/datasource/errors.go。

七、Step 6：添加 Connector 类型常量

在 internal/types/datasource.go 的常量块中追加你的类型标识，与Type()返回值保持一致：

const ( // ... existing types ... ConnectorTypeYourType = "yourtype" )

仓库现有类型常量（见该文件第 17-40 行）包括：feishu、lark、feishu_drive、lark_drive、notion、confluence、yuque、github、google_drive、onedrive、dingtalk、web_crawler、slack、imap、rss、gitlab、ima。同一文件还定义了同步模式（incremental/full）、数据源状态（active/paused/error/deleted）、同步日志状态（running/success/partial/failed/canceled）以及冲突策略（overwrite/skip），这些常量会在上层流程中与你的 Connector 交互，值得一并了解。

八、Step 7：添加元数据（Metadata）

元数据是前端展示 Connector 选项、判断认证方式与能力边界的依据，定义在 internal/datasource/connector.go 的ConnectorMetadataRegistry中：

var ConnectorMetadataRegistry = map[string]ConnectorMetadata{ // ... existing entries ... types.ConnectorTypeYourType: { Type: types.ConnectorTypeYourType, Name: "Your Platform Name", Description: "Sync documents from Your Platform", Priority: X, // Lower number = higher priority in UI AuthType: "oauth2", // or "api_key", "token", "password" Capabilities: []string{"incremental", "webhook", "deletion_sync"}, }, }

ConnectorMetadata结构包含Type、Name、Description、Icon、Priority（UI 排序，数字越小越靠前）、AuthType（oauth2/api_key/token/password/none/custom）、Capabilities（incremental、webhook、deletion_sync、hierarchical等能力标签）。ListAvailableConnectors()会按Priority升序返回全部元数据供前端渲染。仓库现有条目可作参考：飞书与 Lark 使用oauth2且具备incremental, deletion_sync能力；Notion 使用api_key，能力为incremental；GitLab 使用token，能力为incremental, hierarchical；Web Crawler 无需认证（none）；RSS 使用custom。

九、Step 8：编写单元测试

每个新 Connector 都应携带单元测试，覆盖校验、全量抓取与增量抓取三条主路径：

// Example test func TestYourConnectorValidate(t *testing.T) { connector := NewConnector() config := &types.DataSourceConfig{ Type: types.ConnectorTypeYourType, Credentials: map[string]interface{}{ "api_token": "test_token", }, } err := connector.Validate(context.Background(), config) // assert no error } func TestYourConnectorFetchAll(t *testing.T) { connector := NewConnector() config := &types.DataSourceConfig{ Type: types.ConnectorTypeYourType, Credentials: map[string]interface{}{ "api_token": "test_token", }, ResourceIDs: []string{"resource_1"}, } items, err := connector.FetchAll(context.Background(), config, []string{"resource_1"}) // assert results }

源码印证：仓库对测试相当重视。Notion 的测试覆盖client_test.go、connector_test.go、types_test.go、markdown_test.go四个文件；飞书 core 包也包含blocks_test.go、client_retry_test.go、connector_error_reason_test.go、markdown_test.go、region_test.go、tally_test.go等。框架层（internal/datasource/README.md 的“Testing”一节）还特别说明：核心逻辑不依赖外部服务，可以用 Mock Connector 进行测试，数据库操作可隔离与 Mock，便于在 CI 中稳定验证。

十、实施检查清单（Checklist）

在提交前逐项核对，避免遗漏导致注册失败或功能不完整：

创建了internal/datasource/connector/yourtype/包
实现了types.go，含平台数据结构（Config / Resource / Item / Cursor）
实现了client.go，封装平台 API（认证、超时、分页）
实现了connector.go，满足Connector接口全部方法（含parentID与ResolveResourceAncestors）
在internal/types/datasource.go中添加了 Connector 类型常量
在容器初始化（internal/container/container.go的initConnectorRegistry）中注册
在ConnectorMetadataRegistry中添加了元数据条目
补充了单元测试（Validate / FetchAll / FetchIncremental）
使用真实 API 手工联调验证
在文档中记录任何特殊要求（认证前置条件、限流约定、能力限制等）

十一、常见实现模式（Common Patterns）

1. OAuth 流程

OAuth 类平台把令牌存入Config，并在每次请求前确保令牌有效，过期则自动刷新：

type Config struct { AccessToken string RefreshToken string ExpiresAt time.Time } // Refresh tokens when expired func (c *Client) ensureValidToken(ctx context.Context) error { if time.Now().After(c.config.ExpiresAt) { return c.refreshToken(ctx) } return nil }

飞书（oauth2）与 Notion（api_key）代表了两种典型认证形态：前者依赖 OAuth 令牌生命周期管理，后者只需要一个内部集成 Token。

2. 分页（Pagination）

对返回分页结果的平台，封装一个带游标的取页方法，由 Connector 循环直到NextPageToken为空：

func (c *Client) GetDocumentsPage(ctx context.Context, pageToken string) (*Page, error) { // Returns {Items, NextPageToken} }

3. 基于时间戳的增量同步

增量同步通常利用平台的modified_after之类的时间参数：

func (c *Client) GetModifiedSince(ctx context.Context, since time.Time) ([]Item, error) { // Uses API parameter like &modified_after=2026-03-26T10:00:00Z }

增量游标（types.SyncCursor）的持久化由上层负责：DataSource.LastSyncCursor字段（见 internal/types/datasource.go 第 104 行）以 JSONB 存储 Connector 专有状态，同步任务会读取并回传。另外注意FetchIncremental在首次同步（无游标）时的默认回退窗口——模板中为最近 7 天，实际业务可按需调整。

4. 删除跟踪（Deletion Tracking）

支持删除同步的平台，在条目中标记删除状态，让 WeKnora 侧可以联动清理知识库：

type Item struct { IsDeleted bool // Set when item is deleted }

DataSource模型中的SyncDeletions字段（默认true，见 internal/types/datasource.go 第 98 行）控制是否把源端删除同步到知识库，元数据中的deletion_sync能力标签与之一致。同步中部分资源失败时，可返回PartialFetchError（定义在 internal/datasource/errors.go），携带各失败资源的明细，上层会把该次同步标记为partial并保留成功部分的结果与游标。

十二、使用真实 API 联调

单元测试之外，务必用真实凭据走一遍端到端验证：

准备测试凭据（测试账号、受限权限）；
创建一个小型测试资源（例如单篇文档）；
依次调用四个核心方法：

connector := NewConnector() config := &types.DataSourceConfig{...} // Test Validate err := connector.Validate(ctx, config) // Test ListResources resources, err := connector.ListResources(ctx, config, "") // Test FetchAll items, err := connector.FetchAll(ctx, config, []string{resources[0].ExternalID}) // Test FetchIncremental items, cursor, err := connector.FetchIncremental(ctx, config, nil)

联调通过后，还可以通过 REST API 走一遍平台级流程（端点定义见 internal/datasource/README.md）：POST /api/v1/datasource创建数据源 →POST /api/v1/datasource/:id/validate测试连接 →GET /api/v1/datasource/:id/resources列举可选资源 →POST /api/v1/datasource/:id/sync触发同步 →GET /api/v1/datasource/:id/logs查看同步日志。

十三、参考实现：飞书 Connector 与 Notion Connector

飞书（Feishu）：最值得借鉴的第一站

原文档建议以飞书作为第一个参考实现，理由是：飞书 API 文档完善、WeKnora 已有internal/im/feishu/作为模式参考、国内使用场景广泛、且支持 Webhook 实时同步。实际仓库中，飞书 Connector 已经落地并演进出更精细的结构：

internal/datasource/connector/feishu/ ├── core/ (共享核心：Client、Region、block 解析、Markdown 转换) │ ├── client.go │ ├── engine.go │ ├── blocks.go │ ├── markdown.go │ ├── region.go │ ├── shared.go │ ├── types.go │ └── ...(*_test.go) ├── drive/ (飞书/Lark 云盘模式 Connector) └── wiki/ (飞书/Lark Wiki 空间模式 Connector)

其中core/region.go负责区分飞书（国内，feishu.cn）与 Lark（国际，larksuite.com）的 API host 与租户域，这也是为什么同一个飞书 Connector 可以注册出feishu和lark两个类型。文档中提到的两个关键参考文件依然存在：飞书 IM 侧的 internal/im/feishu/adapter.go（Feishu API 调用模式）与 internal/im/feishu/longconn.go（长连接处理）。飞书 Connector 还实现了前面提到的StreamingConnector与ResolveResourceAncestors（Wiki 空间按层级懒加载），是大体量平台实现的标杆。

Notion：扁平化平台的简洁样板

Notion Connector（internal/datasource/connector/notion/connector.go）是另一种极端形态的代表：ListResources通过一次SearchPages拿到带parent_id的完整层级树，前端可直接渲染树形选择器；对非空parentID与ResolveResourceAncestors都返回空结果（因为无需懒加载）；并处理了data_source对象（2025-09-03 后的 Notion API 新增类型）的database_parent归属解析。对比飞书与 Notion 两个实现，可以直观理解parentID懒加载设计在不同平台上的差异化取舍。

结语

从创建包结构到注册进容器，一个 WeKnora 数据源 Connector 的完整生命周期并不复杂，关键在于严格对齐Connector接口契约、补全类型常量与元数据、并针对校验/全量/增量三条主路径编写测试。在此基础上，飞书 Connector 展示了大体量平台的流式同步、层级懒加载与多区域支持，Notion Connector 展示了扁平化平台的最简实现，两者共同构成了新 Connector 的最佳实践模板。若你的目标平台已在internal/types/datasource.go中存在常量（如confluence、github、google_drive、onedrive、dingtalk、slack等）但目前尚未在 internal/container/container.go 的initConnectorRegistry中注册，它们就是社区扩展的直接候选对象。