# # Copyright 2025 The InfiniFlow Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import random import re from copy import deepcopy from common.float_utils import normalize_overlapped_percent from common.token_utils import num_tokens_from_string from rag.flow.base import ProcessBase, ProcessParamBase from rag.flow.chunker.schema import TokenChunkerFromUpstream from rag.flow.parser.pdf_chunk_metadata import ( PDF_POSITIONS_KEY, extract_pdf_positions, finalize_pdf_chunk, restore_pdf_text_previews, ) from rag.nlp import naive_merge # _TAG_RE matches parser-emitted coordinate tags of the form # ``@@\t\t\t\t##``. Mirrors Go's # posTagRemove (internal/ingestion/component/chunker/group.go) so the two # languages strip tags identically. _TAG_RE = re.compile(r"@@[\t0-9.-]+?##") def remove_tag(text): """Strip ``@@...##`` coordinate tags from text. Used both when measuring the overlap prefix (so the cut lands on the tag-free visible text, matching Go's computeOverlapPrefix) and on the final chunk text (so coordinate tags never reach embedding/index). """ return _TAG_RE.sub("", text or "") class TokenChunkerParam(ProcessParamBase): def __init__(self): super().__init__() self.delimiter_mode = "delimiter" self.chunk_token_size = 512 self.delimiters = ["\n"] self.overlapped_percent = 0 self.children_delimiters = [] self.table_context_size = 0 self.image_context_size = 0 def check(self): # Backward-compat: "token_size" was removed but is behaviorally identical # to "delimiter" at runtime (both route through the same code path), so # accept and coerce it instead of rejecting legacy configs / pre-fix # frontends. Only genuinely unknown values are rejected. if self.delimiter_mode == "token_size": self.delimiter_mode = "delimiter" self.check_valid_value(self.delimiter_mode, "Delimiter mode abnormal.", ["delimiter", "one"]) if self.delimiters is None: self.delimiters = [] elif isinstance(self.delimiters, str): self.delimiters = [self.delimiters] else: self.delimiters = [d for d in self.delimiters if isinstance(d, str)] self.delimiters = [d for d in self.delimiters if d] if self.children_delimiters is None: self.children_delimiters = [] elif isinstance(self.children_delimiters, str): self.children_delimiters = [self.children_delimiters] else: self.children_delimiters = [d for d in self.children_delimiters if isinstance(d, str)] self.children_delimiters = [d for d in self.children_delimiters if d] self.check_positive_integer(self.chunk_token_size, "Chunk token size.") self.check_decimal_float(self.overlapped_percent, "Overlapped percentage: [0, 1)") self.check_nonnegative_number(self.table_context_size, "Table context size.") self.check_nonnegative_number(self.image_context_size, "Image context size.") def get_input_form(self) -> dict[str, dict]: return {} def _compile_delimiter_pattern(delimiters): # Build the primary delimiter regex from active delimiters wrapped by backticks. raw_delimiters = "".join(delimiter for delimiter in (delimiters or []) if delimiter) custom_delimiters = [m.group(1) for m in re.finditer(r"`([^`]+)`", raw_delimiters)] if not custom_delimiters: return "" return "|".join(re.escape(text) for text in sorted(set(custom_delimiters), key=len, reverse=True)) def _split_text_by_pattern(text, pattern): # Split text by the compiled delimiter pattern and discard delimiters. # No atom-split is performed; empty segments between consecutive delimiters # are dropped but whitespace-only segments are preserved (the delimiter is # the boundary, not stripped away). if not pattern: return [text or ""] split_texts = re.split(r"(%s)" % pattern, text or "", flags=re.DOTALL) chunks = [] for i in range(0, len(split_texts), 2): chunk = split_texts[i] if chunk: chunks.append(chunk) return chunks def _build_json_chunks(json_result, delimiter_pattern): # Convert upstream JSON items into internal working chunks. chunks = [] for item in json_result: doc_type = str(item.get("doc_type_kwd") or "").strip().lower() if doc_type == "table": ck_type = "table" elif doc_type == "image": ck_type = "image" else: ck_type = "text" text = item.get("text") if not isinstance(text, str): text = item.get("content_with_weight") if not isinstance(text, str): text = "" # Keep PDF coordinates as an internal preview field until the final # output is assembled. This avoids leaking two public coordinate # formats downstream. preview_positions = extract_pdf_positions(item) img_id = item.get("img_id") if ck_type == "text": text_segments = _split_text_by_pattern(text, delimiter_pattern) if delimiter_pattern else [text] for segment in text_segments: if not segment or not segment.strip(): continue chunks.append( { "text": segment, "doc_type_kwd": "text", "ck_type": "text", PDF_POSITIONS_KEY: deepcopy(preview_positions), "tk_nums": num_tokens_from_string(segment), } ) continue chunks.append( { "text": text or "", "doc_type_kwd": ck_type, "ck_type": ck_type, "img_id": img_id, PDF_POSITIONS_KEY: deepcopy(preview_positions), "tk_nums": num_tokens_from_string(text or ""), "context_above": "", "context_below": "", } ) return chunks def _take_sentences(text, need_tokens, from_end=False): # Take text from one side until the target token budget is reached. split_pat = r"([。!??;!\n]|\. )" texts = re.split(split_pat, text or "", flags=re.DOTALL) sentences = [] for i in range(0, len(texts), 2): sentences.append(texts[i] + (texts[i + 1] if i + 1 < len(texts) else "")) iterator = reversed(sentences) if from_end else sentences collected = "" for sentence in iterator: collected = sentence + collected if from_end else collected + sentence if num_tokens_from_string(collected) >= need_tokens: break return collected def _attach_context_to_media_chunks(chunks, table_context_size, image_context_size): # Add surrounding text to table/image chunks when context windows are enabled. for i, chunk in enumerate(chunks): if chunk["ck_type"] not in {"table", "image"}: continue context_size = image_context_size if chunk["ck_type"] == "image" else table_context_size if context_size <= 0: continue remain_above = context_size remain_below = context_size parts_above = [] parts_below = [] prev = i - 1 while prev >= 0 and remain_above > 0: prev_chunk = chunks[prev] if prev_chunk["ck_type"] == "text": if prev_chunk["tk_nums"] >= remain_above: parts_above.insert(0, _take_sentences(prev_chunk["text"], remain_above, from_end=True)) remain_above = 0 break parts_above.insert(0, prev_chunk["text"]) remain_above -= prev_chunk["tk_nums"] prev -= 1 after = i + 1 while after < len(chunks) and remain_below > 0: after_chunk = chunks[after] if after_chunk["ck_type"] == "text": if after_chunk["tk_nums"] >= remain_below: parts_below.append(_take_sentences(after_chunk["text"], remain_below)) remain_below = 0 break parts_below.append(after_chunk["text"]) remain_below -= after_chunk["tk_nums"] after += 1 chunk["context_above"] = "".join(parts_above) chunk["context_below"] = "".join(parts_below) def _merge_text_chunks_by_token_size(chunks, chunk_token_size, overlapped_percent): # Merge adjacent text chunks when delimiter-based splitting is not active. merged = [] prev_text_idx = -1 threshold = chunk_token_size * (100 - overlapped_percent) / 100.0 for chunk in chunks: if chunk["ck_type"] != "text": merged.append(deepcopy(chunk)) prev_text_idx = -1 continue current = deepcopy(chunk) should_start_new = prev_text_idx < 0 or merged[prev_text_idx]["tk_nums"] > threshold # #17799: an over-budget unit stands alone — never merged into the # previous chunk. This matches Python naive_merge and the Go # TokenChunker (all three paths stand the over-budget unit alone), so # the Python JSON path, Python text path, and Go TokenChunker share one # contract. if current["tk_nums"] > chunk_token_size: should_start_new = True if should_start_new: if prev_text_idx >= 0 and overlapped_percent > 0 and merged[prev_text_idx]["text"]: # Mirror Go computeOverlapPrefix: measure the overlap cut on the # tag-free *visible* text, never on the raw text that still # carries @@...## coordinate tags. This keeps the overlap prefix # aligned with Go and prevents a partial tag from leaking into # the next chunk when the cut would land inside a tag. visible = remove_tag(merged[prev_text_idx]["text"]) overlap_start = int(len(visible) * (100 - overlapped_percent) / 100.0) if 0 <= overlap_start < len(visible): overlap_text = visible[overlap_start:] else: overlap_text = "" current["text"] = overlap_text + current["text"] current["tk_nums"] = num_tokens_from_string(current["text"]) merged.append(current) prev_text_idx = len(merged) - 1 continue if merged[prev_text_idx]["text"] and current["text"]: merged[prev_text_idx]["text"] += "\n" + current["text"] else: merged[prev_text_idx]["text"] += current["text"] merged[prev_text_idx][PDF_POSITIONS_KEY].extend(current.get(PDF_POSITIONS_KEY) or []) merged[prev_text_idx]["tk_nums"] += current["tk_nums"] return merged def _finalize_json_chunks(chunks): # Convert internal chunks into the final token chunker output format. docs = [] for chunk in chunks: # Strip parser coordinate tags from the final text so they never # reach embedding/index (coordinates already live in the structured # PDF_POSITIONS_KEY field). Mirrors Go's removeTag at the chunker # output boundary (token.go:544). text = remove_tag((chunk.get("context_above") or "") + (chunk.get("text") or "") + (chunk.get("context_below") or "")) if not text.strip(): continue # The internal preview coordinates are converted exactly once into the # indexed fields consumed downstream. doc = { "text": text, "doc_type_kwd": chunk.get("doc_type_kwd", "text"), } if chunk.get(PDF_POSITIONS_KEY): doc[PDF_POSITIONS_KEY] = deepcopy(chunk[PDF_POSITIONS_KEY]) if chunk.get("mom"): doc["mom"] = chunk["mom"] if chunk.get("img_id"): doc["img_id"] = chunk["img_id"] docs.append(finalize_pdf_chunk(doc)) return docs def _split_chunk_docs_by_children(chunks, pattern): # Apply the secondary children_delimiters split to text chunks only. if not pattern: return chunks docs = [] for chunk in chunks: if chunk.get("doc_type_kwd", "text") != "text": docs.append(chunk) continue split_texts = _split_text_by_pattern(chunk.get("text", ""), pattern) mom = chunk.get("text", "") for text in split_texts: if not text.strip(): continue child = deepcopy(chunk) child["mom"] = mom child["text"] = text docs.append(child) return docs class TokenChunker(ProcessBase): component_name = "TokenChunker" async def _invoke(self, **kwargs): try: from_upstream = TokenChunkerFromUpstream.model_validate(kwargs) except Exception as e: self.set_output("_ERROR", f"Input error: {str(e)}") return # Build the primary delimiter regex. If no active custom delimiter exists, # the token chunker falls back to token-size based merging. delimiter_pattern = _compile_delimiter_pattern(self._param.delimiters) custom_pattern = "|".join(re.escape(t) for t in sorted(set(self._param.children_delimiters), key=len, reverse=True)) self.set_output("output_format", "chunks") self.callback(random.randint(1, 5) / 100.0, "Start to split into chunks.") overlapped_percent = normalize_overlapped_percent(self._param.overlapped_percent) if from_upstream.output_format in ["markdown", "text", "html"]: payload = getattr(from_upstream, f"{from_upstream.output_format}_result") or "" if self._param.delimiter_mode == "one": # Strip parser coordinate tags so they never reach # embedding/index (consistent with the JSON merge path). self.set_output("chunks", [{"text": remove_tag(payload)}] if payload.strip() else []) self.callback(1, "Done.") return if delimiter_pattern: cks = _split_text_by_pattern(payload, delimiter_pattern) else: cks = naive_merge( payload, self._param.chunk_token_size, "".join(self._param.delimiters), overlapped_percent, ) if custom_pattern: docs = [] for c in cks: if not c.strip(): continue for text in _split_text_by_pattern(c, custom_pattern): if not text.strip(): continue docs.append({"text": text, "mom": c}) self.set_output("chunks", docs) else: self.set_output("chunks", [{"text": c.strip()} for c in cks if c.strip()]) self.callback(1, "Done.") return # json json_result = (from_upstream.chunks if from_upstream.output_format == "chunks" else from_upstream.json_result) or [] if self._param.delimiter_mode == "one": sections = [] for item in json_result: text = item.get("text") if not isinstance(text, str): text = item.get("content_with_weight") if isinstance(text, str) and text.strip(): # Strip parser coordinate tags so they never reach # embedding/index (consistent with the JSON merge path). sections.append(remove_tag(text)) merged_text = "\n".join(sections) self.set_output("chunks", [{"text": merged_text}] if merged_text.strip() else []) self.callback(1, "Done.") return # Both branches start from per-item chunks (no pre-split by the # delimiter pattern). The delimiter branch splits the buffered text # stream while preserving per-segment PDF positions; the no-delimiter # branch merges adjacent text items to chunk_token_size (the removed # "token_size" behaviour, and a parity match with the Go JSON path). text_chunks = _build_json_chunks(json_result, "") if delimiter_pattern: chunks = [] text_buffer = [] text_buffer_pos = [] def flush_text_buffer(): if not text_buffer: return # Join buffered text items with "\n" so adjacent item text is not # glued together (e.g. "hello" + "world" must not become "helloworld"). # The delimiter is then applied to the combined text; a segment may # span across item boundaries (the "\n" glue is not itself a # delimiter), so each segment carries only the PDF positions of the # item(s) that contributed to it -- never the union of every item # (which previously leaked page-N coordinates into page-M chunks and # made all segments share one preview image). parts = [] item_ranges = [] # (start, end) of each buffered item in combined_text offset = 0 for text in text_buffer: start = offset parts.append(text) offset += len(text) item_ranges.append((start, offset)) parts.append("\n") offset += 1 combined_text = "".join(parts[:-1]) # drop the trailing glue raw = re.split(r"(%s)" % delimiter_pattern, combined_text, flags=re.DOTALL) segments = [] # (text, start, end) within combined_text pos = 0 for i in range(0, len(raw), 2): seg = raw[i] seg_start = pos seg_end = pos + len(seg) if seg: segments.append((seg, seg_start, seg_end)) pos = seg_end if i + 1 < len(raw): pos += len(raw[i + 1]) for text, seg_start, seg_end in segments: if not text.strip(): continue seg_pos = [] for (istart, iend), item_pos in zip(item_ranges, text_buffer_pos, strict=True): # A segment overlaps an item when their character ranges # intersect; collect that item's coordinates. if seg_start < iend and istart < seg_end: seg_pos.extend(item_pos or []) chunks.append( { "text": text, "doc_type_kwd": "text", "ck_type": "text", PDF_POSITIONS_KEY: deepcopy(seg_pos), "tk_nums": num_tokens_from_string(text), } ) text_buffer.clear() text_buffer_pos.clear() for chunk in text_chunks: if chunk["ck_type"] == "text": text_buffer.append(chunk["text"]) text_buffer_pos.append(chunk.get(PDF_POSITIONS_KEY)) else: flush_text_buffer() chunks.append(chunk) flush_text_buffer() # Apply children_delimiters (secondary split) before finalizing. if custom_pattern: chunks = _split_chunk_docs_by_children(chunks, custom_pattern) _attach_context_to_media_chunks(chunks, self._param.table_context_size, self._param.image_context_size) else: # No active delimiter: merge adjacent text items to chunk_token_size. # This runs on the per-item chunks (NOT a single concatenated chunk), # so the token cap is actually enforced -- matching the previous # "token_size" mode and the Go JSON path. Media chunks break the merge. # Media context is attached on the per-item chunks before merging, as # the removed "token_size" branch did, to preserve context windows. _attach_context_to_media_chunks(text_chunks, self._param.table_context_size, self._param.image_context_size) chunks = _merge_text_chunks_by_token_size(text_chunks, self._param.chunk_token_size, overlapped_percent) if custom_pattern: chunks = _split_chunk_docs_by_children(chunks, custom_pattern) await restore_pdf_text_previews(chunks, from_upstream, self._canvas) self.set_output("chunks", _finalize_json_chunks(chunks)) self.callback(1, "Done.") return