企业官网建设流程全解析

GraphRAG 如何通过自定义 InputReader 扩展新的输入文件格式？

【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphrag

GraphRAG 的索引管道内置了text、csv、json、jsonl、parquet和markitdown六种输入类型（见 docs/config/yaml.md）。如果你的数据以其他格式存放，且不想先写脚本转换成已有格式，官方给出的扩展方式就是：实现一个继承InputReader的类，再把它注册到InputReaderFactory（见 docs/index/inputs.md 的 “Custom File Handling” 一节）。这篇文章就是这条路径：从读懂内置 Reader 的接口约定，到写出自定义 Reader、注册、配置settings.yaml并验证结果。

先理解两个约定：documents 表和 TextDocument

所有输入格式读入后都会变成同一个documentsDataFrame，每行一个文档，列结构是固定的（见 docs/index/inputs.md）：

对应的 Python 数据类是 TextDocument，字段为id、text、title、creation_date、raw_data（可选）。你的自定义 Reader 最终产出的就是这种对象列表，后续的 chunking 行为与内置格式完全一致。

阅读内置 Reader：接口长什么样

InputReader 是一个抽象基类，你的实现只需要关心其中一部分：

• 构造函数接收storage、file_pattern、encoding（默认"utf-8"）以及任意额外关键字参数。file_pattern是一个正则，基类用它调用self._storage.find(re.compile(self._file_pattern))在存储里找文件；
• 抽象方法read_file(self, path: str) -> list[TextDocument]必须实现：把单个文件读成一个或多个TextDocument；
• 基类已经实现了read_files()（收集所有文件）和_iterate_files()（逐文件迭代）。注意两点内置行为：一个文件读取抛异常时只打 warning（Warning! Error loading file %s. Skipping...）并跳过，不会中断整个管道；找不到匹配文件时会打 warningNo {file_pattern} matches found in storage并返回空。

TextFileReader 是最简单的参考实现：从self._storage.get(path, encoding=...)读文本，用gen_sha512_hash({"text": text}, ["text"])生成 id，title 取文件名，creation_date 通过self._storage.get_creation_date(path)获取，raw_data传None。

第一步：实现你的 Reader 并注册

下面以一个假设的.mynotes纯文本格式为例。如果read_file里的解析逻辑要换成你自己的格式解析（二进制、结构化等），替换这一处即可，其余骨架保持不变。注册函数register_input_reader定义在 input_reader_factory.py，签名是register_input_reader(input_reader_type, input_reader_initializer, scope="transient")，其中scope取"transient"或"singleton"（见 Factory.register，singleton 表示相同初始化参数复用同一实例）。

from pathlib import Path from graphrag_input.hashing import gen_sha512_hash from graphrag_input.input_reader import InputReader from graphrag_input.input_reader_factory import register_input_reader from graphrag_input.text_document import TextDocument class MyNotesReader(InputReader): """Reader implementation for .mynotes files.""" def __init__(self, file_pattern: str | None = None, **kwargs): super().__init__( file_pattern=file_pattern if file_pattern is not None else ".*\\.mynotes$", **kwargs, ) async def read_file(self, path: str) -> list[TextDocument]: """Read a .mynotes file into a list of documents.""" text = await self._storage.get(path, encoding=self._encoding) document = TextDocument( id=gen_sha512_hash({"text": text}, ["text"]), title=str(Path(path).name), text=text, creation_date=await self._storage.get_creation_date(path), raw_data=None, ) return [document] # 第一个参数是 settings.yaml 里 input.type 会使用的名字，可以是任意字符串 register_input_reader("mynotes", MyNotesReader)

注册代码必须在create_input_reader被调用之前执行。create_input_reader 的逻辑是：先看config.type是否已在工厂里注册，没有就尝试注册六个内置类型（csv/text/json/jsonl/markitdown/parquet），仍找不到则抛出ValueError，错误信息会列出当前已注册的类型：

