杂乱数据如何变成一张可解释、可审计的知识图谱 3 / 10

第 3 章

解析、归一化与切块

源码核对基于 semantica-agi/semantica commit `1ee2ae88`,tag `course-anchor-20260824`

场景还原

企业知识库上线第三周,审计组发现一条坏事实:「张三 于 2023 年离职」。系统里存的却是一条相反的边,把「离职」接到了「李四」头上。查来查去,问题出在切块。原始文档是一段 4000 字的人事说明,管道用固定窗口每 1000 字符切一刀,第三刀正好落在「张三」和「于 2023 年离职」之间。张三进了上一块,离职动作进了下一块,LLM 抽取时两块各自抽,主语对错了。审计要追责时更麻烦:这条坏边是从哪份原件、哪一段文字来的,翻遍了日志都对不回原文。

这一章沿数据流走 parse、normalize、split 三段。核心问题只有一个:把非结构化文本切成块时,怎么保证切刀不落在语义边界上,以及每一块怎么保留「我从哪里来」的证据。先看解析层怎么把几十种文件收敛成一个入口。

逐行精读

三段管道的全景先画出来,后面每一段对应一个工位:

flowchart LR A[原始文档] --> B[DocumentParser 按扩展名路由] B --> C[TextNormalizer 四步归一化] C --> D[TextSplitter 选方法] D --> E[具体 split 函数切块] E --> F[Chunk 携带 start_index 与 end_index] F --> G[ProvenanceTracker 登记来源]

解析:一个入口,按扩展名路由

DocumentParser 是解析层的门面,它自己不做任何格式解析,只在构造时按格式装好三个具体解析器,并维护一张扩展名到格式名的映射表:

semantica/parse/document_parser.py84:98
84        # Initialize format-specific parsers85        self.pdf_parser = PDFParser(**self.config.get("pdf", {}))86        self.docx_parser = DOCXParser(**self.config.get("docx", {}))87        self.html_parser = HTMLParser(**self.config.get("html", {}))8889        # Supported formats90        self.supported_formats = {91            ".pdf": "pdf",92            ".docx": "docx",93            ".doc": "docx",94            ".html": "html",95            ".htm": "html",96            ".txt": "text",97            ".text": "text",98        }

这张表证明了一件事:解析层把「格式差异」挡在了门面之内。.doc.docx 都归一成 docx.htm.html 都归一成 html,下游看到的永远是一个格式标签。输入是文件路径,输出是 pdfdocxhtmltext 四类之一。真正的分派在 parse_document 里:

semantica/parse/document_parser.py183:194
183            # Route to appropriate parser184            try:185                if file_type == "pdf":186                    result = self.pdf_parser.parse(file_path, **options)187                elif file_type == "docx":188                    result = self.docx_parser.parse(file_path, **options)189                elif file_type == "html":190                    result = self.html_parser.parse(file_path, **options)191                elif file_type == "text":192                    result = self._parse_text(file_path, **options)193                else:194                    raise ValidationError(f"Unsupported document format: {file_type}")

这段 if/elif 链就是全部路由逻辑。值得注意的细节是 else 分支不是静默跳过,而是抛 ValidationError,把「不认识的格式」变成显式错误。输入是 file_path 和可选的 file_type,输出是具体解析器产出的 Dict[str, Any],里面至少带 textmetadata。解析层到这里为止,它负责把文件变成字符串,不负责把字符串变干净。解析层还有一层单文件与批量的统一入口:parseisinstance(source, list) 判断,传列表就拆出路径交给 parse_batch,传单路径就交给 parse_document(document_parser.py:122、134)。批量的容错开关在 continue_on_error(document_parser.py:306),默认 True 时坏文件只记进 failed 列表不打断整批,设为 False 时第一个坏文件就抛错。这个开关决定一批 100 个文件坏 1 个时,管道是继续跑还是停下来。

归一化:切块之前的四道工序

TextNormalizer.normalize_text 是切块前的清洗工位。它按固定顺序跑四步,每一步的输入是上一步的输出:

semantica/normalize/text_normalizer.py164:190
164            normalized = text165166            # Unicode normalization167            normalized = self.unicode_normalizer.normalize_unicode(168                normalized, form=unicode_form169            )170171            # Whitespace normalization172            normalized = self.whitespace_normalizer.normalize_whitespace(173                normalized, line_break_type=line_break_type, **options174            )175176            # Special character processing177            normalized = self.special_char_processor.process_special_chars(178                normalized, normalize_diacritics=normalize_diacritics, **options179            )180181            # Case normalization182            if case == "lower":183                normalized = normalized.lower()184            elif case == "upper":185                normalized = normalized.upper()186            elif case == "title":187                normalized = normalized.title()188189            self.progress_tracker.stop_tracking(tracking_id, status="completed")190            return normalized.strip()

