Pathway RAG 模板 YAML 数据源（Data Sources）接入完整指南：$sources 与五种来源配置详解

本篇文章是 Pathway Live Data Framework 模板 YAML 片段 中 Data Sources 章节的深度展开。你将学会在 Pathway 的 YAML 模板中，用 $sources 变量把文件系统、SharePoint、Google Drive、S3（含 CSV 等本地格式）等来源的数据接入 RAG 流水线，理解"表必须含 bytes 类型的 data 列"这一约束的来龙去脉，并掌握每个连接器的全部必填/可选参数及其底层实现依据。

数据源在 RAG 模板中的定位：为什么是 $sources 与 data 列

在 Pathway 的 YAML 模板体系里，数据源配置不是孤立存在的，它服务的下游组件是 DocumentStore：文档索引器负责对原始文档做解析（parser）、切分（splitter）和向量化索引（retriever_factory）。DocumentStore 的 docs 参数接收的是一组 Pathway 表，而这些表正是由 $sources 变量定义的连接器列表。

于是有一条贯穿全文的硬性约定：

由于数据源通常被放进 DocumentStore 使用，由此产生的表必须包含一个 bytes 类型的 data 列。索引流水线只会把该列视为"一个完整文件的内容"。

这解释了为什么下文所有 RAG 场景的连接器示例都显式开启 format: binary（或由连接器天然返回二进制）——二进制读取保证"一行 = 一个文件、该文件的全部字节放入 data 列"，这样 parser 才能拿到原始字节流去做后续解析。

在 YAML 中，数据源一般定义在名为 $sources 的参数下（注意这个 $，它是 Pathway 模板 YAML 的变量前缀，会被其他组件通过 $sources 引用），值是一个连接器列表：

$sources:
  - !pw.io.fs.read
    path: data
    format: binary
    with_metadata: true
  - !pw.io.csv.read
    path: csv_files
    with_metadata: false

列表里的每一项都是一个带 ! 标签的映射。根据 YAML 配置语法，标签用于把映射解析为对 Python 对象的调用：!pw.io.fs.read 等价于调用 pw.io.fs.read(path=..., format=..., with_metadata=...) 得到一张输入表；两个连接器组成一个表列表，随后整体赋给 $sources。对每个连接器，你都需要按需指定其全部必要参数；完整连接器清单与参数语义参见仓库中的连接器文档 Live Data Framework Connectors。

变量与环境变量的注入：解读示例中的 $SHAREPOINT_URL、$path

在下文各个示例中，你会看到两类 $ 开头的标识符，含义截然不同，理解这一点是正确复制配置的前提：

• $sources、$bucket 这类普通标识符是 YAML 内部变量：在文件中先定义、供后续组件引用，例如 docs: $sources。变量名通常用小写驼峰/小写加下划线。
• $SHAREPOINT_URL、$SHAREPOINT_TENANT、$DRIVE_ID、$s3_access_key 这类仅由大写字母与下划线组成的标识符是环境变量：Pathway 会从运行环境取值，其值如果符合 YAML 语法中的整数、浮点数或布尔值会被自动解析成对应类型，否则按字符串返回。

官方推荐的实践是：把 path、region 等示例中的占位变量保留为 YAML 变量，而把密钥类参数直接指向环境变量，避免把凭据硬编码进配置文件——例如 access_key: $s3_access_key。关于变量与 $ 引用机制更完整的规则（含 YAML 定义优先于同名环境变量等细节），见 YAML 变量章节。

文件系统（File System）：RAG 最常用的本地来源

文件系统连接器 pw.io.fs.read 支持 plaintext、CSV、JSON 等多种基础格式；但在 Document Store 场景下，数据必须以 binary 格式读取。从 python/pathway/io/fs/init.py 的源码签名看，read 的 format 参数支持 "csv"、"json"、"plaintext"、"plaintext_by_file"、"binary"、"only_metadata" 六种取值：

• "binary"：文件以原始字节读入，不做 UTF-8 解析，每个文件对应一行，内容存进单一 data 列；
• "plaintext_by_file"：每个文件同样占一行，但内容以文本解析；
• "plaintext"：每个文件被拆成若干行文本；
• "csv" / "json"：结构化解析，适合走连接器而非 Document Store 的通用 ETL 场景；
• "only_metadata"：只读元数据、不读文件正文（见下）。

开启 with_metadata: true 时，连接器会给表追加一个名为 _metadata 的 JSON 列（fs 连接器的该字段可能包含 created_at 等文件级信息），format: "binary" 时全表即为"一文件一行、data 列存字节、_metadata 列存元数据"的结构。对应当前数据源文档给出的最小可运行示例：

$sources:
  - !pw.io.fs.read
    path: data                   # Path to the data directory
    format: binary               # Format of the data to be read
    with_metadata: true          # Include metadata in the data

path 指向一个目录时，连接器会持续监听该目录（流式模式），新放入的文件会自动进入索引流水线——这正是"Live Data"的体现。若把 with_metadata 改为 false，则表只含 data 一列，适合对元数据无需求的精简索引。

