llama_index索引处理源码分析

版本环境：py310 环境的 LlamaIndex 0.14.25

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import StorageContext
from base_llm import llm, embed_model
import chromadb

#
documents = SimpleDirectoryReader(input_files=["../../data_file/小说.txt"]).load_data()

print("---------------使用chroma进行存储向量--------------------")
# 创建客户端和新的集合
# chroma_client = chromadb.Ephem =eralClient()  # 创建一个内存对象
chroma_client = chromadb.PersistentClient("./chroma_db")  # 创建一个本地存储的对象
chroma_collection = chroma_client.get_or_create_collection("quickstart")


# 设置ChromaVectorStore并加载数据
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)

Storage_Context = StorageContext.from_defaults(vector_store=vector_store)

# 创建索引  VectorStoreIndex 默认的索引工具
index = VectorStoreIndex.from_documents(documents, storage_context=Storage_Context, embed_model=embed_model)
#
# # 创建检索器   向量检索
# res = index.as_retriever().retrieve('萧炎妹妹是谁')
# print(res)

源码直接读取自项目配置引用的解释器环境：

D:\ProgramData\miniconda3\envs\py310\python.exe
D:\ProgramData\miniconda3\envs\py310\Lib\site-packages

文中的源码链接直接定位到本机安装文件的定义行。代码块直接摘取本地文件；省略说明字符串或只摘取某个分支时，会在代码前注明。方法名后的“定义类”与运行时的 self 分开说明，便于对照继承调用。

0. 导入 base_llm：准备模型

脚本先执行：

from base_llm import llm, embed_model

首次导入会执行同目录 base_llm.py，加载环境变量、创建 DashScope 对象，并执行：

embed_model = HuggingFaceEmbedding(
    model_name=r"D:\embedding\bge-m3",
    device="cuda",
)

本地源码：[HuggingFaceEmbedding.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/embeddings/huggingface/base.py:123)，llama_index/embeddings/huggingface/base.py 第 123 行。

运行时 self：HuggingFaceEmbedding 实例。

HuggingFaceEmbedding.__init__() 创建 SentenceTransformer(...) 并保存到 self._model，同时保存设备、批次、归一化等配置。此时准备模型；节点向量在后面的 _embed_with_retry() 中计算。本脚本启用的索引构建只使用 embed_model，没有调用 llm 生成答案。

1. 读取小说：SimpleDirectoryReader(...).load_data()

脚本代码：

documents = SimpleDirectoryReader(input_files=["../../data_file/小说.txt"]).load_data()

1.1 SimpleDirectoryReader.init()：检查文件

本地源码：[SimpleDirectoryReader.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/readers/file/base.py:249)，llama_index/core/readers/file/base.py 第 249 行。

运行时 self：SimpleDirectoryReader 实例。

检查输入文件是否存在，把路径保存到 self.input_files，并保存编码、元数据提取函数等配置。相对路径按进程当前工作目录解析。

1.2 SimpleDirectoryReader.load_data()：按文件调用 load_file()

本地源码：[SimpleDirectoryReader.load_data()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/readers/file/base.py:718)，llama_index/core/readers/file/base.py 第 718 行。

运行时 self：SimpleDirectoryReader 实例。

这版先用 partial() 把读取参数绑定到 load_file()。脚本没有设置多进程，因此进入 else 分支，依次调用 load_file_with_args(input_file)，最后调用 _exclude_metadata()。

源码选段，第 739–750 行：

fs = fs or self.fs
load_file_with_args = partial(
    SimpleDirectoryReader.load_file,
    file_metadata=self.file_metadata,
    file_extractor=self.file_extractor,
    filename_as_id=self.filename_as_id,
    encoding=self.encoding,
    errors=self.errors,
    raise_on_error=self.raise_on_error,
    fs=fs,
)

源码选段，第 774–786 行：

else:
    files_to_process = cast(
        list[Union[Path, PurePosixPath]],
        get_tqdm_iterable(
            self.input_files,
            show_progress=show_progress,
            desc="Loading files",
        ),
    )
    for input_file in files_to_process:
        documents.extend(load_file_with_args(input_file))

return self._exclude_metadata(documents)

1.3 SimpleDirectoryReader.load_file()：创建 Document

本地源码：[SimpleDirectoryReader.load_file()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/readers/file/base.py:555)，llama_index/core/readers/file/base.py 第 555 行。

调用方式：这是静态方法，没有 self 参数。

本例 .txt 未配置专用读取器，走普通读取分支：提取文件元数据 → 打开文件 → 读取并解码 → 创建 Document → 放进列表返回。

源码选段，第 635–646 行：

    # do standard read
    fs = fs or get_default_fs()
    with fs.open(input_file, errors=errors, encoding=encoding) as f:
        data = cast(bytes, f.read()).decode(encoding, errors=errors)

    doc = Document(text=data, metadata=metadata or {})  # type: ignore
    if filename_as_id:
        doc.id_ = str(input_file)

    documents.append(doc)

return documents

本地源码：[SimpleDirectoryReader._exclude_metadata()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/readers/file/base.py:434)，llama_index/core/readers/file/base.py 第 434 行。

运行时 self：SimpleDirectoryReader 实例。

load_data() 最后通过 _exclude_metadata() 设置文件名、时间等字段的 embedding / LLM 排除配置，元数据仍保留在对象上。返回到脚本：documents 是 Document 列表，还没有切分或计算向量。

2. 创建 Chroma 客户端和集合

脚本代码：

