LightRAG 如何开发并注册第三方解析引擎：文本、Native、外部服务三种实现路径

LightRAG 如何开发并注册第三方解析引擎文本、Native、外部服务三种实现路径【免费下载链接】LightRAG[EMNLP2025] LightRAG: Simple and Fast Retrieval-Augmented Generation项目地址: https://gitcode.com/GitHub_Trending/li/LightRAG如果你要为自己的文档格式或自己的解析服务给 LightRAG 增加一个新的解析引擎需要完成两件事实现一个BaseParser子类然后注册一个ParserSpec。LightRAG 的解析层通过统一契约加中央注册表lightrag/parser/registry.py派发所有引擎——内置的native/legacy/mineru/docling与第三方引擎走完全相同的派发路径pipeline worker 和调试 CLI 都通过get_parser(engine).parse(ParseContext(...))驱动对内置引擎没有任何特殊分支。完成实现和注册后你的引擎自动获得独立或共享的解析并发池、三种引擎选择方式文件名 hint、LIGHTRAG_PARSER路由规则、API 的parse_engine参数、后缀能力校验以及单文件调试支持python -m lightrag.parser.cli --engine name。完整的插件编写指南见 docs/ThirdPartyParser.md调试 CLI 用法见 docs/ParserDebugCLI.mdsidecar 文件格式见 docs/LightRAGSidecarFormat.md。1. 理解解析契约BaseParser、ParseContext 与 ParseResult所有引擎含内置格式处理器reuse/passthrough都实现BaseParser契约定义见 lightrag/parser/base.pyclass MyParser(BaseParser): engine_name myengine # 必须与 ParserSpec.engine_name 一致 async def parse(self, ctx: ParseContext) - ParseResult: ...engine_name是注册表键也是--engine、文件名 hint、LIGHTRAG_PARSER使用的引擎名。下文示例统一用myengine作为占位名称你实现时改成自己的名字即可但所有出现处必须保持一致。ParseContext提供的成员摘自 docs/ThirdPartyParser.md成员说明ctx.ragLightRAG 实例用于_persist_parsed_full_docs等ctx.doc_id/ctx.file_path/ctx.content_data文档标识、规范化文件路径、full_docs行ctx.resolve(engine_name)返回ResolvedSource(source_path, document_name, parsed_dir)解析磁盘源文件路径、规范化文档名和派生的__parsed__/base.parsed/产物目录ctx.archive_source(path)解析成功且full_docs同步完成后把源文件归档进__parsed__/ParseResult的字段doc_id/file_path/parse_formatraw或lightrag/content/blocks_path无 sidecar 时为/parse_engine/parse_stage_skipped缓存命中等跳过场景/parse_warnings非致命警告持久化到doc_status.metadata。parse_warnings → doc_status.metadata是对所有 parser 的通用契约pipeline 只镜像 parser 返回的内容不检查键名。第三方 parser 返回的smart_*前缀警告同样会写入doc_status.metadata——只有内置 native DOCX 智能标题引擎私自把smart_/title_block_前缀诊断转存到 sidecar 的base.smart_audit.json那是该引擎的私有策略不是全局前缀规则。2. 三条实现路径按引擎类型选基类2.1 纯文本引擎无 sidecar直接继承BaseParser适合只产出纯文本、不需要 sidecar 块的引擎。参考实现是 lightrag/parser/legacy/parser.py 中的LegacyParser核心骨架class MyTextParser(BaseParser): engine_name myengine async def parse(self, ctx: ParseContext) - ParseResult: rs ctx.resolve(self.engine_name) source rs.source_path if not source.is_file(): raise FileNotFoundError(fmyengine source not found: {source}) text await asyncio.to_thread(my_extract, source) # 把 CPU 工作放到线程里执行 if not text.strip(): raise ValueError(fextracted no usable text from {ctx.file_path}) await ctx.rag._persist_parsed_full_docs(ctx.doc_id, { content: text, file_path: ctx.file_path, parse_format: FULL_DOCS_FORMAT_RAW, parse_engine: self.engine_name, update_time: int(time.time()), }) await ctx.archive_source(str(source)) return ParseResult( doc_idctx.doc_id, file_pathctx.file_path, parse_formatFULL_DOCS_FORMAT_RAW, contenttext, blocks_path, parse_engineself.engine_name, )其中my_extract是你自己的同步提取函数从源文件字节中抽出文本FULL_DOCS_FORMAT_RAW来自lightrag.constants。2.2 本地产出 sidecar 的引擎继承NativeParserBase如果引擎在本地完成解析并产出 sidecar 块继承NativeParserBaselightrag/parser/native_base.py。模板已固定完整流程“预清理产物目录带回滚→ 线程中提取 → 构建 IR → 写 sidecar → 持久化 → 归档”你只需实现两个钩子class MyNativeParser(NativeParserBase): engine_name myengine def extract(self, source, *, parsed_dir, asset_dir, base_name): 同步方法在线程中运行返回 (blocks, warnings, metadata)。 图片等资产可在 write_sidecar 之前写入 asset_dir。 def build_ir(self, blocks, *, document_name, asset_dir_name, metadata) - IRDoc: blocks - IRDoc交给共享的 sidecar writer。可选覆盖validate_source默认只要求文件存在、surface_warnings把提取警告映射到parse_warnings、finalize_parse_warnings完全控制在extract之后运行因此能看到完整警告字典——引擎可以把部分警告转存到 sidecar 审计产物、只把剩余部分返回给 doc_status默认实现就是调用surface_warnings。参考实现是 lightrag/parser/docx/parser.py它的finalize_parse_warnings会写智能标题的base.smart_audit.json。2.3 外部解析服务下载 raw bundle 缓存继承ExternalParserBase如果解析由外部服务完成、LightRAG 负责下载 raw bundle 并缓存继承ExternalParserBaselightrag/parser/external/_base.py。模板固定了流程“raw 缓存命中检查 → 未命中则清空目录重新下载 → 构建 IR → 写 sidecar → 持久化 → 归档”实现三个钩子加两个类属性class MyExternalParser(ExternalParserBase): engine_name myengine raw_dir_suffix .myengine_raw # raw bundle 目录后缀以 . 开头 force_reparse_env LIGHTRAG_FORCE_REPARSE_MYENGINE def is_bundle_valid(self, raw_dir, source_path) - bool: ... # 缓存命中检查 async def download_into(self, raw_dir, source_path, *, upload_name): ... def build_ir(self, raw_dir, document_name) - IRDoc: ...可选覆盖validate_ir构建后校验例如块数为零时失败。参考实现lightrag/parser/external/mineru/parser.py 和 lightrag/parser/external/docling/parser.py。注意文档中的钩子签名做了简化真实基类中is_bundle_valid/download_into/build_ir还带有一个engine_params: Mapping[str, Any] | None None关键字参数逐文件的引擎参数覆盖解码自parse_engine。基类注释明确要求engine_params必须参与缓存签名否则带不同参数覆盖的文档会误命中旧 bundle。写子类签名时以 lightrag/parser/external/_base.py 中的抽象方法为准。2.4 失败语义必须遵守parse(ctx)抛出任何异常时只有该文档被标记 FAILED错误信息写入doc_status.error_msg同批次其他文档不受影响。解析产出空内容时应抛异常而不是返回空字符串否则零知识文档会静默进入切块。所有内置引擎都遵循这一约定。worker 在调用引擎前做后缀守卫如果PENDING_PARSE文档的后缀不在该引擎ParserSpec.suffixes中文档直接 FAILED引擎代码不会被调用。3. 声明 ParserSpec能力元数据from lightrag.parser.registry import ParserSpec, register_parser register_parser(ParserSpec( engine_namemyengine, implmy_pkg.parser:MyParser, # module:Class由 get_parser 懒加载 suffixesfrozenset({pdf, foo}), # 小写、不带点 queue_groupmyengine, # 并发模型见下文 concurrencyint(os.getenv(MAX_PARALLEL_PARSE_MYENGINE, 2)), # 仅外部服务型引擎需要endpoint 未配置时路由会跳过该引擎 endpoint_configuredlambda: bool(os.getenv(MYENGINE_ENDPOINT, ).strip()), endpoint_requirementlambda: MYENGINE_ENDPOINT, ))字段要点完整表格见 docs/ThirdPartyParser.md字段必填说明engine_name是注册表键也是--engine、文件名 hint、LIGHTRAG_PARSER使用的名字。用已有引擎的同名注册会覆盖原注册含内置引擎除非有意替换避免与native/legacy/mineru/docling撞名impl是module:Class字符串只在文档真正被解析时导入。注册阶段绝不能提前导入实现能力查询必须保持轻量导入这是注册表的设计不变量suffixes是引擎可处理的后缀小写、不带点用于路由校验和 worker 侧后缀守卫普通set也可赋值时会被冻结extra_suffixes_env否一个环境变量名其逗号分隔的后缀在每次读取时并入suffixes如 docling 的DOCLING_ADDITIONAL_SUFFIXES。适合“真实格式覆盖依赖服务端可选包”的引擎suffixes声明始终可用的基线其余由部署自行开启。格式错误的条目会让服务器拒绝启动queue_group否并发池分组默认native共享 native 池独立池用唯一的组名concurrency否该组 worker 数只有组的唯一声明者需要填。环境变量覆盖在注册代码里于注册时固化如上例int(os.getenv(...))注册后的值是权威值endpoint_configured/endpoint_requirement否零参闭包只读环境变量、不做网络调用。前者返回外部服务是否已配置后者返回缺失时要展示给用户的配置项名。本地引擎不需要这两个字段默认可用user_selectable否默认TrueFalse表示内部格式处理器如reuse/passthrough不作为可选引擎展示并发模型每个批次会为每个queue_group建一个队列加一组 worker。内置组native/mineru/docling的 worker 数由 LightRAG 实例字段max_parallel_parse_*决定第三方专属组使用组唯一所有者 spec 的concurrency值一个组内只允许一个 spec 声明concurrency否则批次启动失败。共享queue_groupnative时concurrency不生效池大小由max_parallel_parse_native决定被忽略的 spec 级concurrency会在批次启动时记录警告日志。轻量本地引擎如legacy适合共享模式外部服务型引擎一般用独立组避免慢请求阻塞本地解析。4. 注册引擎entry point 自动发现推荐LightRAG 通过lightrag.parsersentry-point 组自动发现第三方引擎实现在 lightrag/parser/plugins.py。第三方包只需要两步。1. 在自己包的pyproject.toml声明 entry point[project.entry-points.lightrag.parsers] myengine my_pkg.lightrag_plugin:register2. 提供零参注册函数保持导入轻量不要在这里导入 parser 实现# my_pkg/lightrag_plugin.py import os from lightrag.parser.registry import ParserSpec, register_parser def register() - None: register_parser(ParserSpec( engine_namemyengine, implmy_pkg.parser:MyParser, # 实现类懒加载 suffixesfrozenset({foo}), queue_groupmyengine, concurrencyint(os.getenv(MAX_PARALLEL_PARSE_MYENGINE, 2)), ))pip install my-pkg之后无需修改 LightRAG 代码即可工作三个入口的行为API Servercreate_app()在验证LIGHTRAG_PARSER路由规则之前调用load_third_party_parsers()所以路由规则可以直接引用第三方引擎名如LIGHTRAG_PARSERfoo:myengine。上传与扫描的后缀守卫完全由“注册表 运行时路由”派生。判定标准是“这个文件能否路由到支持它的引擎”裸后缀无 hint上传必须有LIGHTRAG_PARSER规则把它路由到你的引擎否则文件会被直接拒绝而不是先接受、之后在解析阶段 FAILED带文件名 hint如report.[myengine].foo的上传无需规则即可通过。实用建议发布第三方引擎时在部署文档中提示用户配置对应的LIGHTRAG_PARSERfoo:myengine规则这样裸文件名上传和目录扫描都能自动工作。调试 CLIpython -m lightrag.parser.cli sample.foo --engine myengine直接可用main()在构建--engine选项前先加载插件。对无 sidecar 引擎blocks_pathCLI 打印纯文本摘要而不是块摘要继承ExternalParserBase的引擎自动获得 raw 缓存显示和--force-reparse支持。嵌入式库使用不经 server 或 CLI、直接使用LightRAG类在构建 pipeline 前调用一次from lightrag.parser.plugins import load_third_party_parsers load_third_party_parsers() # 进程内幂等加载语义每进程幂等重复调用无效果单个插件抛异常只记录日志并跳过不影响其他插件或内置引擎也不会阻塞 server 启动——但该引擎会不可用所以要看启动日志中的[parser-plugins]行。不想发布包时也可以跳过 entry point在自己的启动脚本里、启动/调用 LightRAG 之前直接调用register_parser(...)。注册表是进程内模块级单例效果相同只是没有“安装即生效”的行为。5. 路由让文档使用你的引擎引擎选择优先级lightrag/parser/routing.py文件名 hintreport.[myengine].foo允许携带处理选项如report.[myengine-iet].fooLIGHTRAG_PARSER规则例如LIGHTRAG_PARSERfoo:myengine,pdf:mineru按后缀 glob 匹配首个命中生效默认legacy。API 上传显式传parse_enginemyengine时直接锁定该引擎存入PENDING_PARSE行并由 worker 原样遵守不支持的后缀会 FAILED 而不是静默回退。注册了endpoint_configured的引擎在 endpoint 未配置时会被路由跳过hint/规则校验也会展示endpoint_requirement的提示。6. 验证引擎可用按从轻到重的顺序做三层验证1. 看启动日志。load_third_party_parsers成功时会记录[parser-plugins] loaded parser plugin ...单个插件失败会记录[parser-plugins] failed to load parser plugin ...并跳过。没有成功行就说明引擎未注册。2. 单测引擎本身。绕过 CLI 直接调用get_parser(myengine).parse(ParseContext(fake_rag, doc_id, file_path, content_data))。fake_rag只需提供_persist_parsed_full_docs/_resolve_source_file_for_parser/full_docs/doc_status可以参考 lightrag/parser/debug.py 中的build_debug_rag()或 tests/parser/test_legacy_parser.py 里的最小_FakeRag。注意注册表是模块级单例测试里调用register_parser之后要用finally: registry._REGISTRY.pop(myengine, None)清理写法参考 tests/parser/test_registry.pyentry-point 加载逻辑的测试参考 tests/parser/test_plugins.pymonkeypatchlightrag.parser.plugins.entry_points注入假 entry point并重置plugins._loaded标志。3. 调试 CLI 单文件跑通。用真实文件驱动与生产 worker 相同的注册表派发路径python -m lightrag.parser.cli sample.foo --engine myengineCLI 对任意注册引擎都有效输出布局、参数与生产解析路径的差异扁平目录、不归档源文件、raw 缓存只查目录存在性见 docs/ParserDebugCLI.md。无 sidecar 引擎会打印提取文本的前 400 字符而不是块摘要外部服务型引擎额外获得 raw 缓存显示与--force-reparse清空 raw 目录强制重新下载解析。缓存命中raw 目录已存在且非空时不需要外部服务环境变量可用于离线复现解析输出。7. 边界与限制同名覆盖用已有引擎名注册会覆盖原注册含内置引擎除非有意替换不要使用native/legacy/mineru/docling这些名字。实现类懒加载impl的导入只发生在文档真正被解析时能力查询后缀、endpoint、可选引擎列表不导入任何 parser 实现。若插件或注册阶段的导入变重会破坏注册表“导入轻量”的不变量。单插件失败隔离一个插件加载失败只记录日志并跳过不会阻塞 server 启动或其他插件代价是该引擎不可用——排障时先查[parser-plugins]日志行。后缀守卫在引擎之前路由规则或 hint 把你引擎之外的后缀路由进来时文档会在调用引擎代码前就被 FAILEDsuffixes声明含extra_suffixes_env扩展要如实反映引擎的真实能力。本文示例中的myengine、my_pkg、MAX_PARALLEL_PARSE_MYENGINE、MYENGINE_ENDPOINT均为文档示例名替换为你自己的包名、引擎名和环境变量名即可LIGHTRAG_PARSERfoo:myengine中的foo指你引擎处理的后缀。完成以上步骤后你的引擎与内置引擎走同一条get_parser(engine).parse(ParseContext(...))派发路径上传、目录扫描、APIparse_engine与调试 CLI 都会自动识别它。需要深入了解 sidecar 字段与命名约定时继续阅读 docs/LightRAGSidecarFormat.md 与 docs/ParserDebugCLI.md。【免费下载链接】LightRAG[EMNLP2025] LightRAG: Simple and Fast Retrieval-Augmented Generation项目地址: https://gitcode.com/GitHub_Trending/li/LightRAG创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考