顺序是刻意排的:先做 Unicode 归一化,把「同一个字符的多种字节写法」压成一种;再做空白归一化,把制表符、混合换行、连续空格收拢;然后处理特殊字符,把弯引号、破折号换成 ASCII;最后才动大小写。如果先小写再处理 Unicode,某些字符的 case 映射会在归一化后漂移。默认 case="preserve",也就是前三步照跑、第四步不动,最后的 normalized.strip() 保证边界没有残留空白。这个顺序直接决定切块时的字符偏移是否稳定。四个子处理器的共同点是失败不炸:normalize_unicode 在异常时返回原文本(text_normalizer.py:314),handle_encoding 解码失败时退回 UTF-8 加替换符(text_normalizer.py:342)。归一化追求把常见脏数据压成一致形态,接受结果仍可能残留个别异常字符,所以每一步都留了退回原样的后路。代价是偶尔漏掉一个坏字符,收益是整条管道不会因为一个坏字符停摆。

切分入口:方法链与回退

TextSplitter 是所有切分策略的统一门面。它的核心不在切分本身,而在「方法链 + 回退」:

semantica/split/splitter.py115:176
115    def split(self, text: str, **override_options) -> List[Chunk]:116        """117        Split text into chunks using the specified method(s).118119        Args:120            text: Input text to split121            **override_options: Options to override for this split call122123        Returns:124            List of Chunk objects125126        Raises:127            ProcessingError: If all methods fail128        """129        if not text:130            return []131132        # Merge override options133        options = {**self.options, **override_options}134        options["chunk_size"] = options.get("chunk_size", self.chunk_size)135        options["chunk_overlap"] = options.get("chunk_overlap", self.chunk_overlap)136137        # Try each method in fallback chain138        last_error = None139        for method_name in self.methods:140            try:141                self.logger.debug(f"Attempting to split using method: {method_name}")142143                # Get method function144                method_func = get_split_method(method_name)145                if not method_func:146                    self.logger.warning(147                        f"Method '{method_name}' not found, trying next method"148                    )149                    continue150151                # Call method with text and options152                chunks = method_func(text, **options)153154                if chunks:155                    self.logger.info(156                        f"Successfully split text into {len(chunks)} chunks using method: {method_name}"157                    )158                    return chunks159                else:160                    self.logger.warning(161                        f"Method '{method_name}' returned no chunks, trying next method"162                    )163164            except Exception as e:165                self.logger.warning(166                    f"Method '{method_name}' failed: {e}, trying next method"167                )168                last_error = e169                continue170171        # All methods failed172        error_msg = f"All splitting methods failed: {', '.join(self.methods)}"173        if last_error:174            error_msg += f". Last error: {last_error}"175176        raise ProcessingError(error_msg)

输入是一个 text 字符串,输出是 List[Chunk]。构造时可以传一个方法名,也可以传一个方法名列表:列表就是回退链。循环里三个失败出口都走 continue:方法名没注册、切出来是空列表、方法抛异常,全部滑到下一个方法。只有全部方法都失败,才抛 ProcessingError。这个结构说明作者把「某个切分器可能坏」当成了常态,全部失败才升级成显式异常。

回退链的时序画出来,三个失败出口都回到循环头:

sequenceDiagram participant U as 调用方 participant T as TextSplitter participant R as get_split_method participant M as split_entity_aware U->>T: 传入 text 与方法名 entity_aware T->>R: get_split_method entity_aware R-->>T: 返回函数或 None T->>M: 调用 method_func 传入 text alt 返回非空 chunks M-->>T: 返回 List 的 Chunk T-->>U: 直接返回 else 返回空或抛异常 M-->>T: 返回空列表或抛异常 T->>T: 取 methods 下一项 end

递归切分:分隔符优先级表

最基础的 split_recursive 是其它方法的兜底,它的核心是一张分隔符优先级表和一次反向查找:

semantica/split/methods.py182:231
182    if separators is None:183        separators = ["\n\n", "\n", ". ", " ", ""]184185    chunks = []186    text_length = len(text)187    start = 0188189    while start < text_length:190        # Find the best split point191        end = start + chunk_size192        if end >= text_length:193            chunk_text = text[start:]194            chunks.append(195                Chunk(196                    text=chunk_text,197                    start_index=start,198                    end_index=text_length,199                    metadata={"method": "recursive", "chunk_size": len(chunk_text)},200                )201            )202            break203204        # Try each separator in priority order205        split_pos = -1206        for separator in separators:207            if separator:208                pos = text.rfind(separator, start, end)209                if pos > start + chunk_size * 0.5:  # At least 50% of target size210                    split_pos = pos + len(separator)211                    break212            else:213                # Last resort: split at character boundary214                split_pos = end215216        if split_pos == -1:217            split_pos = end218219        chunk_text = text[start:split_pos].strip()220        if chunk_text:221            chunks.append(222                Chunk(223                    text=chunk_text,224                    start_index=start,225                    end_index=split_pos,226                    metadata={"method": "recursive", "chunk_size": len(chunk_text)},227                )228            )229230        # Move to next chunk with overlap231        start = max(start + 1, split_pos - chunk_overlap)

这张表从粗到细:先找双换行(段落),再找单换行,再找句号加空格,再找空格,最后连空格都没有就硬切。rfindend 往前找,找到的分隔符位置还要满足「至少越过目标尺寸的 50%」才接受,这个 0.5 阈值防止在窗口开头一点点就切,产出碎块。切完一块后 start 回退 chunk_overlap 个字符,让相邻块带上重叠。每一块都带着 start_indexend_index 这两个原始坐标。rfind 从右往左找,同一窗口里优先选靠后的分隔符,切出的上一块尽量饱满;方向换成从左往右的 find,会优先选靠前的分隔符,上一块变短、下一块变长。这个方向选择直接影响块的大小分布,也影响重叠区落在句号前还是句号后。除递归外,methods.py 还备了一串通用方法,各自有兜底:split_by_tokens(methods.py:236)优先 tiktoken,再退 transformers,再退「按词近似、1 token 约 4 字符」;split_by_sentences(methods.py:321)先试 spaCy,再按需 NLTK,最后退正则;split_semantic_transformer(methods.py:588)用句向量余弦相似度找语义断点,阈值默认 0.7,模型装不上就退回按句切。这些方法共同点是把「依赖可用」和「切分正确」拆开:依赖缺了不报错,降级到更简单的切法。

块的数据形状:坐标即溯源

所有切分方法返回的同一种对象是 Chunk,它是一个极简 dataclass:

semantica/split/semantic_chunker.py43:50
43@dataclass44class Chunk:45    """Chunk representation."""46    text: str47    start_index: int48    end_index: int49    metadata: Dict[str, Any] = field(default_factory=dict)50    id: Optional[str] = None

六个字段,两个关键:start_indexend_index 是这块文字在原文中的字符区间。块可以丢 metadata,可以没有 id,但必须知道自己在原文的哪一段。这就是溯源的最小形态:一张「块 → 原文区间」的映射,在任何语言里都只是一对整数。id 默认是 None,切分函数不负责生成它,生成动作在 ProvenanceTracker.track_chunk:拿不到 chunk.id 就用 uuid4() 现造一个(provenance_tracker.py:139-141)。「块的坐标」和「块的身份」是两件事,坐标切分时就要有,身份登记溯源时才补。

实体感知:句子含实体边界就不切

EntityAwareChunker 是 KG 导向 chunker 里的第一个,整类都只是把参数转交给底层函数:

semantica/split/kg_chunkers.py62:137
62class EntityAwareChunker:63    """64    Entity boundary-preserving chunker for GraphRAG workflows.6566    Ensures that entities and their associated information are kept together,67    preserving the semantic integrity necessary for accurate graph-based retrieval.68    """6970    def __init__(71        self,72        chunk_size: int = 1000,73        chunk_overlap: int = 200,74        ner_method: str = "ml",75        preserve_entities: bool = True,76        **kwargs,77    ):78        """79        Initialize entity-aware chunker.8081        Args:82            chunk_size: Target chunk size83            chunk_overlap: Overlap between chunks84            ner_method: NER method to use ("pattern", "regex", "ml", "huggingface", "llm")85            preserve_entities: Whether to preserve entity boundaries86            **kwargs: Additional options for NER extractor87        """88        self.chunk_size = chunk_size89        self.chunk_overlap = chunk_overlap90        self.ner_method = ner_method91        self.preserve_entities = preserve_entities92        self.options = kwargs93        self.logger = get_logger("entity_aware_chunker")94        self.progress_tracker = get_progress_tracker()95        # Ensure progress tracker is enabled96        if not self.progress_tracker.enabled:97            self.progress_tracker.enabled = True9899    def chunk(self, text: str, **options) -> List[Chunk]:100        """101        Chunk text preserving entity boundaries.102103        Args:104            text: Input text105            **options: Additional options106107        Returns:108            List of chunks109        """110        tracking_id = self.progress_tracker.start_tracking(111            module="split",112            submodule="EntityAwareChunker",113            message="Chunking text with entity awareness",114        )115116        try:117            merged_options = {**self.options, **options}118            chunks = split_entity_aware(119                text,120                chunk_size=self.chunk_size,121                ner_method=self.ner_method,122                preserve_entities=self.preserve_entities,123                **merged_options,124            )125126            self.progress_tracker.stop_tracking(127                tracking_id,128                status="completed",129                message=f"Created {len(chunks)} entity-aware chunks",130            )131            return chunks132133        except Exception as e:134            self.progress_tracker.stop_tracking(135                tracking_id, status="failed", message=str(e)136            )137            raise