chroma_client = chromadb.PersistentClient("./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("quickstart")

2.1 chromadb.PersistentClient()：设置本地持久化参数

本地源码：[PersistentClient()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/chromadb/init.py:199)，chromadb/__init__.py 第 199 行。

本地方法代码（仅省略方法开头的说明字符串）：

def PersistentClient(
    path: Union[str, Path] = "./chroma",
    settings: Optional[Settings] = None,
    tenant: str = DEFAULT_TENANT,
    database: str = DEFAULT_DATABASE,
) -> ClientAPI:
    if settings is None:
        settings = Settings()
    settings.persist_directory = str(path)
    settings.is_persistent = True

    # Make sure paramaters are the correct types -- users can pass anything.
    tenant = str(tenant)
    database = str(database)

    return ClientCreator(tenant=tenant, database=database, settings=settings)

设置数据库目录与 is_persistent=True，再返回客户端。

2.2 Client.get_or_create_collection()：取得集合

本地源码：[Client.get_or_create_collection()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/chromadb/api/client.py:319)，chromadb/api/client.py 第 319 行。

运行时 self：chromadb.api.client.Client 实例。

先检查集合配置中的 embedding 函数配置，再调用 self._server.get_or_create_collection(...)，最后把返回的集合模型包装成 Collection 返回。到这里只准备了集合，小说数据尚未写入。

3. ChromaVectorStore.init()：保存集合对象

本地源码：[ChromaVectorStore.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/vector_stores/chroma/base.py:172)，llama_index/vector_stores/chroma/base.py 第 172 行。

运行时 self：ChromaVectorStore 实例。

脚本代码：

vector_store = ChromaVectorStore(chroma_collection=chroma_collection)

构造函数初始化适配器配置后判断集合参数。因为本例已经传入集合，执行 else 分支：

self._collection = cast(Collection, chroma_collection)

cast() 是类型标注辅助，这里保存的仍是脚本中的集合对象。该类的 stores_text=True，后面写入时会同时提交正文。返回到脚本：得到 vector_store。

4. StorageContext.from_defaults()：组合存储对象

本地源码：[StorageContext.from_defaults()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/storage/storage_context.py:74)，llama_index/core/storage/storage_context.py 第 74 行。

脚本代码：

Storage_Context = StorageContext.from_defaults(vector_store=vector_store)

本例未传 persist_dir，进入 if persist_dir is None 分支，顺序创建默认 docstore、index_store、图存储等对象，将传入的 Chroma 存入默认向量存储命名空间，最后返回 StorageContext。

源码选段，第 97–111 行：

if persist_dir is None:
    docstore = docstore or SimpleDocumentStore()
    index_store = index_store or SimpleIndexStore()
    graph_store = graph_store or SimpleGraphStore()
    image_store = image_store or SimpleVectorStore()

    if vector_store:
        vector_stores = {DEFAULT_VECTOR_STORE: vector_store}
    else:
        vector_stores = vector_stores or {
            DEFAULT_VECTOR_STORE: SimpleVectorStore()
        }
    if image_store:
        # append image store to vector stores
        vector_stores[IMAGE_VECTOR_STORE_NAMESPACE] = image_store

源码选段，第 143–149 行：

return cls(
    docstore=docstore,
    index_store=index_store,
    vector_stores=vector_stores,  # type: ignore
    graph_store=graph_store,
    property_graph_store=property_graph_store,
)

返回到脚本：Storage_Context.vector_store 是 Chroma 适配器，docstore 和 index_store 使用默认实现。

5. VectorStoreIndex.from_documents()：切分、向量化、入库

脚本代码：

index = VectorStoreIndex.from_documents(documents, storage_context=Storage_Context, embed_model=embed_model)

下面逐步进入这行代码内部。第 5.24 步结束后，脚本才拿到 index。

5.1 BaseIndex.from_documents()：入口方法定义在父类

本地源码：[BaseIndex.from_documents()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/base.py:89)，llama_index/core/indices/base.py 第 89 行。

定义类是 BaseIndex；这是类方法，本例 cls 是 VectorStoreIndex。先看完整控制顺序，再逐层展开里面的方法调用：

本地方法代码（仅省略方法开头的说明字符串）：

@classmethod
def from_documents(
    cls: Type[IndexType],
    documents: Sequence[Document],
    storage_context: Optional[StorageContext] = None,
    show_progress: bool = False,
    callback_manager: Optional[CallbackManager] = None,
    transformations: Optional[List[TransformComponent]] = None,
    **kwargs: Any,
) -> IndexType:
    storage_context = storage_context or StorageContext.from_defaults()
    docstore = storage_context.docstore
    callback_manager = callback_manager or Settings.callback_manager
    transformations = transformations or Settings.transformations

    with callback_manager.as_trace("index_construction"):
        for doc in documents:
            docstore.set_document_hash(doc.id_, doc.hash)

        nodes = run_transformations(
            documents,  # type: ignore
            transformations,
            show_progress=show_progress,
            **kwargs,
        )

        return cls(
            nodes=nodes,
            storage_context=storage_context,
            callback_manager=callback_manager,
            show_progress=show_progress,
            transformations=transformations,
            **kwargs,
        )

先取得存储和回调配置。脚本没传 transformations，因此在取得默认转换链时进入下一步。注意当前版本保存哈希用的是 doc.id_。

5.2 _Settings.transformations → node_parser：取得 SentenceSplitter

本地源码：[_Settings.transformations()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/settings.py:278)，llama_index/core/settings.py 第 278 行。

运行时 self：_Settings，脚本通过全局 Settings 对象访问 实例。

本地方法代码（仅省略方法开头的说明字符串）：

@property
def transformations(self) -> List[TransformComponent]:
    if self._transformations is None:
        self._transformations = [self.node_parser]
    return self._transformations

本地源码：[_Settings.node_parser()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/settings.py:140)，llama_index/core/settings.py 第 140 行。

运行时 self：_Settings 实例。

本地方法代码（仅省略方法开头的说明字符串）：

@property
def node_parser(self) -> NodeParser:
    if self._node_parser is None:
        self._node_parser = SentenceSplitter()

    if self._callback_manager is not None:
        self._node_parser.callback_manager = self._callback_manager

    return self._node_parser

默认未修改配置时，转换链得到 [SentenceSplitter()]。Settings 是 _Settings 的实例，属性的定义类是 _Settings。

本地源码：[SentenceSplitter.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:68)，llama_index/core/node_parser/text/sentence.py 第 68 行。

运行时 self：SentenceSplitter 实例。

创建切分器时设置默认块预算、重叠预算及切分函数；当前本地实现包含以下初始化逻辑：

源码选段，第 101–121 行：

self._chunking_tokenizer_fn = (
    chunking_tokenizer_fn or split_by_sentence_tokenizer()
)
self._tokenizer = tokenizer or get_tokenizer()

self._split_fns = [
    split_by_sep(paragraph_separator),
    self._chunking_tokenizer_fn,
]

if secondary_chunking_regex:
    self._sub_sentence_split_fns = [
        split_by_regex(secondary_chunking_regex),
        split_by_sep(separator),
        split_by_char(),
    ]
else:
    self._sub_sentence_split_fns = [
        split_by_sep(separator),
        split_by_char(),
    ]

返回到 BaseIndex.from_documents()：取得默认转换链，接着记录文档哈希。

5.3 docstore.set_document_hash()：保存源文档哈希

仍在 BaseIndex.from_documents() 中：

with callback_manager.as_trace("index_construction"):
    for doc in documents:
        docstore.set_document_hash(doc.id_, doc.hash)

按文档 ID 保存哈希，然后继续调用 run_transformations(...)。入口中没有通过比较旧哈希而跳过文档的分支。

5.4 run_transformations()：调用转换组件

本地源码：[run_transformations()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/ingestion/pipeline.py:72)，llama_index/core/ingestion/pipeline.py 第 72 行。

本地方法代码（仅省略方法开头的说明字符串）：

def run_transformations(
    nodes: Sequence[BaseNode],
    transformations: Sequence[TransformComponent],
    in_place: bool = True,
    cache: Optional[IngestionCache] = None,
    cache_collection: Optional[str] = None,
    show_progress: bool = False,
    **kwargs: Any,
) -> Sequence[BaseNode]:
    if not in_place:
        nodes = list(nodes)

    transformations_with_progress = get_tqdm_iterable(
        transformations, show_progress, "Applying transformations"
    )

    for transform in transformations_with_progress:
        if cache is not None:
            hash = get_transformation_hash(nodes, transform)
            cached_nodes = cache.get(hash, collection=cache_collection)
            if cached_nodes is not None:
                nodes = cached_nodes
            else:
                nodes = transform(nodes, **kwargs)
                cache.put(hash, nodes, collection=cache_collection)
        else:
            nodes = transform(nodes, **kwargs)

    return nodes

本例未配置缓存，进入循环中的 else，执行 transform(nodes, **kwargs)。当前 transform 是 SentenceSplitter 实例，输入 nodes 是 documents。实例被当函数调用时，进入继承来的 NodeParser.__call__()。

5.5 NodeParser.call() → NodeParser.get_nodes_from_documents()

先确定方法归属：运行时 self 始终是当前的 SentenceSplitter 实例；方法查找按继承顺序进行：

SentenceSplitter → MetadataAwareTextSplitter → TextSplitter → NodeParser

本地源码：[NodeParser.call()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/interface.py:203)，llama_index/core/node_parser/interface.py 第 203 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def __call__(self, nodes: Sequence[BaseNode], **kwargs: Any) -> List[BaseNode]:
    return self.get_nodes_from_documents(nodes, **kwargs)  # type: ignore

本地源码：[NodeParser.get_nodes_from_documents()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/interface.py:157)，llama_index/core/node_parser/interface.py 第 157 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def get_nodes_from_documents(
    self,
    documents: Sequence[Document],
    show_progress: bool = False,
    **kwargs: Any,
) -> List[BaseNode]:
    doc_id_to_document = {doc.id_: doc for doc in documents}

    with self.callback_manager.event(
        CBEventType.NODE_PARSING, payload={EventPayload.DOCUMENTS: documents}
    ) as event:
        nodes = self._parse_nodes(documents, show_progress=show_progress, **kwargs)
        nodes = self._postprocess_parsed_nodes(nodes, doc_id_to_document)

        event.on_end({EventPayload.NODES: nodes})

    return nodes

先按文档 ID 建立映射，再开启 NODE_PARSING 回调，调用 _parse_nodes()，随后调用 _postprocess_parsed_nodes()，结束回调后返回节点。虽然当前方法定义在 NodeParser，self._parse_nodes() 仍按实际对象的继承顺序查找，命中 MetadataAwareTextSplitter 的实现。

如果 IDE 跳到了 NodeParser._parse_nodes() 的抽象声明，或者 TextSplitter._parse_nodes()，可以在当前方法的断点中查看真实绑定：

type(self)
self.get_nodes_from_documents.__func__.__qualname__
self._parse_nodes.__func__.__qualname__

本例默认切分器对应的结果应分别是 SentenceSplitter、NodeParser.get_nodes_from_documents、MetadataAwareTextSplitter._parse_nodes。

5.6 MetadataAwareTextSplitter._parse_nodes()：逐篇处理

本地源码：[MetadataAwareTextSplitter._parse_nodes()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/interface.py:283)，llama_index/core/node_parser/interface.py 第 283 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _parse_nodes(
    self, nodes: Sequence[BaseNode], show_progress: bool = False, **kwargs: Any
) -> List[BaseNode]:
    all_nodes: List[BaseNode] = []
    nodes_with_progress = get_tqdm_iterable(nodes, show_progress, "Parsing nodes")

    for node in nodes_with_progress:
        metadata_str = self._get_metadata_str(node)
        splits = self.split_text_metadata_aware(
            node.get_content(metadata_mode=MetadataMode.NONE),
            metadata_str=metadata_str,
        )
        all_nodes.extend(
            build_nodes_from_splits(splits, node, id_func=self.id_func)
        )

    return all_nodes

本地源码：[MetadataAwareTextSplitter._get_metadata_str()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/interface.py:270)，llama_index/core/node_parser/interface.py 第 270 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _get_metadata_str(self, node: BaseNode) -> str:
    embed_metadata_str = node.get_metadata_str(mode=MetadataMode.EMBED)
    llm_metadata_str = node.get_metadata_str(mode=MetadataMode.LLM)

    # use the longest metadata str for splitting
    if len(embed_metadata_str) > len(llm_metadata_str):
        metadata_str = embed_metadata_str
    else:
        metadata_str = llm_metadata_str

    return metadata_str

先取得 embedding、LLM 模式的元数据字符串，按字符串长度选择较长者；取纯正文，调用 self.split_text_metadata_aware()。这里的方法实现来自 SentenceSplitter。切分返回字符串后，才调用 build_nodes_from_splits() 包装节点。

5.7 SentenceSplitter.split_text_metadata_aware()：计算正文预算

本地源码：[SentenceSplitter.split_text_metadata_aware()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:156)，llama_index/core/node_parser/text/sentence.py 第 156 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def split_text_metadata_aware(self, text: str, metadata_str: str) -> List[str]:
    metadata_len = len(self._tokenizer(metadata_str))
    effective_chunk_size = self.chunk_size - metadata_len
    if effective_chunk_size <= 0:
        raise ValueError(
            f"Metadata length ({metadata_len}) is longer than chunk size "
            f"({self.chunk_size}). Consider increasing the chunk size or "
            "decreasing the size of your metadata to avoid this."
        )
    elif effective_chunk_size < 50:
        print(
            f"Metadata length ({metadata_len}) is close to chunk size "
            f"({self.chunk_size}). Resulting chunks are less than 50 tokens. "
            "Consider increasing the chunk size or decreasing the size of "
            "your metadata to avoid this.",
            flush=True,
        )

    return self._split_text(text, chunk_size=effective_chunk_size)

计算元数据 token 数并从 chunk_size 中扣除；剩余预算不足时输出提示或报错，正常时进入 _split_text()。

5.8 SentenceSplitter._split_text()：拆分与合并的组织方法

本地源码：[SentenceSplitter._split_text()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:179)，llama_index/core/node_parser/text/sentence.py 第 179 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _split_text(self, text: str, chunk_size: int) -> List[str]:
    if text == "":
        return [text]

    with self.callback_manager.event(
        CBEventType.CHUNKING, payload={EventPayload.CHUNKS: [text]}
    ) as event:
        splits = self._split(text, chunk_size)
        chunks = self._merge(splits, chunk_size)

        event.on_end(payload={EventPayload.CHUNKS: chunks})

    return chunks

先执行 _split() 得到小片段，再执行 _merge() 得到最终字符串块。两步都处于 CHUNKING 回调中，完成后返回 chunks。

5.9 SentenceSplitter._split() → _get_splits_by_fns()：递归切分

本地源码：[SentenceSplitter._split()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:198)，llama_index/core/node_parser/text/sentence.py 第 198 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _split(self, text: str, chunk_size: int) -> List[_Split]:
    token_size = self._token_size(text)
    if token_size <= chunk_size:
        return [_Split(text, is_sentence=True, token_size=token_size)]

    text_splits_by_fns, is_sentence = self._get_splits_by_fns(text)

    text_splits = []
    for text_split_by_fns in text_splits_by_fns:
        token_size = self._token_size(text_split_by_fns)
        if token_size <= chunk_size:
            text_splits.append(
                _Split(
                    text_split_by_fns,
                    is_sentence=is_sentence,
                    token_size=token_size,
                )
            )
        elif len(text_splits_by_fns) == 1:
            # Could not split any further (a single indivisible unit larger
            # than chunk_size, e.g. a multi-token CJK / emoji character with
            # a small chunk_size). Recursing on the same text loops forever
            # and raises RecursionError, so keep it as an oversized split --
            # _merge() then raises the intended "Single token exceeded chunk
            # size" ValueError instead of crashing.
            text_splits.append(
                _Split(
                    text_split_by_fns,
                    is_sentence=is_sentence,
                    token_size=token_size,
                )
            )
        else:
            recursive_text_splits = self._split(
                text_split_by_fns, chunk_size=chunk_size
            )
            text_splits.extend(recursive_text_splits)
    return text_splits

本地源码：[SentenceSplitter._get_splits_by_fns()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:334)，llama_index/core/node_parser/text/sentence.py 第 334 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _get_splits_by_fns(self, text: str) -> Tuple[List[str], bool]:
    for split_fn in self._split_fns:
        splits = split_fn(text)
        if len(splits) > 1:
            return splits, True

    for split_fn in self._sub_sentence_split_fns:
        splits = split_fn(text)
        if len(splits) > 1:
            break

    return splits, False

处理顺序：

1. _token_size() 判断当前文本是否已在预算内，是则直接返回 _Split。
2. 超出时通过 _get_splits_by_fns() 依次尝试段落、分句器、备用正则、分隔符、字符切分。
3. 对拆出的每段重新计算长度，没超预算则收集；仍超预算且拆出了多段，则递归调用 _split()。
4. 当前本地版本有 elif len(text_splits_by_fns) == 1 分支：确实无法继续拆分时保留超长片段，避免对同一段文本无限递归；随后由 _merge() 抛出明确的长度异常。

返回到 _split_text() 后，继续执行 _merge()。

5.10 SentenceSplitter._merge()：合并片段并处理重叠

本地源码：[SentenceSplitter._merge()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:247)，llama_index/core/node_parser/text/sentence.py 第 247 行。

运行时 self：SentenceSplitter 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _merge(self, splits: List[_Split], chunk_size: int) -> List[str]:
    chunks: List[str] = []
    cur_chunk: List[Tuple[str, int]] = []  # list of (text, length)
    last_chunk: List[Tuple[str, int]] = []
    cur_chunk_len = 0
    new_chunk = True

    def close_chunk() -> None:
        nonlocal chunks, cur_chunk, last_chunk, cur_chunk_len, new_chunk

        chunks.append("".join([text for text, length in cur_chunk]))
        last_chunk = cur_chunk
        cur_chunk = []
        cur_chunk_len = 0
        new_chunk = True

        # add overlap to the next chunk using the last one first
        if len(last_chunk) > 0:
            last_index = len(last_chunk) - 1
            while (
                last_index >= 0
                and cur_chunk_len + last_chunk[last_index][1] <= self.chunk_overlap
            ):
                overlap_text, overlap_length = last_chunk[last_index]
                cur_chunk_len += overlap_length
                cur_chunk.insert(0, (overlap_text, overlap_length))
                last_index -= 1

    split_idx = 0
    while split_idx < len(splits):
        cur_split = splits[split_idx]
        if cur_split.token_size > chunk_size:
            raise ValueError("Single token exceeded chunk size")
        if cur_chunk_len + cur_split.token_size > chunk_size and not new_chunk:
            # if adding split to current chunk exceeds chunk size: close out chunk
            close_chunk()
        else:
            # If this is a new chunk with overlap, and adding the split would
            # exceed chunk_size, remove overlap to make room
            if new_chunk and cur_chunk_len + cur_split.token_size > chunk_size:
                # Remove overlap from the beginning until split fits
                while (
                    len(cur_chunk) > 0
                    and cur_chunk_len + cur_split.token_size > chunk_size
                ):
                    _, length = cur_chunk.pop(0)
                    cur_chunk_len -= length

            if (
                cur_split.is_sentence
                or cur_chunk_len + cur_split.token_size <= chunk_size
                or new_chunk  # new chunk, always add at least one split
            ):
                # add split to chunk
                cur_chunk_len += cur_split.token_size
                cur_chunk.append((cur_split.text, cur_split.token_size))
                split_idx += 1
                new_chunk = False
            else:
                # close out chunk
                close_chunk()

    # handle the last chunk
    if not new_chunk:
        chunk = "".join([text for text, length in cur_chunk])
        chunks.append(chunk)

    # run postprocessing to remove blank spaces
    return self._postprocess_chunks(chunks)

close_chunk() 结束当前块，并把上一块的尾部片段放到下一块作为重叠内容。外层循环用 split_idx 逐个处理输入片段。

你本地这版有删除重叠内容的逻辑：新块如果因重叠部分导致放不下当前片段，会从 cur_chunk 开头移除重叠片段，直到腾出空间。单个不可拆片段自身超过预算时则抛出 Single token exceeded chunk size。

本地源码：[SentenceSplitter._postprocess_chunks()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/text/sentence.py:318)，llama_index/core/node_parser/text/sentence.py 第 318 行。

运行时 self：SentenceSplitter 实例。

最后调用 _postprocess_chunks() 去掉空白块和首尾空白。依次返回 _split_text() → split_text_metadata_aware() → MetadataAwareTextSplitter._parse_nodes()，现在得到的是 List[str]。

5.11 build_nodes_from_splits()：字符串块包装为 TextNode

本地源码：[build_nodes_from_splits()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/node_utils.py:29)，llama_index/core/node_parser/node_utils.py 第 29 行。

此函数先准备来源关系，再按输入类型创建节点。对本例 Document，执行下面的准备代码和 elif isinstance(document, Document) 分支：

源码选段，第 36–42 行：

ref_doc = ref_doc or document
id_func = id_func or default_id_func
nodes: List[TextNode] = []
"""Calling as_related_node_info() on a document recomputes the hash for the whole text and metadata"""
"""It is not that bad, when creating relationships between the nodes, but is terrible when adding a relationship"""
"""between the node and a document, hence we create the relationship only once here and pass it to the nodes"""
relationships = {NodeRelationship.SOURCE: ref_doc.as_related_node_info()}

源码选段，第 61–73 行：

elif isinstance(document, Document):
    node = TextNode(
        id_=id_func(i, document),
        text=text_chunk,
        embedding=document.embedding,
        excluded_embed_metadata_keys=document.excluded_embed_metadata_keys,
        excluded_llm_metadata_keys=document.excluded_llm_metadata_keys,
        metadata_seperator=document.metadata_separator,
        metadata_template=document.metadata_template,
        text_template=document.text_template,
        relationships=relationships,
    )
    nodes.append(node)

本地源码：[default_id_func()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/node_utils.py:25)，llama_index/core/node_parser/node_utils.py 第 25 行。

默认 ID 函数返回 UUID。每个节点继承文档的向量、元数据排除配置及文本模板，并通过 SOURCE 关联原文档。本例文档未带向量，节点的 embedding 仍为 None。

返回顺序：build_nodes_from_splits() 返回节点列表 → _parse_nodes() 将各文档的节点汇总到 all_nodes → 返回 NodeParser.get_nodes_from_documents()。

5.12 NodeParser._postprocess_parsed_nodes()：补全位置、元数据与关系

本地源码：[NodeParser._postprocess_parsed_nodes()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/node_parser/interface.py:84)，llama_index/core/node_parser/interface.py 第 84 行。

运行时 self：SentenceSplitter 实例。

该方法定义在 NodeParser。调用时传入上一步的节点列表和源文档映射。你本地的完整方法如下：

本地方法代码（仅省略方法开头的说明字符串）：

def _postprocess_parsed_nodes(
    self, nodes: List[BaseNode], parent_doc_map: Dict[str, Document]
) -> List[BaseNode]:
    # Track search position per document to handle duplicate text correctly
    # Nodes are assumed to be in document order from _parse_nodes
    # We track the START position (not end) to allow for overlapping chunks
    doc_search_positions: Dict[str, int] = {}

    for i, node in enumerate(nodes):
        parent_doc = parent_doc_map.get(node.ref_doc_id or "", None)
        parent_node = node.source_node

        if parent_doc is not None:
            if parent_doc.source_node is not None:
                node.relationships.update(
                    {
                        NodeRelationship.SOURCE: parent_doc.source_node,
                    }
                )

            # Get or initialize search position for this document
            doc_id = node.ref_doc_id or ""
            search_start = doc_search_positions.get(doc_id, 0)

            # Search for node content starting from the last found position
            node_content = node.get_content(metadata_mode=MetadataMode.NONE)
            start_char_idx = parent_doc.text.find(node_content, search_start)

            # update start/end char idx
            if start_char_idx >= 0 and isinstance(node, TextNode):
                node.start_char_idx = start_char_idx
                node.end_char_idx = start_char_idx + len(node_content)
                # Update search position to start from next character after this node's START
                # This allows overlapping chunks to be found correctly
                doc_search_positions[doc_id] = start_char_idx + 1

            # update metadata
            if self.include_metadata:
                # Merge parent_doc.metadata into nodes.metadata, giving preference to node's values
                node.metadata = {**parent_doc.metadata, **node.metadata}

        if parent_node is not None:
            if self.include_metadata:
                parent_metadata = parent_node.metadata

                combined_metadata = {**parent_metadata, **node.metadata}

                # Merge parent_node.metadata into nodes.metadata, giving preference to node's values
                node.metadata.update(combined_metadata)

        if self.include_prev_next_rel:
            # establish prev/next relationships if nodes share the same source_node
            if (
                i > 0
                and node.source_node
                and nodes[i - 1].source_node
                and nodes[i - 1].source_node.node_id == node.source_node.node_id  # type: ignore
            ):
                node.relationships[NodeRelationship.PREVIOUS] = nodes[
                    i - 1
                ].as_related_node_info()
            if (
                i < len(nodes) - 1
                and node.source_node
                and nodes[i + 1].source_node
                and nodes[i + 1].source_node.node_id == node.source_node.node_id  # type: ignore
            ):
                node.relationships[NodeRelationship.NEXT] = nodes[
                    i + 1
                ].as_related_node_info()

    return nodes

本版为每个文档维护 doc_search_positions。每次从上次找到的起点之后继续搜索，既能处理正文重复的情况，也允许节点重叠。找到后设置 start_char_idx、end_char_idx，随后按配置合并元数据、建立同源相邻节点的前后关系。

依次返回：_postprocess_parsed_nodes() → get_nodes_from_documents() → NodeParser.__call__() → run_transformations() → BaseIndex.from_documents()。现在入口方法里的 nodes 已经是切好的节点列表。

5.13 回到 BaseIndex.from_documents()：执行 return cls(...)

继续执行第 5.1 步入口中的最后一段：

return cls(
    nodes=nodes,
    storage_context=storage_context,
    callback_manager=callback_manager,
    show_progress=show_progress,
    transformations=transformations,
    **kwargs,
)

本例 cls 是 VectorStoreIndex，embed_model 经 **kwargs 传入。接下来先进入子类构造，再由子类调用父类构造。

5.14 VectorStoreIndex.init() → resolve_embed_model()

本地源码：[VectorStoreIndex.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:50)，llama_index/core/indices/vector_store/base.py 第 50 行。

运行时 self：VectorStoreIndex 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def __init__(
    self,
    nodes: Optional[Sequence[BaseNode]] = None,
    # vector store index params
    use_async: bool = False,
    store_nodes_override: bool = False,
    embed_model: Optional[EmbedType] = None,
    insert_batch_size: int = 2048,
    # parent class params
    objects: Optional[Sequence[IndexNode]] = None,
    index_struct: Optional[IndexDict] = None,
    storage_context: Optional[StorageContext] = None,
    callback_manager: Optional[CallbackManager] = None,
    transformations: Optional[List[TransformComponent]] = None,
    show_progress: bool = False,
    **kwargs: Any,
) -> None:
    self._use_async = use_async
    self._store_nodes_override = store_nodes_override
    self._embed_model = resolve_embed_model(
        embed_model or Settings.embed_model, callback_manager=callback_manager
    )

    self._insert_batch_size = insert_batch_size
    super().__init__(
        nodes=nodes,
        index_struct=index_struct,
        storage_context=storage_context,
        show_progress=show_progress,
        objects=objects,
        callback_manager=callback_manager,
        transformations=transformations,
        **kwargs,
    )

本脚本默认 use_async=False、store_nodes_override=False、insert_batch_size=2048。该版对 embed_model or Settings.embed_model 调用 resolve_embed_model()，再保存模型并进入 super().__init__()。

本地源码：[resolve_embed_model()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/embeddings/utils.py:30)，llama_index/core/embeddings/utils.py 第 30 行。

本例传入已有 HuggingFaceEmbedding 实例，跳过字符串模型和其他框架适配分支，最终确认 BaseEmbedding 接口、设置回调并返回。

源码选段，第 135–139 行：

assert isinstance(embed_model, BaseEmbedding)

embed_model.callback_manager = callback_manager or Settings.callback_manager

return embed_model

5.15 BaseIndex.init()：绑定存储并触发构建

本地源码：[BaseIndex.init()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/base.py:37)，llama_index/core/indices/base.py 第 37 行。

运行时 self：VectorStoreIndex 实例。

父类先校验 nodes、objects 与 index_struct 的组合，再执行以下代码。注意方法定义在 BaseIndex，self 仍是 VectorStoreIndex。

源码选段，第 64–86 行：

self._storage_context = storage_context or StorageContext.from_defaults()
self._docstore = self._storage_context.docstore
self._show_progress = show_progress
self._vector_store = self._storage_context.vector_store
self._graph_store = self._storage_context.graph_store
self._callback_manager = callback_manager or Settings.callback_manager

objects = objects or []
self._object_map = {obj.index_id: obj.obj for obj in objects}
for obj in objects:
    obj.obj = None  # clear the object to avoid serialization issues

with self._callback_manager.as_trace("index_construction"):
    if index_struct is None:
        nodes = nodes or []
        index_struct = self.build_index_from_nodes(
            nodes + objects,  # type: ignore
            **kwargs,  # type: ignore
        )
    self._index_struct = index_struct
    self._storage_context.index_store.add_index_struct(self._index_struct)

self._transformations = transformations or Settings.transformations

本例未传已有索引结构，因此执行 self.build_index_from_nodes(nodes + objects, **kwargs)；objects 为空。该方法在子类有重写，下一步进入 VectorStoreIndex.build_index_from_nodes()。注册索引的后半段要等构建返回后才执行。

5.16 VectorStoreIndex.build_index_from_nodes()：过滤空内容

本地源码：[VectorStoreIndex.build_index_from_nodes()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:286)，llama_index/core/indices/vector_store/base.py 第 286 行。

运行时 self：VectorStoreIndex 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def build_index_from_nodes(
    self,
    nodes: Sequence[BaseNode],
    **insert_kwargs: Any,
) -> IndexDict:
    # Filter out the nodes that don't have content
    content_nodes = [
        node
        for node in nodes
        if node.get_content(metadata_mode=MetadataMode.EMBED) != ""
    ]

    # Report if some nodes are missing content
    if len(content_nodes) != len(nodes):
        print("Some nodes are missing content, skipping them...")

    return self._build_index_from_nodes(content_nodes, **insert_kwargs)

按 MetadataMode.EMBED 内容过滤空节点，然后进入 _build_index_from_nodes()。

5.17 VectorStoreIndex._build_index_from_nodes()：创建索引结构

本地源码：[VectorStoreIndex._build_index_from_nodes()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:260)，llama_index/core/indices/vector_store/base.py 第 260 行。

运行时 self：VectorStoreIndex 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _build_index_from_nodes(
    self,
    nodes: Sequence[BaseNode],
    **insert_kwargs: Any,
) -> IndexDict:
    index_struct = self.index_struct_cls()
    if self._use_async:
        tasks = [
            self._async_add_nodes_to_index(
                index_struct,
                nodes,
                show_progress=self._show_progress,
                **insert_kwargs,
            )
        ]
        run_async_tasks(tasks)
    else:
        self._add_nodes_to_index(
            index_struct,
            nodes,
            show_progress=self._show_progress,
            **insert_kwargs,
        )
    return index_struct

创建 IndexDict；本例 _use_async=False，执行 else 中的 _add_nodes_to_index()。

5.18 VectorStoreIndex._add_nodes_to_index()：每批先向量化再入库

本地源码：[VectorStoreIndex._add_nodes_to_index()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:219)，llama_index/core/indices/vector_store/base.py 第 219 行。

运行时 self：VectorStoreIndex 实例。

先看该方法前半段，回写判断在第 5.23 步继续：

源码选段，第 227–232 行：

if not nodes:
    return

for nodes_batch in iter_batch(nodes, self._insert_batch_size):
    nodes_batch = self._get_node_with_embedding(nodes_batch, show_progress)
    new_ids = self._vector_store.add(nodes_batch, **insert_kwargs)

通过 iter_batch() 每次取最多 2048 个节点。先进入 _get_node_with_embedding()，它首先调用 embed_nodes()；该方法返回后，才设置副本上的向量，再执行向量库 add()。

5.19 embed_nodes()：收集缺少向量的节点文本

本地源码：[embed_nodes()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/utils.py:170)，llama_index/core/indices/utils.py 第 170 行。

本地方法代码（仅省略方法开头的说明字符串）：

def embed_nodes(
    nodes: Sequence[BaseNode], embed_model: BaseEmbedding, show_progress: bool = False
) -> Dict[str, List[float]]:
    id_to_embed_map: Dict[str, List[float]] = {}

    texts_to_embed = []
    ids_to_embed = []
    for node in nodes:
        if node.embedding is None:
            ids_to_embed.append(node.node_id)
            texts_to_embed.append(node.get_content(metadata_mode=MetadataMode.EMBED))
        else:
            id_to_embed_map[node.node_id] = node.embedding

    new_embeddings = embed_model.get_text_embedding_batch(
        texts_to_embed, show_progress=show_progress
    )

    for new_id, text_embedding in zip(ids_to_embed, new_embeddings):
        id_to_embed_map[new_id] = text_embedding

    return id_to_embed_map

已有向量的节点直接加入映射；缺少向量的节点按相同顺序记录 ID 和 MetadataMode.EMBED 文本，调用模型批量接口。当前模型对象是 HuggingFaceEmbedding，get_text_embedding_batch() 的定义来自父类 BaseEmbedding。

5.20 BaseEmbedding.get_text_embedding_batch() → HuggingFaceEmbedding._embed_with_retry()

本地源码：[BaseEmbedding.get_text_embedding_batch()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/base/embeddings/base.py:486)，llama_index/core/base/embeddings/base.py 第 486 行。

运行时 self：HuggingFaceEmbedding 实例。

本地方法代码（仅省略方法开头的说明字符串）：

@dispatcher.span
def get_text_embedding_batch(
    self,
    texts: List[str],
    show_progress: bool = False,
    **kwargs: Any,
) -> List[Embedding]:
    cur_batch: List[str] = []
    result_embeddings: List[Embedding] = []

    queue_with_progress = enumerate(
        get_tqdm_iterable(texts, show_progress, "Generating embeddings")
    )

    model_dict = self.to_payload()
    for idx, text in queue_with_progress:
        cur_batch.append(text)
        if idx == len(texts) - 1 or len(cur_batch) == self.embed_batch_size:
            # flush
            dispatcher.event(
                EmbeddingStartEvent(
                    model_dict=model_dict,
                )
            )
            with self.callback_manager.event(
                CBEventType.EMBEDDING,
                payload={EventPayload.SERIALIZED: model_dict},
            ) as event:
                if self.rate_limiter is not None:
                    self.rate_limiter.acquire()
                if not self.embeddings_cache:
                    embeddings = self._get_text_embeddings(cur_batch)
                elif self.embeddings_cache is not None:
                    embeddings = self._get_text_embeddings_cached(cur_batch)
                result_embeddings.extend(embeddings)
                event.on_end(
                    payload={
                        EventPayload.CHUNKS: cur_batch,
                        EventPayload.EMBEDDINGS: embeddings,
                    },
                )
            dispatcher.event(
                EmbeddingEndEvent(
                    chunks=cur_batch,
                    embeddings=embeddings,
                )
            )
            cur_batch = []

    return result_embeddings

这版按 embed_batch_size 分批，带有回调、追踪事件、可选限流及 embedding 缓存分支。本脚本创建模型时没有设置限流和缓存，默认调用 self._get_text_embeddings(cur_batch)。

本地源码：[HuggingFaceEmbedding._get_text_embeddings()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/embeddings/huggingface/base.py:322)，llama_index/embeddings/huggingface/base.py 第 322 行。

运行时 self：HuggingFaceEmbedding 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _get_text_embeddings(self, texts: List[str]) -> List[List[float]]:
    return self._embed(texts, prompt_name="text")

本地源码：[HuggingFaceEmbedding._embed()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/embeddings/huggingface/base.py:248)，llama_index/embeddings/huggingface/base.py 第 248 行。

运行时 self：HuggingFaceEmbedding 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _embed(
    self,
    inputs: List[Union[str, BytesIO]],
    prompt_name: Optional[str] = None,
) -> List[List[float]]:
    return self._embed_with_retry(inputs, prompt_name)

本地源码：[HuggingFaceEmbedding._embed_with_retry()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/embeddings/huggingface/base.py:202)，llama_index/embeddings/huggingface/base.py 第 202 行。

运行时 self：HuggingFaceEmbedding 实例。

本地方法代码（仅省略方法开头的说明字符串）：

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    reraise=True,
)
def _embed_with_retry(
    self,
    inputs: List[Union[str, BytesIO]],
    prompt_name: Optional[str] = None,
) -> List[List[float]]:
    try:
        if self._parallel_process:
            pool = self._model.start_multi_process_pool(
                target_devices=self._target_devices
            )
            emb = self._model.encode_multi_process(
                inputs,
                pool=pool,
                batch_size=self.embed_batch_size,
                prompt_name=prompt_name,
                normalize_embeddings=self.normalize,
                show_progress_bar=self.show_progress_bar,
            )
            self._model.stop_multi_process_pool(pool=pool)
        else:
            emb = self._model.encode(
                inputs,
                batch_size=self.embed_batch_size,
                prompt_name=prompt_name,
                normalize_embeddings=self.normalize,
                show_progress_bar=self.show_progress_bar,
            )
        return emb.tolist()
    except Exception as e:
        logger.warning(f"Embedding attempt failed: {e!s}")
        raise