SharePoint：从企业文档站点直接读取

如果你的文档托管在 SharePoint 站点，可以使用 Pathway 的 SharePoint 连接器 pw.xpacks.connectors.sharepoint.read。需要特别提醒许可前提：该连接器仅在 Pathway Live Data Framework 的 Scale 与 Enterprise 许可下可用（许可对照见 licensing guide）。从源码签名可确认其必填参数为 url、tenant、client_id、cert_path、thumbprint、root_path，并含默认值 30 秒的 refresh_interval。

它会以二进制格式返回单列 data 的表（每个文件占一行）。连接器基于 SharePoint Graph API 做增量同步，每隔 refresh_interval 秒轮询一次站点以发现新增、更新与删除（源码中对应"Failed to get snapshot diff... Retrying"的日志与 time.sleep(self._refresh_interval) 重试循环）。典型 YAML：

$sources:
  - !pw.xpacks.connectors.sharepoint.read
    url: $SHAREPOINT_URL               # URL of the SharePoint site
    tenant: $SHAREPOINT_TENANT         # Tenant ID for SharePoint
    client_id: $SHAREPOINT_CLIENT_ID   # Client ID for authentication
    cert_path: sharepointcert.pem      # Path to the certificate file
    thumbprint: $SHAREPOINT_THUMBPRINT # Thumbprint of the certificate
    root_path: $SHAREPOINT_ROOT        # Root path in SharePoint
    with_metadata: true                # Include metadata in the data
    refresh_interval: 30               # Interval to refresh data (in seconds)

参数说明（对应源码 docstring）：tenant 通常是 SharePoint 应用的 GUID；client_id 是拥有所需授权的 SharePoint 应用 ID；cert_path 是加入该应用的证书路径（通常为 .pem 文件）；thumbprint 是该证书的指纹；root_path 是 SharePoint 空间中要同步的目录或文件路径，例如 Shared Documents/Data。把 tenant、client_id 等敏感值写成全大写环境变量占位符，运行时由环境注入。

Google Drive：借助服务账号接入云端目录

要使用 Google Drive 连接器 pw.io.gdrive.read，需要先完成前置准备：在 Google Cloud 创建项目、启用 Google Drive API、建立用于访问的服务账号，并把生成的凭据 JSON 文件（示例中的 gdrive_indexer.json）放到可访问路径。具体开通步骤参见 docs/2.developers 下的 Google Drive 连接器说明。该连接器同样返回单列 data（二进制）的表。

对照 python/pathway/io/gdrive/init.py 的源码，read 支持以下筛选与同步参数：

• object_id：Google Drive 中被监听的目录/文件 ID（可视为要读取的 Drive 空间根）；
• service_user_credentials_file：服务账号凭据 JSON 的路径；
• file_name_pattern：文件名过滤模式（单个 glob 字符串或 glob 列表，如 "*.pdf"、"*.pptx"），源码中用 fnmatch 逐个匹配跳过不命中的文件；
• object_size_limit：单文件大小上限（字节），超出则跳过（None 表示不限）；源码在拉取文件列表后按 size > limit 直接过滤；
• with_metadata：是否附加元数据列；
• refresh_interval：对云端目录的轮询间隔（秒），源码中轮询失败会打印日志并等待该间隔后重试。

数据源文档给出的最小示例：

$sources:
  - !pw.io.gdrive.read
    object_id: $DRIVE_ID
    service_user_credentials_file: gdrive_indexer.json
    file_name_pattern:
      - "*.pdf"
      - "*.pptx"
    object_size_limit: null
    with_metadata: true
    refresh_interval: 30

S3：面向对象存储的二进制文档源

针对存放在 S3（含兼容 S3 的对象存储）中的数据，使用 pw.io.s3.read。RAG 场景下需要把 format 显式配置为 "binary"，连接器于是返回单列 data（每行一个文件的原始字节）的表。

S3 连接的鉴权配置通过内嵌的 AwsS3Settings 完成（源码中该设置类的字段即 bucket_name、region、access_key、secret_access_key，并有面向特定云厂商的派生设置类）。模板中的最小示例：

$sources:
  - !pw.io.s3.read
    path: $path
    format: "binary"
    aws_s3_setting: !pw.io.s3.AwsS3Settings
      bucket_name: $bucket
      region: "eu-west-3"
      access_key: $s3_access_key
      secret_access_key: $s3_secret_access_key

其中 $path 是桶内对象前缀，$bucket 由你在 YAML 文件顶部定义（如 $bucket: "my-documents-bucket"），access_key / secret_access_key 建议交给环境变量注入，region 则按桶实际所在区域填写（示例为 eu-west-3）。

四种来源参数速查对比

