本指南面向希望为 RAGFlow 扩展数据源能力的社区开发者，旨在以专业、可复用的流程说明如何实现并接入新的 connector。RAGFlow 的 connector 框架深受 Onyx 开源项目启发，特此致谢。
在实际使用中，除了从本地文件系统导入文件，RAGFlow 还需要从大量异构系统中获取数据。
为此，引入了统一的 datasource / connector 组件：每一种外部数据源通过一个 connector 完成“认证、拉取文档、增量更新以及按 checkpoint 续传” 等一系列动作，从而让上层检索与问答逻辑完全解耦于底层数据源。
围绕这一目标，RAGFlow 暴露了四个主要接口，覆盖了从简单到复杂的大部分场景：
- LoadConnector：定义全量同步（full sync，load_from_state）的行为，用于重建或恢复完整数据视图。
- PollConnector：定义增量同步（incremental sync，poll_source）的行为，用于在已有数据基础上只拉取变化。
- CheckpointedConnector：在需要分页或“断点续传”时，通过 load_from_checkpoint + ConnectorCheckpoint 显式管理游标与进度。
- CheckpointedConnectorWithPermSync：在需要同时考虑内容与权限（permission）时，在 checkpoint 流中附带权限与失败信息，支持更精细的同步与重试策略。
对于“结构简单”的数据源（例如对象存储、单一 API 拉取），通常只需实现 LoadConnector 和 PollConnector 即可完成接入；
对于需要分页、断点续扫或权限同步的“复杂数据源”（例如 Confluence，Jira，Google Drive），可以进一步实现 CheckpointedConnector 或 CheckpointedConnectorWithPermSync。
只要按照本文说明实现并接好这四类接口，就可以将你的数据源无缝接入 RAGFlow。
本文共有五个部分：体系概览、核心抽象接口、实现流程、接入 SyncBase 的示例解析，以及一个交付检查清单。
体系概览
1. 目录结构：各类数据源代码位于 common/data_source/，复杂源（如 Jira）会拆分为子包。公共工具位于 common/data_source/utils.py。 2. 调度入口：rag/svr/sync_data_source.py 通过 Trio 协程调度同步任务，并根据 FileSource 选择具体实现。 3. 常量定义：common/constants.py 中的 FileSource 与 common/data_source/config.py 中的 DocumentSource 分别决定任务标识和文档来源标签。新增数据源时必须同步更新。
整体结构可以抽象为三层：
- Sync 层：rag/svr/sync_data_source.py 中的 SyncBase 及其子类（例如 S3、Jira），负责根据任务配置决定何时调用 load_from_state / poll_source / load_from_checkpoint，并将批次写入知识库。
- Connector 实现层：common/data_source/ 下具体的 connector 类（例如 BlobStorageConnector、JiraConnector），实现 LoadConnector、PollConnector 和/或 CheckpointedConnector(WithPermSync) 接口，专注于“如何从外部系统取数并组装为 Document”。
- 基础服务层：SyncLogsService 与 KnowledgebaseService，分别负责同步日志统计与文档入库。
从贡献者视角，可以简单地理解为：
- 在 connector 层实现好“如何产生 Document”（含凭证、分页、checkpoint 等细节）；
- 在 Sync 层实现好“何时调用 connector、如何把批次交给基础服务”，并按照第 3 节的流程将新 connector 挂接进整个体系。
SyncBase 是调度流程的核心。所有数据源的执行逻辑都会在 __call__ 中被统一处理。一般不需要被修改。
SyncBase 负责统一的批量写入、日志与 checkpoint 更新，而_generate() 由各数据源实现，负责返回 Iterable[list[Document]]。
class SyncBase:
SOURCE_NAME: str = None
async def __call__(self, task: dict):
...
async def _generate(self, task: dict):
raise NotImplementedError核心抽象接口
文档模型
所有 connector 必须产出 Document。doc_updated_at 必须是 UTC 时间，以保证增量同步精度。
class Document(BaseModel):
id: str
source: str
semantic_identifier: str
extension: str
blob: bytes
doc_updated_at: datetime
size_bytes: int- id：在同一数据源内必须唯一，并且在多次同步中尽量保持不变。常见形式是 "source_type:business_key"。
- semantic_identifier：尽量包含标题、作者或位置等关键信息，方便检索。
同步接口：全量同步（Load）与增量同步（Poll）
- 全量同步（Load，load_from_state）：
- 用于重建当前 connector 下的“完整视图”，通常在任务 reindex == "1" 或首次同步时触发。实现上，一般会从“时间起点”（如 1970-01-01）扫到当前时间，将符合条件的所有文档分批 yield list[Document]。在这一模式下，connector 不需要关心上一次同步到哪里，只需要确保能完整遍历外部系统中应被索引的对象。
class LoadConnector(ABC):
@abstractmethod
def load_credentials(self, credentials: Dict[str, Any]) -> Dict[str, Any] | None: ...
@abstractmethod
def load_from_state(self) -> Generator[list[Document], None, None]:
"""load all documents up to now"""
...
@abstractmethod
def validate_connector_settings(self) -> None: ...- 增量同步（Poll，poll_source）：
- 用于在已有数据基础上“只拉变化”，避免重复处理历史内容。系统会为每次任务提供一个时间窗口 (start, end]（例如从上一次成功同步的时间到现在），connector 应当只返回在该窗口内新增或更新过的文档。实现时，应严格遵守 (start, end] 条件，并依赖 Document.doc_updated_at 的准确性，防止遗漏或重复。
class PollConnector(ABC):
@abstractmethod
def poll_source(self, start: SecondsSinceUnixEpoch, end: SecondsSinceUnixEpoch) -> Generator[list[Document], None, None]:
"""load documents from start to end"""
...下面的代码展示了 S3 对应的具体实现：
class S3(SyncBase):
SOURCE_NAME: str = FileSource.S3
asyncdef_generate(self, task: dict):
self.connector = BlobStorageConnector(
bucket_type=self.conf.get("bucket_type", "s3"),
bucket_name=self.conf["bucket_name"],
prefix=self.conf.get("prefix", "")
)
self.connector.load_credentials(self.conf["credentials"])
document_batch_generator = (
self.connector.load_from_state()
if task["reindex"] == "1"ornot task["poll_range_start"]
elseself.connector.poll_source(
task["poll_range_start"].timestamp(),
datetime.now(timezone.utc).timestamp()
)
)
return document_batch_generator对应的 connector 的实现：
class BlobStorageConnector(LoadConnector, PollConnector):
defload_from_state(self) -> GenerateDocumentsOutput:
returnself._yield_blob_objects(
start=datetime(1970, 1, 1, tzinfo=timezone.utc),
end=datetime.now(timezone.utc),
)
defpoll_source(self, start: SecondsSinceUnixEpoch, end: SecondsSinceUnixEpoch) -> GenerateDocumentsOutput:
start_datetime = datetime.fromtimestamp(start, tz=timezone.utc)
end_datetime = datetime.fromtimestamp(end, tz=timezone.utc)
for batch inself._yield_blob_objects(start_datetime, end_datetime):
yield batch
defvalidate_connector_settings(self) -> None:
...可以看到：
- 调度层（S3._generate）只关心“全量 vs 增量”的切换。
- connector 层通过统一的 _yield_blob_objects 分别实现“从起点到当前”的一次性加载与“窗口内”的滚动拉取。
Checkpoint 支持
对于只能分页读取或需要“断点续扫”的系统，推荐使用 checkpoint 抽象来管理游标状态。这里的 checkpoint 可以理解为“本轮同步结束时的游标快照”，通常包含：
- 当前分页位置或 offset（例如某一页的起始索引）
- 已处理 ID 集合或光标（cursor），用来避免重复拉取
- 是否还有剩余数据的标记（如 has_more）
典型的使用场景包括：
- 外部 API 只支持分页访问，且每页大小有限（如 Jira、Confluence、Google Drive 等）
- 同步过程可能被中断，需要从上一次完成的位置继续，而不是从头重跑
- 希望按较小的批次反复调用 connector，每次只处理一部分数据
在这类场景下，每次调用 load_from_checkpoint 都会：
1. 基于传入的 checkpoint（上一轮的游标快照）决定本轮要读取的数据范围 2. 逐个返回 Document 或 ConnectorFailure 3. 在生成器结束时返回一个“新的 checkpoint”，供下一轮调用使用
当前在 RAGFlow 中主要有两种典型用法，也对应了“简单 connector”与“复杂 connector”的典型划分：
相对简单的CheckpointedConnector
以 Confluence 为例
对于“相对简单”的 connector（只需要按时间和分页遍历内容，不关心权限，也不需要把复杂失败信息编码进 checkpoint），通常实现 CheckpointedConnector 或甚至只实现 LoadConnector / PollConnector 即可，例如 Confluence 内容拉取或 S3 这类存储源。这类 connector 的关注点是“把所有需要索引的对象可靠地遍历一遍”。
接口定义如下：
class CheckpointedConnector(BaseConnector[CT]):
@abc.abstractmethod
defload_from_checkpoint(
self,
start: SecondsSinceUnixEpoch,
end: SecondsSinceUnixEpoch,
checkpoint: CT,
) -> CheckpointOutput[CT]:
...
@abc.abstractmethod
defbuild_dummy_checkpoint(self) -> CT:
...
@abc.abstractmethod
defvalidate_checkpoint_json(self, checkpoint_json: str) -> CT:
...Confluence 的实现只关注“遍历内容”，不在 checkpoint 中携带权限信息：
class ConfluenceCheckpoint(ConnectorCheckpoint):
next_page_url: str | None
classConfluenceConnector(
CheckpointedConnector[ConfluenceCheckpoint],
SlimConnector,
SlimConnectorWithPermSync,
CredentialsConnector,
):
defload_from_checkpoint(
self,
start: SecondsSinceUnixEpoch,
end: SecondsSinceUnixEpoch,
checkpoint: ConfluenceCheckpoint,
) -> CheckpointOutput[ConfluenceCheckpoint]:
end += ONE_DAY # handle time zone weirdness
try:
returnself._fetch_document_batches(checkpoint, start, end)
except Exception as e:
...
defbuild_dummy_checkpoint(self) -> ConfluenceCheckpoint:
return ConfluenceCheckpoint(has_more=True, next_page_url=None)
defvalidate_checkpoint_json(self, checkpoint_json: str) -> ConfluenceCheckpoint:
return ConfluenceCheckpoint.model_validate_json(checkpoint_json)配合工具函数可以一次性加载或增量加载文档：
for doc in load_all_docs_from_checkpoint_connector(
connector=confluence_connector,
start=start,
end=end,
):
print(doc)相对复杂的CheckpointedConnectorWithPermSync
以 Jira 为例
对于“相对复杂”的 connector（需要结合权限、失败记录、外部系统特有游标等信息），更推荐实现 CheckpointedConnectorWithPermSync，并在 checkpoint 中显式记录分页游标、剩余状态等，例如 Jira，Google Drive，Slack，Teams。这类 connector 的关注点除了内容本身，还包括“谁能看到什么”和“哪些对象在某一轮中失败了，需要后续重试”。
接口如下:
class CheckpointedConnectorWithPermSync(ABC):
@abstractmethod
defload_from_checkpoint(
self,
start: SecondsSinceUnixEpoch,
end: SecondsSinceUnixEpoch,
checkpoint: ConnectorCheckpoint,
) -> Generator[Document | ConnectorFailure, None, ConnectorCheckpoint]:
...
@abstractmethod
defload_from_checkpoint_with_perm_sync(
self,
start: SecondsSinceUnixEpoch,
end: SecondsSinceUnixEpoch,
checkpoint: ConnectorCheckpoint,
) -> Generator[Document | ConnectorFailure, None, ConnectorCheckpoint]:
...
@abstractmethod
defbuild_dummy_checkpoint(self) -> ConnectorCheckpoint:
...
@abstractmethod
defvalidate_checkpoint_json(self, checkpoint_json: str) -> ConnectorCheckpoint:
...Jira 同时需要内容与权限/元数据同步，因此实现了带权限的 checkpoint 接口，并使用专门的 checkpoint 类型来跟踪分页状态：
class JiraCheckpoint(ConnectorCheckpoint):
"""Checkpoint that tracks which slice of the current JQL result set was emitted."""
start_at: int = 0
cursor: str | None = None
ids_done: bool = False
all_issue_ids: list[list[str]]
class JiraConnector(CheckpointedConnectorWithPermSync):
...在调度侧，rag/svr/sync_data_source.py 显式编写 checkpoint 循环，利用 CheckpointOutputWrapper 统一处理成功与失败：
def document_batches():
checkpoint = self.connector.build_dummy_checkpoint()
pending_docs = []
...
while checkpoint.has_more:
wrapper = CheckpointOutputWrapper()
generator = wrapper(
self.connector.load_from_checkpoint(
start_time,
end_time,
checkpoint,
)
)
for document, failure, next_checkpoint in generator:
if failure isnotNone:
continue
if document isnotNone:
pending_docs.append(document)
iflen(pending_docs) >= batch_size:
yield pending_docs
pending_docs = []
if next_checkpoint isnotNone:
checkpoint = next_checkpoint
...
if pending_docs:
yield pending_docs对贡献者的建议：
- 仅内容同步时，可参考 Confluence，实现 CheckpointedConnector 并复用 load_all_docs_from_checkpoint_connector。
参考https://github.com/infiniflow/ragflow/issues/11376 - 内容与权限同步或复杂错误策略时，可参考 Jira，实现 CheckpointedConnectorWithPermSync 并在 sync_data_source.py 中编写借助CheckpointOutputWrapper显式编写 checkpoint 循环。
参考https://github.com/infiniflow/ragflow/blob/bd4bc57009fe2990b3be1000564a4d5559477cfc/rag/svr/sync_data_source.py#L351
实现流程
最小交付要求如下：
通过以上六步即可得到一个结构完整、可被调度器识别的最小实现。
接入 SyncBase 示例
以下示例展示了如何在 SyncBase._generate 中接入不同类型的 connector，并从调度逻辑到批量产出 Document 的完整链路，可作为实现接入步骤的参考。
S3 Blob Storage 示例
SyncBase 中的调度逻辑
class S3(SyncBase):
SOURCE_NAME: str = FileSource.S3
asyncdef_generate(self, task: dict):
self.connector = BlobStorageConnector(
bucket_type=self.conf.get("bucket_type", "s3"),
bucket_name=self.conf["bucket_name"],
prefix=self.conf.get("prefix", "")
)
self.connector.load_credentials(self.conf["credentials"])
document_batch_generator = (
self.connector.load_from_state()
if task["reindex"] == "1"ornot task["poll_range_start"]
elseself.connector.poll_source(
task["poll_range_start"].timestamp(),
datetime.now(timezone.utc).timestamp()
)
)
begin_info = "totally"if task["reindex"] == "1"ornot task["poll_range_start"] elsef"from {task['poll_range_start']}"
logging.info(
f"Connect to S3: {self.conf['bucket_name']}/"
f"{self.conf.get('prefix', '')} {begin_info}"
)
return document_batch_generatorConnector 逻辑
class BlobStorageConnector(LoadConnector, PollConnector):
def__init__(self, bucket_type: str, bucket_name: str, prefix: str = "", batch_size: int = INDEX_BATCH_SIZE, european_residency: bool = False) -> None:
self.bucket_type: BlobType = BlobType(bucket_type)
self.bucket_name = bucket_name.strip()
self.prefix = prefix ifnot prefix or prefix.endswith("/") else prefix + "/"
self.batch_size = batch_size
self.s3_client: Optional[Any] = None
defload_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None:
ifself.bucket_type == BlobType.S3:
authentication_method = credentials.get("authentication_method", "access_key")
if authentication_method == "access_key":
ifnotall(credentials.get(key) for key in ["aws_access_key_id", "aws_secret_access_key"]):
raise ConnectorMissingCredentialError("Amazon S3")
elif authentication_method == "iam_role":
ifnot credentials.get("aws_role_arn"):
raise ConnectorMissingCredentialError("Amazon S3 IAM role ARN is required")
...
self.s3_client = create_s3_client(self.bucket_type, credentials, self.european_residency)
returnNone
def_yield_blob_objects(self, start: datetime, end: datetime) -> GenerateDocumentsOutput:
paginator = self.s3_client.get_paginator("list_objects_v2")
pages = paginator.paginate(Bucket=self.bucket_name, Prefix=self.prefix)
batch: list[Document] = []
for page in pages:
for obj in page.get("Contents", []):
...
file_name = os.path.basename(obj["Key"])
blob = download_object(self.s3_client, self.bucket_name, obj["Key"], self.size_threshold)
if blob isNone:
continue
batch.append(Document(
id=f"{self.bucket_type}:{self.bucket_name}:{obj['Key']}",
blob=blob,
source=DocumentSource(self.bucket_type.value),
semantic_identifier=file_name,
extension=get_file_ext(file_name),
doc_updated_at=last_modified,
size_bytes=extract_size_bytes(obj) or0
))
iflen(batch) == self.batch_size:
yield batch
if batch:
yield batch实现要点
1. SyncBase 层只负责根据任务配置在“全量/增量”之间切换，实际抓取与分页逻辑全部封装在 connector 内。 2. Document.id 使用 bucket_type:bucket:key 组合，天然避免重复。 3. _yield_blob_objects 同时服务于 load_from_state 与 poll_source，可确保两种模式行为一致。
交付前检查清单
1. FileSource 与 DocumentSource 已添加对应条目。 2. load_credentials() 校验了所有必需字段，如涉及 OAuth 亦支持凭证刷新与持久化。 3. load_from_state() 与 poll_source() 在适用时共享批处理实现，或在行为不完全一致时有清晰的设计理由和文档说明。 4. 生成的 Document 填写了 id、semantic_identifier、extension、doc_updated_at（UTC）及 size_bytes。 5. SyncBase 子类根据 task["reindex"] 正确切换同步模式，并输出明确日志。 6. 涉及 checkpoint/分页的实现具备终止条件或迭代上限，避免无限循环。 7. 已完成至少一次全量同步与一次增量同步的本地验证，并在 PR 中注明参考 Onyx 的实现要点。
完成上述检查后，即可提交 PR。期待你的贡献帮助 RAGFlow 覆盖更多企业数据源，与社区共同构建稳健的知识底座。