本例 parallel_process=False，执行 _embed_with_retry() 的 else，调用 SentenceTransformer.encode()，把文本计算成向量并转成列表。_embed_with_retry() 上还有源码中的重试装饰器。

返回顺序：encode() → _embed_with_retry() → _embed() → _get_text_embeddings() → get_text_embedding_batch() 汇总各模型批次 → embed_nodes() 建立 node_id → embedding 映射。

5.21 回到 VectorStoreIndex._get_node_with_embedding()：设置向量

本地源码：[VectorStoreIndex._get_node_with_embedding()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:126)，llama_index/core/indices/vector_store/base.py 第 126 行。

运行时 self：VectorStoreIndex 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _get_node_with_embedding(
    self,
    nodes: Sequence[BaseNode],
    show_progress: bool = False,
) -> List[BaseNode]:
    id_to_embed_map = embed_nodes(
        nodes, self._embed_model, show_progress=show_progress
    )

    results = []
    for node in nodes:
        embedding = id_to_embed_map[node.node_id]
        result = node.model_copy()
        result.embedding = embedding
        results.append(result)
    return results

embed_nodes() 返回后，对每个节点调用 model_copy()，把对应向量赋给副本。model_copy() 默认浅拷贝。返回带向量的节点批次，回到 _add_nodes_to_index()，继续执行 self._vector_store.add(...)。

