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.gotypes.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.goclient.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、ErrConnectorTypeEmptyGet在找不到类型时返回ErrConnectorNotFoundList返回所有已注册类型。相关错误统一定义在 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、PriorityUI 排序数字越小越靠前、AuthTypeoauth2/api_key/token/password/none/custom、Capabilitiesincremental、webhook、deletion_sync、hierarchical等能力标签。ListAvailableConnectors()会按Priority升序返回全部元数据供前端渲染。仓库现有条目可作参考飞书与 Lark 使用oauth2且具备incremental, deletion_sync能力Notion 使用api_key能力为incrementalGitLab 使用token能力为incremental, hierarchicalWeb Crawler 无需认证noneRSS 使用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 Patterns1. 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与 Notionapi_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_after2026-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.mdPOST /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.goFeishu API 调用模式与 internal/im/feishu/longconn.go长连接处理。飞书 Connector 还实现了前面提到的StreamingConnector与ResolveResourceAncestorsWiki 空间按层级懒加载是大体量平台实现的标杆。Notion扁平化平台的简洁样板Notion Connectorinternal/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中注册它们就是社区扩展的直接候选对象。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考