真正的判断逻辑在 split_entity_aware 里。它先跑 NER 拿到实体的起止字符坐标,组成一个边界集合,然后逐句检查:如果当前句子覆盖了某个实体边界,就不在这里切:

semantica/split/methods.py903:920
903            # Check if sentence contains entity boundaries904            has_entity_boundary = any(905                sentence_start <= boundary <= sentence_end906                for boundary in entity_boundaries907            )908909            # Check size limit910            if current_size + sentence_size > chunk_size and current_chunk:911                # Try to split at entity boundary if possible912                if preserve_entities and has_entity_boundary:913                    # Don't split here, add to current chunk914                    pass915                else:916                    # Create chunk917                    chunk_text = current_chunk.strip()918                    text_pos = text.find(chunk_text[:50], text_start)919                    if text_pos == -1:920                        text_pos = text_start

输入是 text,输出是 List[Chunk],每个 Chunk 的 metadata 里带 entity_countentities。关键决策点是 if preserve_entities and has_entity_boundary 分支里那个 pass:当句子已经越界、但句子里躺着实体边界时,选择「不切、继续往里塞」,哪怕这一块超过 chunk_size。这就是「语义边界优先于尺寸边界」在代码里的具体落点。产出块里还带了抽取结果本身:entity_count 统计落在块区间内的实体数,entities 直接把实体对象塞进 metadata(methods.py:931、938),下游不用重新抽一遍就能知道每块含哪些实体。另一个细节是整段逻辑包在 try/except 里,任何一步抛异常都退回 split_recursive。所以 entity_aware 是「尽力保护边界,保护不了就退回通用切分」,语义质量降级,但管道不断。

关系感知:三元组不被拆开

实体感知保的是「实体名不被切开」,关系感知保的是「主语-谓语-宾语不被拆进两块」。先算每个三元组覆盖的字符区间:

semantica/split/methods.py1027:1031
1027        triplet_boundaries = []1028        for relation in relations:1029            start = min(relation.subject.start_char, relation.object.start_char)1030            end = max(relation.subject.end_char, relation.object.end_char)1031            triplet_boundaries.append((start, end))

主语的起点和宾语的终点之间,就是这条三元组的「领地」。逐句切分时,只要句子落在这块领地里,就越界也不切:

semantica/split/methods.py1047:1058
1047            # Check if sentence is part of a triplet1048            is_in_triplet = any(1049                start <= sentence_start <= end or start <= sentence_end <= end1050                for start, end in triplet_boundaries1051            )10521053            # Check size limit1054            if current_size + sentence_size > chunk_size and current_chunk:1055                # Don't split if it would break a triplet1056                if preserve_triplets and is_in_triplet:1057                    # Add to current chunk even if it exceeds size1058                    pass

和实体感知同构:越界时先问「切了会不会拆散一条关系」,会就 pass,把句子塞进当前块。两块逻辑的差别只在前置工序:实体感知只跑 NER,关系感知要先跑 NER 再跑关系抽取,关系抽取依赖实体坐标。

图结构切分:社区就是块的边界

split_graph_based 走得更远:它不满足于「不切烂实体和关系」,而是先把文本抽成一张图,再用图算法决定块怎么划:

semantica/split/methods.py1189:1214
1189        G = nx.Graph()1190        entity_map = {}11911192        for entity in entities:1193            G.add_node(entity.text, type="entity", label=entity.label)1194            entity_map[entity.text] = entity11951196        for relation in relations:1197            subject_text = relation.subject.text1198            object_text = relation.object.text1199            if subject_text in entity_map and object_text in entity_map:1200                G.add_edge(subject_text, object_text, label=relation.predicate)12011202        if len(G.nodes()) == 0:1203            return split_recursive(text, chunk_size=chunk_size, **kwargs)12041205        # Apply graph-based strategy1206        if strategy == "community":1207            if algorithm == "louvain" and COMMUNITY_AVAILABLE:1208                communities = community_louvain.best_partition(G)1209            else:1210                # Fallback to simple connected components1211                communities = {}1212                for i, component in enumerate(nx.connected_components(G)):1213                    for node in component:1214                        communities[node] = i