5.22 ChromaVectorStore.add() → Collection.add()：写入集合

本地源码：[ChromaVectorStore.add()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/vector_stores/chroma/base.py:284)，llama_index/vector_stores/chroma/base.py 第 284 行。

运行时 self：ChromaVectorStore 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
    if not self._collection:
        raise ValueError("Collection not initialized")

    max_chunk_size = MAX_CHUNK_SIZE
    node_chunks = chunk_list(nodes, max_chunk_size)

    all_ids = []
    for node_chunk in node_chunks:
        embeddings = []
        metadatas = []
        ids = []
        documents = []
        for node in node_chunk:
            embeddings.append(node.get_embedding())
            metadata_dict = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )
            for key in metadata_dict:
                if metadata_dict[key] is None:
                    metadata_dict[key] = ""
            metadatas.append(metadata_dict)
            ids.append(node.node_id)
            documents.append(node.get_content(metadata_mode=MetadataMode.NONE))

        self._collection.add(
            embeddings=embeddings,
            ids=ids,
            metadatas=metadatas,
            documents=documents,
        )
        all_ids.extend(ids)

    return all_ids

按适配器批次限制分组，对每个节点取向量、序列化元数据、取节点 ID 和纯正文，再把四组数组交给 Chroma 集合。self._collection 就是脚本中的 chroma_collection。

