LlamaIndex 数据加载与转换完全指南从 Document 到 Node 的标准化数据摄入管线【免费下载链接】llama_indexLlamaIndex is the document processing platform for AI项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index导读本文以 LlamaIndex 官方模块指南 Loading Data 为核心骨架系统讲解数据摄入data ingestion的两大关键环节——加载Loading与转换Transformations如何把本地目录、云存储、数据库等任意数据源加载为统一的Document再通过NodeParser与IngestionPipeline加工为可供索引与检索的Node。读完本文你将掌握SimpleDirectoryReader、LlamaHub 连接器、各类文本分割器、可缓存可并行的摄取管线以及Document/Node的深度定制能力可直接应用于 RAG 应用的数据准备阶段。一、数据摄入的核心模型Document → Transformations → NodesLlamaIndex 的数据摄入建立在一条简洁的抽象链上先加载Loading再转换Transformations最终产出节点Nodes。Document是围绕任意数据源PDF、API 输出、数据库记录等的通用容器存储文本与元数据Transformations是对已加载文档执行的加工操作最常见的即是文本切分splittingNode是源 Document 的块文本块、图片块等是 LlamaIndex 中的一等公民承载元数据与节点间关系。在 Node Parser 使用指南 中这一过程被进一步明确节点解析器接收一组 Document将其切分为Node且父文档的属性metadata、文本与元数据模板等会全部继承给子节点。一旦完成加载与转换即可在之上构建 Index通过 Query Engine 提问、通过 Chat Engine 对话。二、加载LoadingSimpleDirectoryReader 实战详解2.1 最简洁的本地文件加载器SimpleDirectoryReader是从本地文件加载数据的最简单方式。对生产环境更推荐使用 LlamaHub 上的专用 Reader但它仍是快速上手的最佳选择。from llama_index.core import SimpleDirectoryReader reader SimpleDirectoryReader(input_dirpath/to/directory) documents reader.load_data()2.2 默认支持的文件类型默认情况下SimpleDirectoryReader会尝试读取目录中的任何文件并按纯文本处理此外根据文件扩展名自动识别以下类型扩展名说明.csv逗号分隔值.docxMicrosoft Word.epubEPUB 电子书格式.hwp韩文文字处理器Hangul Word Processor.ipynbJupyter Notebook.jpeg/.jpgJPEG 图片.mboxMBOX 邮件归档.mdMarkdown.mp3/.mp4音频与视频.pdf便携式文档格式.png便携式网络图形.ppt/.pptm/.pptxMicrosoft PowerPoint注意JSON 并不在默认列表内官方推荐使用独立的 JSON Loader。2.3 并行加载与流式迭代加载大量文件时可启用并行处理。源码中 base.py 的load_data通过num_workers参数控制并行度documents reader.load_data(num_workers4)需要留意平台差异Windows 与 Linux/macOS 在multiprocessing上的启动方式不同spawn 与 forkserverWindows 用户可能获得较少甚至没有性能提升而 Linux/macOS 用户对同一批文件能明显获益。如需边加载边处理可使用iter_data()按文件逐个产出reader SimpleDirectoryReader(input_dirpath/to/directory, recursiveTrue) all_docs [] for docs in reader.iter_data(): # do something with the documents per file all_docs.extend(docs)2.4 目录遍历与文件过滤默认仅读取目录顶层读取子目录需设置recursiveTrue只加载指定文件列表用input_files排除指定文件用exclude仅加载指定扩展名用required_exts源码 base.py 中据此过滤ref.suffix限制最大文件数用num_files_limit源码中其默认上限取max_files与显式值中的较小者。SimpleDirectoryReader(input_dirpath/to/directory, recursiveTrue) SimpleDirectoryReader(input_files[path/to/file1, path/to/file2]) SimpleDirectoryReader( input_dirpath/to/directory, exclude[path/to/file1, path/to/file2] ) SimpleDirectoryReader( input_dirpath/to/directory, required_exts[.pdf, .docx] ) SimpleDirectoryReader(input_dirpath/to/directory, num_files_limit100)2.5 文件编码SimpleDirectoryReader默认期望文件为utf-8编码可通过encoding参数覆盖SimpleDirectoryReader(input_dirpath/to/directory, encodinglatin-1)2.6 自动元数据与自定义元数据加载时每个Document会自动附带metadata字典默认包含file_path完整路径、file_name含后缀文件名、file_type由mimetypes.guess_type()推断的 MIME 类型、file_size字节数以及creation_date、last_modified_date、last_accessed_date统一归一化到 UTC 时区格式为%Y-%m-%d若日期看起来相差一天多为 UTC 午夜偏移所致。可通过file_metadata参数传入自定义函数接收文件路径、返回元数据字典替换默认逻辑def get_meta(file_path): return {foo: bar, file_path: file_path} reader SimpleDirectoryReader( input_dirpath/to/directory, file_metadataget_meta ) docs reader.load_data() print(docs[0].metadata[foo]) # prints bar源码中该钩子由 default_file_metadata_func 与file_metadata属性实现并在load_data内部对每个文件调用。2.7 扩展到自定义文件类型通过file_extractor传入扩展名 →BaseReader实例的映射即可让SimpleDirectoryReader读取其他文件类型。BaseReader需实现读取文件并返回Document列表from llama_index.core import SimpleDirectoryReader from llama_index.core.readers.base import BaseReader from llama_index.core import Document class MyFileReader(BaseReader): def load_data(self, file, extra_infoNone): with open(file, r) as f: text f.read() # load_data returns a list of Document objects return [Document(texttext Foobar, extra_infoextra_info or {})] reader SimpleDirectoryReader( input_dir./data, file_extractor{.myfile: MyFileReader()} ) documents reader.load_data() print(documents)注意该映射会覆盖对应类型的默认提取器如需同时支持原有类型需自行把默认提取器加回映射。2.8 支持外部文件系统fsspecSimpleDirectoryReader接受可选fs参数可遍历远程文件系统。任何实现了 fsspec 协议的文件系统对象均可使用AWS S3、Azure Blob DataLake、Google Drive、SFTP 等。以 S3 为例from s3fs import S3FileSystem s3_fs S3FileSystem(key..., secret...) bucket_name my-document-bucket reader SimpleDirectoryReader( input_dirbucket_name, fss3_fs, recursiveTrue, # recursively searches all subdirectories ) documents reader.load_data() print(documents)完整的远程文件系统示例可参考 simple_directory_reader_remote_fs.ipynb。三、加载LoadingLlamaHub 数据连接器数据连接器Data Connector即Reader负责把不同数据源与格式的数据摄入为统一的Document表示文本 简单元数据。这些连接器通过 LlamaHub 提供——一个包含数百个即插即用数据加载库的开放注册中心。典型用法如下以 Google Docs 为例from llama_index.core import download_loader from llama_index.readers.google import GoogleDocsReader loader GoogleDocsReader() documents loader.load_data(document_ids[...])常见连接器示例本地文件目录SimpleDirectoryReader支持.pdf、.jpg、.png、.docx等广泛文件类型NotionNotionPageReaderGoogle DocsGoogleDocsReaderSlackSlackReaderDiscordDiscordReaderApify ActorsApifyActor可爬取网页、抓取页面、提取文本并下载.pdf、.jpg、.png、.docx等文件。完整列表见 connector/modules.md详细使用模式见 connector/usage_pattern.md。四、转换TransformationsNode Parser 的使用模式Node Parser 是接收 Document 列表并将其切分为Node的简单抽象。可在三种场景中使用1. 独立使用from llama_index.core import Document from llama_index.core.node_parser import SentenceSplitter node_parser SentenceSplitter(chunk_size1024, chunk_overlap20) nodes node_parser.get_nodes_from_documents( [Document(textlong text)], show_progressFalse )2. 作为摄取管线的转换步骤from llama_index.core import SimpleDirectoryReader from llama_index.core.ingestion import IngestionPipeline from llama_index.core.node_parser import TokenTextSplitter documents SimpleDirectoryReader(./data).load_data() pipeline IngestionPipeline(transformations[TokenTextSplitter(), ...]) nodes pipeline.run(documentsdocuments)3. 在索引构建时自动生效通过全局Settings.text_splitter或索引级transformations参数使from_documents()自动执行切分from llama_index.core import SimpleDirectoryReader, VectorStoreIndex from llama_index.core.node_parser import SentenceSplitter documents SimpleDirectoryReader(./data).load_data() # global from llama_index.core import Settings Settings.text_splitter SentenceSplitter(chunk_size1024, chunk_overlap20) # per-index index VectorStoreIndex.from_documents( documents, transformations[SentenceSplitter(chunk_size1024, chunk_overlap20)], )五、转换TransformationsNode Parser 模块全览5.1 文件型 Node Parser针对内容结构JSON、Markdown、HTML 等进行解析。最简单的方式是组合FlatFileReader与SimpleFileNodeParser让系统自动为每种内容选择最佳解析器随后可再链上文本型解析器以控制文本长度。SimpleFileNodeParser自动选择最佳的文件解析器。from llama_index.core.node_parser import SimpleFileNodeParser from llama_index.readers.file import FlatReader from pathlib import Path md_docs FlatReader().load_data(Path(./test.md)) parser SimpleFileNodeParser() md_nodes parser.get_nodes_from_documents(md_docs)HTMLNodeParser基于beautifulsoup解析原始 HTML。默认解析[p, h1, h2, h3, h4, h5, h6, li, b, i, u, section]标签可通过tags覆盖。from llama_index.core.node_parser import HTMLNodeParser parser HTMLNodeParser(tags[p, h1]) # optional list of tags nodes parser.get_nodes_from_documents(html_docs)JSONNodeParser解析原始 JSON。from llama_index.core.node_parser import JSONNodeParser parser JSONNodeParser() nodes parser.get_nodes_from_documents(json_docs)MarkdownNodeParser解析原始 Markdown 文本。from llama_index.core.node_parser import MarkdownNodeParser parser MarkdownNodeParser() nodes parser.get_nodes_from_documents(markdown_docs)5.2 文本分割器Text SplittersCodeSplitter按编程语言切分代码文本支持语言清单见 py-tree-sitter-languages。from llama_index.core.node_parser import CodeSplitter splitter CodeSplitter( languagepython, chunk_lines40, # lines per chunk chunk_lines_overlap15, # lines overlap between chunks max_chars1500, # max chars per chunk ) nodes splitter.get_nodes_from_documents(documents)LangchainNodeParser包装任意 langchain 文本分割器。from langchain.text_splitter import RecursiveCharacterTextSplitter from llama_index.core.node_parser import LangchainNodeParser parser LangchainNodeParser(RecursiveCharacterTextSplitter()) nodes parser.get_nodes_from_documents(documents)Chunker多用途解析器包装 chonkie 分块器可用别名初始化完整别名见Chunker.valid_chunker_types也可直接传入 chonkie 实例。from llama_index.node_parser.chonkie import Chunker parser Chunker(recursive, chunk_size2048) nodes parser.get_nodes_from_documents(documents)SentenceSplitter尊重句子边界切分文本默认chunk_size1024、chunk_overlap20。SentenceWindowNodeParser将文档切分为单个句子并把每句周围的窗口句子存入节点元数据该元数据对 LLM 与嵌入模型不可见。常用于生成范围极精确的嵌入再配合MetadataReplacementNodePostProcessor在送入 LLM 前用上下文替换句子from llama_index.core.node_parser import SentenceWindowNodeParser node_parser SentenceWindowNodeParser.from_defaults( # how many sentences on either side to capture window_size3, # the metadata key that holds the window of surrounding sentences window_metadata_keywindow, # the metadata key that holds the original sentence original_text_metadata_keyoriginal_sentence, )完整示例见 metadatareplacementdemo。SemanticSplitterNodeParser语义分块不采用固定块大小而是利用嵌入相似度在句子间自适应选择断点保证块内句子语义相关。注意两点正则主要适用于英文句子可能需要调优断点百分位阈值。from llama_index.core.node_parser import SemanticSplitterNodeParser from llama_index.embeddings.openai import OpenAIEmbedding embed_model OpenAIEmbedding() splitter SemanticSplitterNodeParser( buffer_size1, breakpoint_percentile_threshold95, embed_modelembed_model )完整示例见 semantic_chunking。TokenTextSplitter按原始 token 数保持一致的块大小。from llama_index.core.node_parser import TokenTextSplitter splitter TokenTextSplitter( chunk_size1024, chunk_overlap20, separator , ) nodes splitter.get_nodes_from_documents(documents)5.3 关系型 Node ParserHierarchicalNodeParser将节点切分为层级结构单一输入被切分为多个尺寸层级的块每个节点持有对其父节点的引用。与AutoMergingRetriever配合时当多数子节点被检索到即可自动替换为其父节点为响应合成提供更完整的上下文from llama_index.core.node_parser import HierarchicalNodeParser node_parser HierarchicalNodeParser.from_defaults( chunk_sizes[2048, 512, 128] )完整示例见 auto_merging_retriever。六、串联全流程IngestionPipelineIngestionPipeline是把一切串起来的摄入管线它应用一组Transformations处理输入数据产出的 Nodes 要么直接返回要么若指定自动插入向量数据库。每个 nodetransformation 组合都会被哈希缓存后续运行若缓存已持久化可直接复用结果节省时间。6.1 基础用法from llama_index.core import Document from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.core.node_parser import SentenceSplitter from llama_index.core.extractors import TitleExtractor from llama_index.core.ingestion import IngestionPipeline, IngestionCache # create the pipeline with transformations pipeline IngestionPipeline( transformations[ SentenceSplitter(chunk_size25, chunk_overlap0), TitleExtractor(), OpenAIEmbedding(), ] ) # run the pipeline nodes pipeline.run(documents[Document.example()])真实场景中文档通常来自SimpleDirectoryReader或 LlamaHub 的其他 Reader。Document.example()可快速创建示例文档用于原型验证。6.2 连接向量数据库可让管线直接把结果节点写入远程向量库之后再基于该向量库构建索引from llama_index.core import Document from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.core.node_parser import SentenceSplitter from llama_index.core.extractors import TitleExtractor from llama_index.core.ingestion import IngestionPipeline from llama_index.vector_stores.qdrant import QdrantVectorStore import qdrant_client client qdrant_client.QdrantClient(location:memory:) vector_store QdrantVectorStore(clientclient, collection_nametest_store) pipeline IngestionPipeline( transformations[ SentenceSplitter(chunk_size25, chunk_overlap0), TitleExtractor(), OpenAIEmbedding(), ], vector_storevector_store, ) # Ingest directly into a vector db pipeline.run(documents[Document.example()]) # Create your index from llama_index.core import VectorStoreIndex index VectorStoreIndex.from_vector_store(vector_store)关键约束连接向量库时嵌入计算必须是管线的一个阶段否则后续实例化索引会失败若仅需产出节点列表、不接向量库则可以省略嵌入步骤。6.3 缓存机制每个 nodetransformation 组合会被哈希并缓存源码见 pipeline.py 中get_transformation_hash与cache.get/put的配合。本地缓存管理# save pipeline.persist(./pipeline_storage) # load and restore state new_pipeline IngestionPipeline( transformations[ SentenceSplitter(chunk_size25, chunk_overlap0), TitleExtractor(), ], ) new_pipeline.load(./pipeline_storage) # will run instantly due to the cache nodes pipeline.run(documents[Document.example()])缓存过大时可清空# delete all context of the cache cache.clear()远程缓存管理支持RedisCache、MongoDBCache、FirestoreCache等后端以 Redis 为例from llama_index.core import Document from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.core.node_parser import SentenceSplitter from llama_index.core.extractors import TitleExtractor from llama_index.core.ingestion import IngestionPipeline, IngestionCache from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache ingest_cache IngestionCache( cacheRedisCache.from_host_and_port(host127.0.0.1, port6379), collectionmy_test_cache, ) pipeline IngestionPipeline( transformations[ SentenceSplitter(chunk_size25, chunk_overlap0), TitleExtractor(), OpenAIEmbedding(), ], cacheingest_cache, ) # Ingest directly into a vector db nodes pipeline.run(documents[Document.example()])使用远程缓存时无需显式persist写入过程即实时缓存到指定集合。6.4 异步支持nodes await pipeline.arun(documentsdocuments)源码中arun_transformations与run_transformations保持同构的缓存逻辑pipeline.py。6.5 文档管理去重为管线附加docstore即启用文档管理以document.doc_id或node.ref_doc_id为基准主动查找重复文档。其工作方式对应源码 pipeline.py 的_handle_duplicate_docs与_handle_upserts存储doc_id→document_hash的映射附加了向量库时检测到重复doc_id且哈希已变 → 重新处理并 upsert哈希未变 → 跳过该节点未附加向量库时逐一检查节点哈希重复则跳过否则处理此时仅能检查和移除重复输入。from llama_index.core.ingestion import IngestionPipeline from llama_index.core.storage.docstore import SimpleDocumentStore pipeline IngestionPipeline( transformations[...], docstoreSimpleDocumentStore() )完整演练见 document_management_pipeline以 Redis 作为完整摄取栈的指南见 redis_ingestion_pipeline。6.6 并行处理run方法可通过multiprocessing.Pool将节点批次分发到多进程执行。源码pipeline.py会在num_workers超过 CPU 数时自动下调至最大 CPU 数from llama_index.core.ingestion import IngestionPipeline pipeline IngestionPipeline( transformations[...], ) pipeline.run(documents[...], num_workers4)更多主题见 transformations 指南 及 advanced_ingestion_pipeline、async_ingestion_pipeline、parallel_execution_ingestion_pipeline 等示例。七、核心抽象Document 与 Node7.1 概念与基础用法Document是围绕任意数据源的通用容器PDF、API 输出、数据库数据等可手动构造或由数据加载器自动创建。默认存储文本及若干属性metadata可附加到文本的注解字典、relationships与其他 Document/Node 的关系字典。仓库同时提供多模态能力的 beta 支持。Node表示源 Document 的块文本块、图片等同样包含元数据与关系信息是 LlamaIndex 的一等公民。默认情况下从 Document 派生的每个 Node 都会继承该 Document 的元数据例如file_name字段会传播到每个 Node。基础用法from llama_index.core import Document, VectorStoreIndex text_list [text1, text2, ...] documents [Document(textt) for t in text_list] # build index index VectorStoreIndex.from_documents(documents)from llama_index.core.node_parser import SentenceSplitter # load documents ... # parse nodes parser SentenceSplitter() nodes parser.get_nodes_from_documents(documents) # build index index VectorStoreIndex(nodes)7.2 定制 DocumentDocument 是TextNode的子类因此以下设置同样适用于TextNode。元数据注入metadata字典中的信息会出现在源文档派生的每个节点中并默认注入到嵌入生成与 LLM 调用的文本里。三种设置方式# 1. 构造时 document Document( texttext, metadata{filename: doc_file_name, category: category}, ) # 2. 构造后 document.metadata {filename: doc_file_name} # 3. 通过 SimpleDirectoryReader 的 file_metadata 钩子自动设置 from llama_index.core import SimpleDirectoryReader filename_fn lambda filename: {file_name: filename} documents SimpleDirectoryReader( ./data, file_metadatafilename_fn ).load_data()若对接向量数据库注意部分向量库要求 key 为字符串、value 为扁平类型str/float/int。自定义 IDdoc_id用于索引中文档的高效刷新。SimpleDirectoryReader可通过filename_as_idTrue将doc_id自动设为完整文件路径也可直接赋值document.doc_id My new document id!或通过node_id、id_属性设置。精细控制 LLM 与嵌入模型所见文本默认情况下所有元数据都会同时进入嵌入与 LLM。可用excluded_llm_metadata_keys让某些键不出现在 LLM 合成响应时读取的文本中从而只偏置检索、不改变 LLM 阅读内容用excluded_embed_metadata_keys排除嵌入可见的元数据document.excluded_llm_metadata_keys [file_name] from llama_index.core.schema import MetadataMode print(document.get_content(metadata_modeMetadataMode.LLM))document.excluded_embed_metadata_keys [file_name] from llama_index.core.schema import MetadataMode print(document.get_content(metadata_modeMetadataMode.EMBED))自定义元数据注入格式元数据以文本形式注入时由三个属性控制Document.metadata_seperator默认\n—— 各 key/value 对之间的分隔符Document.metadata_template默认{key}: {value}—— 单个键值对的格式必须包含key、value变量Document.text_template默认{metadata_str}\n\n{content}—— 元数据字符串与正文内容拼接的模板必须包含metadata、content变量。综合示例from llama_index.core import Document from llama_index.core.schema import MetadataMode document Document( textThis is a super-customized document, metadata{ file_name: super_secret_document.txt, category: finance, author: LlamaIndex, }, excluded_llm_metadata_keys[file_name], metadata_seperator::, metadata_template{key}{value}, text_templateMetadata: {metadata_str}\n-----\nContent: {content}, ) print( The LLM sees this: \n, document.get_content(metadata_modeMetadataMode.LLM), ) print( The Embedding model sees this: \n, document.get_content(metadata_modeMetadataMode.EMBED), )此外还可利用 LLM 自身进行自动元数据提取初始示例见 usage_metadata_extractor。7.3 手动构造 Node 与关系Node 也可直接手动构造并通过NodeRelationship与RelatedNodeInfo显式定义关系from llama_index.core.schema import TextNode, NodeRelationship, RelatedNodeInfo node1 TextNode(texttext_chunk, id_node_id) node2 TextNode(texttext_chunk, id_node_id) # set relationships node1.relationships[NodeRelationship.NEXT] RelatedNodeInfo( node_idnode2.node_id ) node2.relationships[NodeRelationship.PREVIOUS] RelatedNodeInfo( node_idnode1.node_id ) nodes [node1, node2]RelatedNodeInfo还可携带额外元数据node2.relationships[NodeRelationship.PARENT] RelatedNodeInfo( node_idnode1.node_id, metadata{key: val} )每个节点拥有自动生成的node_id未手动指定时可用于存储更新、通过IndexNode定义节点间关系等也可直接读写print(node.node_id) node.node_id My new node_id!八、小结一条从数据源到索引的完整链路LlamaIndex 的数据摄入可归纳为一条可复现、可缓存、可并行的标准链路加载用SimpleDirectoryReader本地/远程文件系统或 LlamaHub 连接器Notion、Slack、Google Docs 等把任意来源的数据统一为Document转换用NodeParser文件型、文本分割器、关系型三大类把Document切分为结构清晰、语义合适的Node并按需附加元数据编排用IngestionPipeline把多步转换串成管线接入向量库、启用缓存本地/远程、文档去重与多进程并行最终把Node落库或直接交给VectorStoreIndex构建索引定制通过metadata、doc_id、excluded_llm_metadata_keys/excluded_embed_metadata_keys及三大文本模板精细控制进入嵌入模型与 LLM 的实际内容。掌握这套体系即可在真实项目中高效、稳定地完成 RAG 应用的数据准备阶段。后续可继续阅读 Indexing 模块指南、Query Engine 指南 与 Chat Engine 指南把摄入的节点真正用起来。【免费下载链接】llama_indexLlamaIndex is the document processing platform for AI项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考