InputConfig.type 'mynotes' is not registered in the InputReaderFactory. Registered types: ...

这条报错本身就是第一条验证手段：如果忘了注册或名字写错，管道启动时就会在这里失败，而不是静默读不到文件。

另外，InputConfig 显式开启了extra="allow"，用途注释写明是 “Allow extra fields to support custom reader implementations”——也就是说input配置块里的额外字段会连同storage、file_pattern、encoding等一起作为关键字参数传给你的 Reader 构造函数，需要额外配置的 Reader 可以靠这个机制接收。

第二步：配置 settings.yaml

在数据项目根目录的settings.yaml中，把input.type设成你注册的名字。graphrag-input 包 README 中 markitdown 的配置示例展示了 input 与输入存储的写法，可以照此组织：

input: type: mynotes file_pattern: ".*\\.mynotes$" encoding: utf-8 input_storage: type: file base_dir: "input"

各字段的说明依据 docs/config/yaml.md 的input一节：file_pattern是匹配输入文件的正则，内置类型有由type推导出的默认值（如.*\.csv$），自定义类型没有内置默认，实际生效的会是你 Reader 构造函数里的回退默认值（示例中为.*\.mynotes$），配置里显式写出可以覆盖它；input块还支持storage（type为file|memory|blob|cosmosdb、base_dir等）来指定输入文件存放位置。id_column、title_column、text_column属于结构化格式的字段映射配置，纯文本类 Reader 用不到。

第三步：运行索引

有两种入口：

• API：build_index 接收GraphRagConfig，内部通过PipelineFactory创建管道并执行。你的注册代码放在同一脚本里、build_index调用之前执行即可。
• CLI：graphrag index，具体参数见 docs/cli.md 与 docs/get_started.md。

可选替代路径：如果你的数据最终在 DataFrame 里，build_index还支持input_documents参数直接传入 pandas DataFrame，完全绕开文件读取（见 docs/index/inputs.md 的 “Bring-your-own DataFrame”）。这条路径要求你自行保证 DataFrame 符合上文 documents 表的结构。它适合内容在自定义存储里的场景，与本文的文件 Reader 扩展是两条并列路线，二选一即可。

验证：文件被找到了、文档进管道了

文档给出的可核对信号有这三处，全部来自实际代码行为：

1. 启动即验证注册：input.type未注册时，create_input_reader抛出上文那个带已注册类型列表的ValueError。
2. 读取阶段日志：基类_iterate_files在读取完成后会输出Found %d %s files, loading %d（匹配到的文件数、文件模式、加载的文档数）。若某个文件解析失败，日志里会有Warning! Error loading file ...且该文件被跳过——出现这类 warning 说明个别文件解析有问题，需要单独排查。
3. 管道产物：索引完成后会落documents表（parquet），结构即上文五列 schema；其最终 schema 细节见 docs/index/outputs.md。核对id、text、title、creation_date是否符合预期，即可确认自定义 Reader 的输出正确接入了后续 chunking。

边界与限制

• Reader 里读取的文件内容如何变成TextDocument完全由你决定：raw_data填了内容，就可以配合chunking.prepend_metadata把源文件字段前缀到每个 chunk（见 docs/index/inputs.md 的 “Field Prepending”）。
• 单个文件读取异常只告警并跳过，不会让整次索引失败；“零文件匹配”同样只是 warning。这两点意味着输入目录放错时管道仍会跑完，要特别留意日志里的 warning。
• 工厂允许用任意字符串名字注册实现，也可以直接覆盖内置类型名（text、csv等）。覆盖内置行为前建议先理解对应内置 Reader 的实现。
• build_index 所在的 API 模块 标注了 “under development”，不保证向后兼容，依赖它做二次开发时留意这一点。

【免费下载链接】graphragA modular graph-based Retrieval-Augmented Generation (RAG) system项目地址: https://gitcode.com/GitHub_Trending/gr/graphrag