本地源码：[node_to_metadata_dict()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/vector_stores/utils.py:40)，llama_index/core/vector_stores/utils.py 第 40 行。

本地方法代码（仅省略方法开头的说明字符串）：

def node_to_metadata_dict(
    node: BaseNode,
    remove_text: bool = False,
    text_field: str = DEFAULT_TEXT_KEY,
    text_resource_field: str = DEFAULT_TEXT_RESOURCE_KEY,
    flat_metadata: bool = False,
) -> Dict[str, Any]:
    # Using mode="json" here because BaseNode may have fields of type bytes (e.g. images in ImageBlock),
    # which would cause serialization issues.
    node_dict = node.model_dump(mode="json")
    metadata: Dict[str, Any] = node_dict.get("metadata", {})

    if flat_metadata:
        _validate_is_flat_dict(metadata)

    # store entire node as json string - some minor text duplication
    if remove_text and text_field in node_dict:
        node_dict[text_field] = ""
    if remove_text and text_resource_field in node_dict:
        del node_dict[text_resource_field]

    # remove embedding from node_dict
    node_dict["embedding"] = None

    # dump remainder of node_dict to json string
    metadata["_node_content"] = json.dumps(node_dict, ensure_ascii=False)
    metadata["_node_type"] = node.class_name()

    # store ref doc id at top level to allow metadata filtering
    # kept for backwards compatibility, will consolidate in future
    metadata["document_id"] = node.ref_doc_id or "None"  # for Chroma
    metadata["doc_id"] = node.ref_doc_id or "None"  # for Pinecone, Qdrant, Redis
    metadata["ref_doc_id"] = node.ref_doc_id or "None"  # for Weaviate

    return metadata

