总入口 IndexingRunner.run（core/indexing_runner.py:68）——一眼看清整条流水线：

# core/indexing_runner.py:68
def run(self, dataset_documents: list[DatasetDocument]):
    for dataset_document in dataset_documents:
        try:
            dataset = db.session.get(Dataset, requeried_document.dataset_id)     # 找到所属知识库
            processing_rule = db.session.scalar(...)                             # 取处理规则(切分参数等)
            index_type = requeried_document.doc_form
            index_processor = IndexProcessorFactory(index_type).init_index_processor()  # ① 按类型选工艺
            text_docs = self._extract(index_processor, requeried_document, processing_rule.to_dict())  # ② 抽取
            documents = self._transform(index_processor, dataset, text_docs, ...) # ③ 清洗+切分
            self._load_segments(dataset, requeried_document, documents)          # ④ 存片段到业务库
            self._load(index_processor=index_processor, dataset=dataset,         # ⑤ 向量化+建索引
                       dataset_document=requeried_document, documents=documents)
        except DocumentIsPausedError:
            raise DocumentIsPausedError(...)                                     # 用户暂停：原样抛出
        except ProviderTokenNotInitError as e:
            self._handle_indexing_error(document_id, e)                          # 没配 embedding key：记错
        except Exception as e:
            self._handle_indexing_error(document_id, e)                          # 兜底：把错写进文档状态

切分是索引质量的命门。切分器由 BaseIndexProcessor._get_splitter（core/rag/index_processor/index_processor_base.py:100）按规则挑：

# core/rag/index_processor/index_processor_base.py:100
def _get_splitter(self, processing_rule_mode, max_tokens, chunk_overlap, separator,
                  embedding_model_instance) -> TextSplitter:
    if processing_rule_mode in ["custom", "hierarchical"]:      # 用户自定义规则
        max_segmentation_tokens_length = dify_config.INDEXING_MAX_SEGMENTATION_TOKENS_LENGTH
        if max_tokens < 50 or max_tokens > max_segmentation_tokens_length:
            raise ValueError(f"Custom segment length should be between 50 and {max_segmentation_tokens_length}.")
        character_splitter = FixedRecursiveCharacterTextSplitter.from_encoder(
            chunk_size=max_tokens, chunk_overlap=chunk_overlap,
            fixed_separator=separator,
            separators=["\n\n", "。", ". ", " ", ""],           # ★ 优先在段落/句号处断
            embedding_model_instance=embedding_model_instance)
    else:                                                       # 自动模式：用内置默认参数
        character_splitter = EnhanceRecursiveCharacterTextSplitter.from_encoder(
            chunk_size=DatasetProcessRule.AUTOMATIC_RULES["segmentation"]["max_tokens"],
            chunk_overlap=DatasetProcessRule.AUTOMATIC_RULES["segmentation"]["chunk_overlap"],
            separators=["\n\n", "。", ". ", " ", ""],
            embedding_model_instance=embedding_model_instance)
    return character_splitter

以最常见的"普通段落"工艺为例，看 ParagraphIndexProcessor.transform（core/rag/index_processor/processor/paragraph_index_processor.py:75）：

# core/rag/index_processor/processor/paragraph_index_processor.py:75
def transform(self, documents, current_user=None, **kwargs) -> list[Document]:
    process_rule = kwargs.get("process_rule")
    rules = Rule.model_validate(...)                    # 解析切分规则
    splitter = self._get_splitter(...)                  # ① 拿到 L04 的切分器
    all_documents = []
    for document in documents:
        document_text = CleanProcessor.clean(document.page_content, kwargs.get("process_rule", {}))  # ② 清洗
        document.page_content = document_text
        document_nodes = splitter.split_documents([document])   # ③ ★真正切分
        split_documents = []
        for document_node in document_nodes:
            if document_node.page_content.strip():
                doc_id = str(uuid.uuid4())
                hash = helper.generate_text_hash(document_node.page_content)  # ④ 内容指纹
                document_node.metadata["doc_id"] = doc_id
                document_node.metadata["doc_hash"] = hash
                page_content = remove_leading_symbols(document_node.page_content).strip()
                if len(page_content) > 0:
                    document_node.page_content = page_content
                    split_documents.append(document_node)
        all_documents.extend(split_documents)
    return all_documents

片段切好后，ParagraphIndexProcessor.load（core/rag/index_processor/processor/paragraph_index_processor.py:125）负责把它们送进索引：

# core/rag/index_processor/processor/paragraph_index_processor.py:125
def load(self, dataset, documents, multimodal_documents=None, with_keywords=True, **kwargs):
    if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:   # 高质量：走向量
        vector = Vector(dataset)
        vector.create(documents)                        # ★ 向量化 + 写进向量数据库
        if multimodal_documents and dataset.is_multimodal:
            vector.create_multimodal(multimodal_documents)
        with_keywords = False
    if with_keywords:                                    # 经济模式：走关键词索引（不花 embedding 钱）
        keywords_list = kwargs.get("keywords_list")
        keyword = Keyword(dataset)
        if keywords_list and len(keywords_list) > 0:
            keyword.add_texts(documents, keywords_list=keywords_list)
        else:
            keyword.add_texts(documents)

而真正调 embedding 模型、并发向量化的编排在 IndexingRunner._load（core/indexing_runner.py:573）里：

# core/indexing_runner.py:573
def _load(self, index_processor, dataset, dataset_document, documents):
    embedding_model_instance = None
    if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
        embedding_model_instance = self._get_model_manager(dataset.tenant_id).get_model_instance(
            tenant_id=dataset.tenant_id, provider=dataset.embedding_model_provider,
            model_type=ModelType.TEXT_EMBEDDING, model=dataset.embedding_model)   # ★ 拿 embedding 模型
    ...
    max_workers = 10
    if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            document_groups = [[] for _ in range(max_workers)]
            for document in documents:
                hash = helper.generate_text_hash(document.page_content)
                group_index = int(hash, 16) % max_workers        # ★ 按内容 hash 分组，避免同片撞车
                document_groups[group_index].append(document)

存片段那一步 _load_segments（core/indexing_runner.py:816）除了写库，还推进文档的状态机：

# core/indexing_runner.py:816
def _load_segments(self, dataset, dataset_document, documents):
    doc_store = DatasetDocumentStore(dataset=dataset, user_id=dataset_document.created_by,
                                     document_id=dataset_document.id)
    doc_store.add_documents(docs=documents, save_child=...)      # 片段写进 document_segments 表
    self._update_document_index_status(                          # ★ 文档状态 → INDEXING
        document_id=dataset_document.id,
        after_indexing_status=IndexingStatus.INDEXING,
        extra_update_params={..., DatasetDocument.word_count: sum(len(d.page_content) for d in documents)})
    self._update_segments_by_document(                           # ★ 每个片段状态 → INDEXING
        dataset_document_id=dataset_document.id,
        update_params={DocumentSegment.status: SegmentStatus.INDEXING, ...})

cd /Users/bitmart/work/codes/github/AI_WORK/dify/api

# 1. 主流水线
sed -n '68,127p'  core/indexing_runner.py                 # run() 五步
sed -n '780,814p' core/indexing_runner.py                 # _transform（挑 embedding 模型）
sed -n '573,620p' core/indexing_runner.py                 # _load（并发向量化）

# 2. 切分与转换
sed -n '100,137p' core/rag/index_processor/index_processor_base.py     # _get_splitter
sed -n '75,145p'  core/rag/index_processor/processor/paragraph_index_processor.py  # transform + load

# 3. 有哪些处理工艺
ls core/rag/index_processor/processor/