输入是 text,中间产物是一张 networkx.Graph,节点是实体、边是关系。切块策略两种:community 用 Louvain 社区检测,把同一社区的实体的上下文文本聚成一块;社区包不可用时退到连通分量。centrality 策略则按中心度排序,围绕高中心度节点取 k 跳邻居的上下文。注意 if len(G.nodes()) == 0 这一行:抽不出任何节点时,直接退回递归切分。图切分在空图上没有意义,这一行就是那条明确的降级路径。centrality 策略走了另一条路:先算节点中心度(betweenness、degree 或 degree_centrality,methods.py:1267-1271),按中心度从高到低排序,再围绕每个高中心度节点取 k 跳邻居(默认 k_hop=2,methods.py:1285),把邻居实体的上下文文本拼成一块。社区策略回答「哪些实体天然抱团」,中心度策略回答「哪些实体最重要」,两种策略产出的块边界逻辑完全不同,但共享同一张图和同一个兜底。

溯源:块怎么指回原文

切块不附带来源,前面所有坐标就白留了。ProvenanceTracker 负责把「块」登记成「来源记录」,它的最小数据形状是 ProvenanceInfo

semantica/split/provenance_tracker.py57:69
57@dataclass58class ProvenanceInfo:59    """Provenance information representation."""6061    chunk_id: str62    source_document: str63    source_path: Optional[str] = None64    start_index: int = 065    end_index: int = 066    parent_chunk_id: Optional[str] = None67    metadata: Dict[str, Any] = field(default_factory=dict)68    version: str = "1.0"69    timestamp: Optional[str] = None

chunk_id 定位块,source_documentsource_path 定位文件,start_indexend_index 定位区间,parent_chunk_id 串起父子链。一条块被再次切分时,新块的 parent_chunk_id 指回旧块,于是能往上追:

semantica/split/provenance_tracker.py375:388
375    def _get_chunk_lineage_legacy(self, chunk_id: str) -> List[ProvenanceInfo]:376        """Legacy get chunk lineage implementation."""377        lineage = []378        current_chunk_id = chunk_id379380        while current_chunk_id:381            provenance = self._get_provenance_legacy(current_chunk_id)382            if not provenance:383                break384385            lineage.insert(0, provenance)386            current_chunk_id = provenance.parent_chunk_id387388        return lineage

这个 while 循环就是溯源链的完整实现:从当前块出发,沿 parent_chunk_id 一路往上跳,每跳一步把记录插到列表头部,最后返回「最老祖先到最新后代」的链。输入是 chunk_id,输出是 List[ProvenanceInfo]。审计场景里「这条坏边从哪来」的答案,就是沿着这条链走到头的那条 source_documentstart_indextrack_chunk 还有一个统一后端的分支:构造时若 use_unified 且能导入 semantica.provenance.ProvenanceManager,就委托给统一管理器(provenance_tracker.py:104-110),否则退回内存里的 _provenance_store 字典。统一后端走 W3C PROV-O 那一套,传统后端就是两个 dict。溯源能力分了档:想轻量跑就内存字典,想审计级就挂统一管理器,接口对上层完全一致。

设计决策分析

为什么要有实体感知、关系感知、图结构、本体感知、层级五种 KG chunker?答案是它们各自护住一层语义边界。实体感知护住实体名,关系感知护住三元组,图结构护住社区,层级护住章节结构。护住的层不同,切刀能落的地方就不同。docs/modules.md 把这一层定位成「Chunks text for embedding and RAG pipelines with awareness of semantic boundaries」,docs/glossary.md 给 Chunking 的定义是「Splitting large documents into smaller pieces while preserving semantic context」。两条规范性材料都把「preserving semantic context」当作这一层的验收标准,五种 chunker 就是这条标准的五种实现。glossary 里 Chunking 和 Normalization 的定义原文如下:

docs/glossary.md40:47
40**Chunking**41Splitting large documents into smaller pieces while preserving semantic context. Semantica supports recursive, semantic boundary, entity-aware, relation-aware, sliding window, structural, and table-aware chunking strategies.4243**Ingestion**44Loading data from external sources: files, databases, APIs, streams: into the pipeline as a unified `SourceDocument`. The first stage in every Semantica pipeline.4546**Normalization**47Standardizing data into a consistent canonical form: converting dates to ISO format, canonicalizing entity names, fixing encoding issues, stripping noise. Ensures downstream extraction works on clean, consistent text.