当前版本使用 model_dump(mode="json")，将节点信息写入 _node_content 等字段；正文和向量单独存储，序列化内容中去掉其重复部分。

本地源码：[Collection.add()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/chromadb/api/models/Collection.py:78)，chromadb/api/models/Collection.py 第 78 行。

运行时 self：chromadb.api.models.Collection.Collection 实例。

Chroma 集合内部先标准化和校验数据，再调用底层客户端 _add()。下面是 Collection.add() 的实际写入段：

源码选段，第 109–127 行：

add_request = self._validate_and_prepare_add_request(
    ids=ids,
    embeddings=embeddings,
    metadatas=metadatas,
    documents=documents,
    images=images,
    uris=uris,
)

self._client._add(
    collection_id=self.id,
    ids=add_request["ids"],
    embeddings=add_request["embeddings"],
    metadatas=add_request["metadatas"],
    documents=add_request["documents"],
    uris=add_request["uris"],
    tenant=self.tenant,
    database=self.database,
)

本地源码：[CollectionCommon._validate_and_prepare_add_request()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/chromadb/api/models/CollectionCommon.py:210)，chromadb/api/models/CollectionCommon.py 第 210 行。

运行时 self：Collection，方法继承自 CollectionCommon 实例。

_validate_and_prepare_add_request() 检查已有 embeddings，本例已经由 LlamaIndex 提供，因此使用传入的向量。底层 _add() 完成后，ChromaVectorStore.add() 汇总并返回 ID 列表，回到 _add_nodes_to_index()。