连接器	入口 Tag	返回结构	关键参数	轮询/同步方式	前置条件
文件系统	!pw.io.fs.read	binary 时单列 data（+可选 _metadata）	path、format、with_metadata	流式监听目录	无
SharePoint	!pw.xpacks.connectors.sharepoint.read	单列 data	url、tenant、client_id、cert_path、thumbprint、root_path、refresh_interval	refresh_interval 秒轮询 Graph API	Scale/Enterprise 许可、应用证书
Google Drive	!pw.io.gdrive.read	单列 data	object_id、service_user_credentials_file、file_name_pattern、object_size_limit、with_metadata、refresh_interval	refresh_interval 秒轮询	Google Cloud 项目 + 服务账号
S3	!pw.io.s3.read	单列 data（需 format: binary）	path、format、aws_s3_setting	对象存储增量监听	桶访问凭据

共性要点有二：其一，接入 Document Store 时务必保证产出表含 bytes 类型的 data 列（本地 fs 用 format: binary，SharePoint/Drive 天然二进制，S3 显式声明 binary）；其二，本地文件系统与云端来源的差别主要在"同步语义"上——文件系统目录是流式监听，云端来源则按 refresh_interval 周期性扫描新文件/变更。

组装成完整模板：从 $sources 到 DocumentStore

下面把数据源片段接入一个可运行的 RAG 模板主体（串联 LLM、embedder、splitter、parser、retriever_factory、DocumentStore 与 question_answerer），以便看清 $sources 在整条链中的引用关系；完整可直接运行的 YAML 可对照 Full Pipelines 示例 与 RAG 配置示例：

# 1) 数据源：把本地目录与 Google Drive 文档同时喂给索引器
$sources:
  - !pw.io.fs.read
    path: data
    format: binary
    with_metadata: true

  - !pw.io.gdrive.read
    object_id: $DRIVE_ID
    service_user_credentials_file: gdrive_indexer.json
    file_name_pattern:
      - "*.pdf"
      - "*.pptx"
    object_size_limit: null
    with_metadata: true
    refresh_interval: 30

# 2) LLM 与向量化组件
$llm: !pw.xpacks.llm.llms.OpenAIChat
  model: "gpt-4o-mini"
  retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy
    max_retries: 6
  cache_strategy: !pw.udfs.DefaultCache {}
  temperature: 0
  capacity: 8

$embedder: !pw.xpacks.llm.embedders.OpenAIEmbedder
  model: "text-embedding-3-small"

# 3) 文档预处理
$splitter: !pw.xpacks.llm.splitters.TokenCountSplitter
  max_tokens: 400

$parser: !pw.xpacks.llm.parsers.UnstructuredParser

# 4) 向量索引
$retriever_factory: !pw.stdlib.indexing.BruteForceKnnFactory
  reserved_space: 1000
  embedder: $embedder
  metric: !pw.stdlib.indexing.BruteForceKnnMetricKind.COS

# 5) 关键一步：DocumentStore 通过 docs: $sources 消费数据源表
$document_store: !pw.xpacks.llm.document_store.DocumentStore
  docs: $sources
  parser: $parser
  splitter: $splitter
  retriever_factory: $retriever_factory

# 6) 对外暴露问答入口
question_answerer: !pw.xpacks.llm.question_answering.BaseRAGQuestionAnswerer
  llm: $llm
  indexer: $document_store

注意第 5 步：docs: $sources 直接把第一步定义的多源列表作为文档输入——这正是文档强调"数据源表必须含 bytes 类型 data 列"的落地位置：无论文档来自本地磁盘、SharePoint、Google Drive 还是 S3，只要在 $sources 层保证输出结构一致，DocumentStore 便能以统一方式完成解析、切分与索引，上层 question_answerer 无需关心数据究竟存放在哪里。

常见调整与排错要点

• 只想换数据源、不动其余配置：在 $sources 列表里增删连接器条目即可。可以把某个云端连接器整段注释掉、保留文件系统条目作为兜底，随时切换，无需改动 parser/splitter/index 的配置。
• 新增来源后表结构不满足 Document Store：若你自行使用 !pw.io.fs.read 且未设 format: binary（如默认 CSV），产出表会缺 data 字节列，DocumentStore 将无法索引。排查时优先确认该来源是否开启了二进制语义。
• 敏感参数：凡涉及密钥、租户、凭据的字段，统一使用全大写环境变量占位（$SHAREPOINT_URL、$s3_secret_access_key），并在启动模板前通过环境注入，避免密钥落入配置文件版本库。
• 云来源取不到新文件：检查 refresh_interval（默认 30 秒）与过滤条件：Google Drive 的 file_name_pattern（fnmatch 语义）与 object_size_limit（None 为不限）会直接跳过不匹配文件；S3 场景确认 path 为对象前缀且 region 与桶一致。

至此，你已掌握 Pathway RAG YAML 模板中数据源层的完整配置方法：从理解 $sources 变量与 data 列约束，到文件系统、SharePoint、Google Drive、S3 四种来源的参数级配置，再到把它们接入 DocumentStore 组装成端到端流水线。相关概念（连接器全览、YAML 变量语义、DocumentStore 与索引器）可分别查阅 YAML 配置语法、RAG 配置示例 与 模板运行指引 继续深入。