为什么默认档位给了 recursive,没给 entity_aware?看 TextSplitter.__init__ 的默认参数,method="recursive"。原因在成本:entity_aware 每次切分都要先跑一遍 NER,关系感知还要再跑关系抽取,图结构更要先建图。对一段 4000 字的文本,recursive 是纯字符串操作,entity_aware 要调用一次模型。语义保护有价,默认档位留给零依赖的递归,需要时再显式升级。这是从默认参数和调用链反推的成本权衡,spec 和 docs 没有明说,属于推断。

provenance_tracker.py 存在的原因在注释里写得直白:它把数据血缘维护到「chunking process」全程,并接入了统一的 semantica.provenance.ProvenanceManager。块必须能指回原文,这条不是锦上添花,是整本书的暗线「溯源」在切块这一环的落地。一个块丢了 start_index,下游抽出的每一条事实就都成了无源之水。

再看这一层的组织方式。split/ 目录下职责分得很清:config.py 管配置,_load_env_varsSPLIT_CHUNK_SIZE 等环境变量转成参数(config.py:91-95);registry.py 管自定义方法的注册与查找;methods.py 放纯函数,kg_chunkers.py 放薄的类包装;splitter.py 是统一门面。纯函数负责算法,类包装负责参数和进度追踪,门面负责方法链。这个三层结构把「新增一个切分方法」的成本压到只写一个函数、再在 _SPLIT_METHODS 表登记一行。

至于为什么归一化在前、切块在后:切块的坐标依赖字符偏移,偏移依赖字符串稳定。同一份文本如果先切块再分别归一化,两块之间的空白、Unicode 写法可能被各自处理得不一致,块边界的 start_indexend_index 就对不齐原文了。先归一化再切块,保证切块看到的是唯一一种字节形态。这是从代码顺序能直接读出的结论,spec 和 docs 没有明写这条,属于推断。

边界条件剖析

如果传进来的 text 是空字符串会怎样? TextSplitter.split 在方法链之前先判空,直接返回空列表:

semantica/split/splitter.py129:130
129        if not text:130            return []

空文本不会走到任何切分器,也不会触发回退链,下游拿到 [],连一个含空字符串的单元素块都不会出现。

如果 NER 或关系抽取器不可用会怎样? split_entity_aware 开头检查 SEMANTIC_EXTRACT_AVAILABLE,语义抽取模块没装上就退回递归切分:

semantica/split/methods.py860:876
860    Entity boundary-preserving splitting.861862    Args:863        text: Input text864        chunk_size: Target chunk size865        ner_method: NER method to use866        preserve_entities: Whether to preserve entity boundaries867        **kwargs: Additional options868869    Returns:870        List of chunks871    """872    if not SEMANTIC_EXTRACT_AVAILABLE:873        logger.warning(874            "semantic_extract not available, falling back to recursive splitting"875        )876        return split_recursive(text, chunk_size=chunk_size, **kwargs)

split_graph_based 的门槛更严,要求语义抽取和 networkx 同时可用,缺一个就退:

semantica/split/methods.py1170:1174
1170    if not SEMANTIC_EXTRACT_AVAILABLE or not NETWORKX_AVAILABLE:1171        logger.warning(1172            "Required dependencies not available, falling back to recursive splitting"1173        )1174        return split_recursive(text, chunk_size=chunk_size, **kwargs)

也就是说,KG 切块的高级能力是「尽力而为」,底座永远落在 split_recursive。这对生产部署意味着:装不上模型也能跑,只是切块质量退到通用档。

如果坐标体系自己打架会怎样? 这是我在逐行核对时发现的一处隐患。split_entity_aware 里实体坐标来自 NER 的 entity.start_charentity.end_char,是原文坐标;而 sentence_startsentence_end 来自 char_pos 累加:

semantica/split/methods.py895:907
895        sentences = _split_sentences_regex(text)896        char_pos = 0897898        for sentence in sentences:899            sentence_size = len(sentence)900            sentence_start = char_pos901            sentence_end = char_pos + sentence_size902903            # Check if sentence contains entity boundaries904            has_entity_boundary = any(905                sentence_start <= boundary <= sentence_end906                for boundary in entity_boundaries907            )

_split_sentences_regex 用的是 re.split(r"(?<=[.!?])\s+", text),切分时把句号后的空白吃掉了:

semantica/split/methods.py421:425
421def _split_sentences_regex(text: str) -> List[str]:422    """Fallback regex-based sentence splitting."""423    # Simple sentence splitting424    sentences = re.split(r"(?<=[.!?])\s+", text)425    return [s.strip() for s in sentences if s.strip()]

两套坐标的零点虽然都是 0,但句子坐标按「去掉分隔空白后的长度」累加,会越走越偏。实体坐标是原文坐标,句子坐标是压缩坐标,sentence_start <= boundary <= sentence_end 这个判断用的是两套刻度混着比。作者在产出 Chunk 时用 text.find(chunk_text[:50], text_start) 把块坐标重新对回原文,所以最终块的 start_indexend_index 是对的;但「这个句子含不含实体边界」的判断本身可能误判,导致某些本该硬切的点被放过、或反过来。这是读代码能直接看出的两套坐标并存,至于它在真实语料上的影响有多大,需要构造长文本实测,属于推断。

横向对比

同一道「把文本切成块」的题,GraphRAG 的答案薄得多。它的 ChunkerType 枚举只有两种策略:

packages/graphrag-chunking/graphrag_chunking/chunk_strategy_type.py9:13
9class ChunkerType(StrEnum):10    """ChunkerType class definition."""1112    Tokens = "tokens"13    Sentence = "sentence"

默认配置是 token 固定窗口,1200 token 一块、100 token 重叠:

packages/graphrag-chunking/graphrag_chunking/chunking_config.py11:36
11class ChunkingConfig(BaseModel):12    """Configuration section for chunking."""1314    model_config = ConfigDict(extra="allow")15    """Allow extra fields to support custom cache implementations."""1617    type: str = Field(18        description="The chunking type to use.",19        default=ChunkerType.Tokens,20    )21    encoding_model: str | None = Field(22        description="The encoding model to use.",23        default=None,24    )25    size: int = Field(26        description="The chunk size to use.",27        default=1200,28    )29    overlap: int = Field(30        description="The chunk overlap to use.",31        default=100,32    )33    prepend_metadata: list[str] | None = Field(34        description="Metadata fields from the source document to prepend on each chunk.",35        default=None,36    )

切块实现就是一段滑窗,start_idx 每次前进 chunk_size - chunk_overlap

packages/graphrag-chunking/graphrag_chunking/token_chunker.py45:69
45def split_text_on_tokens(46    text: str,47    chunk_size: int,48    chunk_overlap: int,49    encode: Callable[[str], list[int]],50    decode: Callable[[list[int]], str],51) -> list[str]:52    """Split a single text and return chunks using the tokenizer."""53    result = []54    input_tokens = encode(text)5556    start_idx = 057    cur_idx = min(start_idx + chunk_size, len(input_tokens))58    chunk_tokens = input_tokens[start_idx:cur_idx]5960    while start_idx < len(input_tokens):61        chunk_text = decode(list(chunk_tokens))62        result.append(chunk_text)  # Append chunked text as string63        if cur_idx == len(input_tokens):64            break65        start_idx += chunk_size - chunk_overlap66        cur_idx = min(start_idx + chunk_size, len(input_tokens))67        chunk_tokens = input_tokens[start_idx:cur_idx]6869    return result

这段循环里没有任何实体、句子、关系概念,边界落在词中间就硬切。双侧对照很清晰:GraphRAG 在切块阶段不做语义保护,Semantica 在切块阶段把 NER、关系抽取、图算法都搬进来护边界。

双侧的边界决策画成对照图,左边是硬切,右边是传感器抬刀:

flowchart TB subgraph G1[GraphRAG 固定窗口] A1[token 计数到 1200] --> A2[overlap 100 滑窗] A2 --> A3[边界落在词中间也硬切] end subgraph G2[Semantica 实体感知] B1[先跑 NER 取实体坐标] --> B2[句子含实体边界则不在此切] B2 --> B3[块坐标用 text.find 对回原文] end

为什么 GraphRAG 可以只有两种?因为它的输入语料是「已整理的文本语料」,切块之后紧接着是一整条 LLM 抽取流水线,边界切烂的损失由下游的实体关系抽取和 prepend_metadata(把文档元数据前插到每块)来补。它没有企业多源、多格式的解析负担,切块阶段自然可以退化成单纯的 token 滑窗。Semantica 面向的语料更乱、来源更多,它选择在切块阶段就吃掉语义边界问题,把代价前置。prepend_metadata 补的具体是什么,看 create_base_text_units.pychunk_document:它把文档的 id、title、creation_date 收集起来,用 add_metadata 拼到每个块开头,再交给 chunker。块内的语义断了,块外的文档身份还在,下游 LLM 抽取时仍知道这块出自哪份文件。这补的是溯源身份,补不了语义边界,实体被硬切的问题依然存在,只是被下游的抽取兜住。检索证据:GraphRAG 切块相关代码集中在 packages/graphrag-chunking/graphrag_chunking/create_base_text_units.pychunker.chunk(doc["text"], transform=transformer) 是唯一调用点;Semantica 侧 methods.py_SPLIT_METHODS 表登记了 22 个方法名,其中 KG 类占一半。

互动演示设计

形态是格式实验台。读者把一段脏文本倒进左边漏斗,右边逐级亮起三个工位:归一化、切分、溯源。

一句话结论:好切块的第一标准是「切刀不落在语义边界上」,第二标准是「每一块都记得自己在原文的第几个字符」。

舞台元素与比喻:文本是一条传送带,归一化工位是一组刷子(刷掉脏字符和混杂空白),切分工位是一把带传感器的剪刀(传感器就是实体/关系坐标),溯源工位是一个盖章机(给每块盖「原文第 X 到第 Y 字符」的章)。

分步动画:

  1. 倒入一段带制表符、弯引号、混合换行的文本,刷子工位亮起。
  2. 字幕:「先做 Unicode、空白、特殊字符、大小写四步,字符偏移才稳定。」
  3. 传送带前进到剪刀工位,先演示 recursive:剪刀按分隔符优先级找落点,落在句号后。
  4. 字幕:「找不到更细的分隔符,就退到空格,再退到硬切。」
  5. 切换 entity_aware:剪刀前方的传感器亮起,检测到当前句子里有实体边界,剪刀抬起、不切。
  6. 字幕:「句子含实体边界时,宁可让块超长也不切。」
  7. 每个输出块经过盖章机,start_indexend_index 印上去,登记进溯源表。
  8. 字幕:「块丢了坐标,下游事实就丢了来源。」

读者可操作项:往文本里插入一个自定义实体名(比如公司全称),切换 recursiveentity_aware 两种方法,观察切点位置变化;把 chunk_size 从 1000 调到 400,看实体感知在越界频繁时是否更频繁地「不切」;清空文本,看输出是不是空列表。

逻辑轨迹面板伪代码,右侧标真实行号:

text
split(text, entity_aware)                 # splitter.py:115
  text 为空? -> 返回 []                    # splitter.py:129-130
  get_split_method(entity_aware)           # splitter.py:141
  调 split_entity_aware(text, ...)         # splitter.py:150
    NER 抽实体 -> entity_boundaries        # methods.py:878-887
    逐句累加 char_pos                      # methods.py:895-901
    句子含实体边界? has_entity_boundary     # methods.py:903-907
    越界且含边界 -> pass 不切              # methods.py:910-914
    否则 -> 产 Chunk 带 start/end          # methods.py:915-919
  返回非空 -> split 直接返回               # splitter.py:154-158

可迁移结论

值得抄的是 Chunk 的坐标形状:一个块必须带 start_indexend_index 这两个原文偏移。这条不依赖 Python,任何语言里都是一对整数,但它是后面所有溯源能力的根基。第二值得抄的是「分隔符优先级表 + 反向查找」,一个循环加一张表就能实现不切烂段落的通用切分,是最小成本形态。第三值得抄的是回退链:把「某个切分器可能坏」当常态,方法名列表就是降级顺序。还有一条比算法选择更靠前的结论:把「块坐标」当成数据模型的一等字段。任何存储层,只要每条记录带 source_id 和 offset 两个字段,就能在出问题时一路回追到原文。算法可以换,坐标不能丢。

过度设计的部分在 22 个方法名里。split_ontology_aware 只是调了 split_entity_aware,注释明说「Ontology-aware splitting using entity-aware method as base」;split_topic_based 只是调了 split_semantic_transformer,注释明说「full implementation pending」。这些是占位别名,用户把它们当独立能力用,会得到与名称不符的近似实现。真正有独立逻辑的 KG 切块只有三个:实体感知、关系感知、图结构。抄的时候砍掉别名层,保留三个核心加一个递归兜底,就够覆盖大部分场景。

思考题

  1. 固定窗口切块为什么会拆散三元组?在 split_relation_aware 里,具体是哪一行阻止了「把主语和宾语切进两块」?
  2. 动手验证:把 semantica/split/methods.py 第 183 行 split_recursive 的默认 separators 里的 ". " 删掉,然后运行 python3 -c "from semantica.split.methods import split_recursive; [print(c.start_index, c.end_index, repr(c.text[:30])) for c in split_recursive('Alice met Bob. Carol met Dave. Eve met Frank.', chunk_size=20, chunk_overlap=0)]",对比删除前后切点位置的变化,说明句号分隔符的作用。
  3. 设计题:如果你只能用 100 行代码实现「不切烂语义边界」,你会保留实体感知、关系感知、图结构、层级中的哪几个机制,为什么?GraphRAG 只留了 token 滑窗,它的什么设计补上了切块阶段的缺失?