5.23 回到 _add_nodes_to_index()：判断是否额外保存节点

接着执行同一个方法的后半段：

源码选段，第 234–258 行：

if not self._vector_store.stores_text or self._store_nodes_override:
    # NOTE: if the vector store doesn't store text,
    # we need to add the nodes to the index struct and document store
    for node, new_id in zip(nodes_batch, new_ids):
        # NOTE: remove embedding from node to avoid duplication
        node_without_embedding = node.model_copy()
        node_without_embedding.embedding = None

        index_struct.add_node(node_without_embedding, text_id=new_id)
        self._docstore.add_documents(
            [node_without_embedding], allow_update=True
        )
else:
    # NOTE: if the vector store keeps text,
    # we only need to add image and index nodes
    for node, new_id in zip(nodes_batch, new_ids):
        if isinstance(node, (ImageNode, IndexNode)):
            # NOTE: remove embedding from node to avoid duplication
            node_without_embedding = node.model_copy()
            node_without_embedding.embedding = None

            index_struct.add_node(node_without_embedding, text_id=new_id)
            self._docstore.add_documents(
                [node_without_embedding], allow_update=True
            )

本例 stores_text=True、store_nodes_override=False，走 else。普通 TextNode 不满足 ImageNode / IndexNode 条件，所以不会额外保存到 docstore 和 IndexDict；正文已经提交给 Chroma。然后处理下一批节点。

全部批次结束后依次返回：_add_nodes_to_index() → _build_index_from_nodes() 返回 IndexDict → build_index_from_nodes() → BaseIndex.__init__()。

5.24 回到 BaseIndex.init()：注册结构并返回 index

现在才继续执行父类构造中的两句：

self._index_struct = index_struct
self._storage_context.index_store.add_index_struct(self._index_struct)

保存索引结构并注册到 index_store。随后父类、子类构造先后结束，BaseIndex.from_documents() 中的 cls(...) 得到构造好的对象并返回，脚本变量 index 接收结果。

当前启用的脚本执行到这里完成：文档已经切成节点、计算向量、写入 quickstart 集合。

6. 原脚本中已注释的加载与检索代码

这些代码在原脚本中处于注释状态。只有取消对应注释才会进入下列路径。

6.1 VectorStoreIndex.from_vector_store()：连接已有集合

本地源码：[VectorStoreIndex.from_vector_store()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:87)，llama_index/core/indices/vector_store/base.py 第 87 行。

原脚本先重新创建 PersistentClient，调用 get_collection("quickstart") 获取已有集合，再创建 ChromaVectorStore。最后调用 from_vector_store()：

本地方法代码（仅省略方法开头的说明字符串）：

@classmethod
def from_vector_store(
    cls,
    vector_store: BasePydanticVectorStore,
    embed_model: Optional[EmbedType] = None,
    **kwargs: Any,
) -> "VectorStoreIndex":
    if not vector_store.stores_text:
        raise ValueError(
            "Cannot initialize from a vector store that does not store text."
        )

    kwargs.pop("storage_context", None)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)

    return cls(
        nodes=[],
        embed_model=embed_model,
        storage_context=storage_context,
        **kwargs,
    )

检查向量库保存正文后，创建新的 StorageContext，用 nodes=[] 构造索引。空节点会让 _add_nodes_to_index() 直接返回，不重新切分或写入文档向量。

6.2 VectorStoreIndex.as_retriever()：创建检索器

本地源码：[VectorStoreIndex.as_retriever()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/base.py:112)，llama_index/core/indices/vector_store/base.py 第 112 行。

运行时 self：VectorStoreIndex 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def as_retriever(self, **kwargs: Any) -> BaseRetriever:
    # NOTE: lazy import
    from llama_index.core.indices.vector_store.retrievers import (
        VectorIndexRetriever,
    )

    return VectorIndexRetriever(
        self,
        node_ids=list(self.index_struct.nodes_dict.values()),
        callback_manager=self._callback_manager,
        object_map=self._object_map,
        **kwargs,
    )

返回 VectorIndexRetriever。此时只准备检索器，问题向量在 retrieve() 路径中计算。

6.3 retrieve()：按当前本地版本检索

本地源码：[BaseRetriever.retrieve()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/base/base_retriever.py:192)，llama_index/core/base/base_retriever.py 第 192 行。

运行时 self：VectorIndexRetriever，方法继承自 BaseRetriever 实例。

本地源码：[VectorIndexRetriever._retrieve()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/retrievers/retriever.py:104)，llama_index/core/indices/vector_store/retrievers/retriever.py 第 104 行。

运行时 self：VectorIndexRetriever 实例。

本地方法代码（仅省略方法开头的说明字符串）：

@dispatcher.span
def _retrieve(
    self,
    query_bundle: QueryBundle,
) -> List[NodeWithScore]:
    if self._needs_embedding():
        if query_bundle.embedding is None and len(query_bundle.embedding_strs) > 0:
            query_bundle.embedding = (
                self._embed_model.get_agg_embedding_from_queries(
                    query_bundle.embedding_strs
                )
            )
    return self._get_nodes_with_embeddings(query_bundle)

本地源码：[VectorIndexRetriever._get_nodes_with_embeddings()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/core/indices/vector_store/retrievers/retriever.py:227)，llama_index/core/indices/vector_store/retrievers/retriever.py 第 227 行。

运行时 self：VectorIndexRetriever 实例。

本地方法代码（仅省略方法开头的说明字符串）：

def _get_nodes_with_embeddings(
    self, query_bundle_with_embeddings: QueryBundle
) -> List[NodeWithScore]:
    query = self._build_vector_store_query(query_bundle_with_embeddings)
    query_result = self._vector_store.query(query, **self._kwargs)

    nodes_to_fetch = self._determine_nodes_to_fetch(query_result)
    if nodes_to_fetch:
        # Fetch any missing nodes from the docstore and insert them into the query result
        fetched_nodes: List[BaseNode] = self._docstore.get_nodes(
            node_ids=nodes_to_fetch, raise_error=False
        )

        query_result.nodes = self._insert_fetched_nodes_into_query_result(
            query_result, fetched_nodes
        )

    log_vector_store_query_result(query_result)

    return self._convert_nodes_to_scored_nodes(query_result)

本地源码：[ChromaVectorStore.query()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/vector_stores/chroma/base.py:371)，llama_index/vector_stores/chroma/base.py 第 371 行。

运行时 self：ChromaVectorStore 实例。

本地源码：[ChromaVectorStore._query()](D:/ProgramData/miniconda3/envs/py310/Lib/site-packages/llama_index/vector_stores/chroma/base.py:426)，llama_index/vector_stores/chroma/base.py 第 426 行。

运行时 self：ChromaVectorStore 实例。

当前本地版本的主要调用顺序：

BaseRetriever.retrieve()
  → 字符串问题包装为 QueryBundle
  → VectorIndexRetriever._retrieve()
    → _needs_embedding() 判断是否需要问题向量
    → get_agg_embedding_from_queries() 计算问题向量
    → _get_nodes_with_embeddings()
      → _build_vector_store_query()
      → ChromaVectorStore.query()
        → _query() → Collection.query()
        → 恢复命中节点，返回 VectorStoreQueryResult
      → _determine_nodes_to_fetch()
        → 需要时从 docstore 补取节点
      → _convert_nodes_to_scored_nodes()
        → 返回 List[NodeWithScore]
  → BaseRetriever 处理递归检索与去重等收尾
  → 返回脚本中的 response / res

普通文本节点由 Chroma 返回时通常不需要额外补取。这里返回的是命中节点与分数，没有执行 LLM 答案合成。

7. 当前启用代码的完整调用链

导入 base_llm
  → HuggingFaceEmbedding.__init__() → 创建 SentenceTransformer

SimpleDirectoryReader.__init__()
  → load_data() → load_file() → Document
  → _exclude_metadata() → 返回 documents

chromadb.PersistentClient()
  → Client.get_or_create_collection() → 返回 collection
ChromaVectorStore.__init__() → 保存 collection
StorageContext.from_defaults() → 返回 Storage_Context

BaseIndex.from_documents()                         [cls = VectorStoreIndex]
  ├─ _Settings.transformations → _Settings.node_parser
  │   → SentenceSplitter.__init__()
  ├─ docstore.set_document_hash(doc.id_, doc.hash)
  ├─ run_transformations()
  │   → NodeParser.__call__()                       [self = SentenceSplitter]
  │     → NodeParser.get_nodes_from_documents()
  │       ├─ MetadataAwareTextSplitter._parse_nodes()
  │       │   ├─ _get_metadata_str()
  │       │   ├─ SentenceSplitter.split_text_metadata_aware()
  │       │   │   → _split_text()
  │       │   │     ├─ _split() → _get_splits_by_fns() → 递归切分
  │       │   │     └─ _merge() → _postprocess_chunks()
  │       │   └─ build_nodes_from_splits() → TextNode
  │       └─ NodeParser._postprocess_parsed_nodes()
  └─ VectorStoreIndex.__init__()
      ├─ resolve_embed_model()
      └─ BaseIndex.__init__()                       [self = VectorStoreIndex]
          ├─ VectorStoreIndex.build_index_from_nodes()
          │   → _build_index_from_nodes()
          │     → _add_nodes_to_index()             [按节点批次循环]
          │       ├─ _get_node_with_embedding()
          │       │   ├─ embed_nodes()
          │       │   │   → BaseEmbedding.get_text_embedding_batch()
          │       │   │     → HuggingFaceEmbedding._get_text_embeddings()
          │       │   │       → _embed()
          │       │   │         → _embed_with_retry()
          │       │   │           → SentenceTransformer.encode()
          │       │   └─ 节点副本设置 embedding
          │       ├─ ChromaVectorStore.add()
          │       │   ├─ node_to_metadata_dict()
          │       │   └─ Collection.add()
          │       │       → _validate_and_prepare_add_request()
          │       │       → 底层客户端 _add()
          │       └─ 判断额外回写，本例普通 TextNode 跳过
          └─ index_store.add_index_struct()

BaseIndex.from_documents() 返回 VectorStoreIndex 对象
  → 脚本变量 index 接收

8. 在 IDE 中核对方法所属类

在 run_transformations() 的 transform(nodes, **kwargs) 处停下，可以查看 type(transform)；进入解析器后，在调试表达式中查看：

type(self)
type(self).__mro__
self.get_nodes_from_documents.__func__.__qualname__
self._parse_nodes.__func__.__qualname__

type(self) 确认实际对象；__qualname__ 确认当前绑定的方法定义在哪个类。源码跳转落点与运行时结果不一致时，以实际对象及其方法绑定为准。

下面的本地定位表按本文出现顺序列出已核对的方法，点击即可打开定义位置。行号对应本次读取的本地文件。