From bd93ccb8905bb3168966399c384c74780cce853c Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:08:25 -0700 Subject: [PATCH] refactor(gateway): shared fence-aware markdown chunker core (yuanbao-derived) + canonical table-row splitter --- gateway/platforms/helpers.py | 435 +++++++++++++++++++++++++++- gateway/platforms/weixin.py | 38 +-- gateway/platforms/yuanbao.py | 330 ++------------------- gateway/stream_consumer.py | 71 ++--- hermes_cli/web_server.py | 8 +- tests/gateway/test_fence_chunker.py | 248 ++++++++++++++++ 6 files changed, 743 insertions(+), 387 deletions(-) create mode 100644 tests/gateway/test_fence_chunker.py diff --git a/gateway/platforms/helpers.py b/gateway/platforms/helpers.py index c99020abc8948..9df6a8e0525f0 100644 --- a/gateway/platforms/helpers.py +++ b/gateway/platforms/helpers.py @@ -315,13 +315,16 @@ def is_table_row(line: str) -> bool: def split_markdown_table_row(line: str) -> list[str]: - """Split a GFM table row into stripped cell values.""" - stripped = line.strip() - if stripped.startswith("|"): - stripped = stripped[1:] - if stripped.endswith("|"): - stripped = stripped[:-1] - return [cell.strip() for cell in stripped.split("|")] + """Split a GFM table row into stripped cell values. + + Thin delegate to the canonical implementation in + :mod:`agent.markdown_tables` (``split_table_row``) so the three + formerly byte-identical copies (here, ``agent/markdown_tables.py``, + ``weixin._split_table_row``) share one body. + """ + from agent.markdown_tables import split_table_row + + return split_table_row(line) def _render_table_block(table_block: list[str]) -> str: @@ -519,3 +522,421 @@ def compile_mention_patterns( except re.error as exc: log.warning("[%s] Invalid mention pattern %r: %s", log_prefix, text, exc) return compiled + + +# ─── Fence-Aware Markdown Chunking ─────────────────────────────────────────── +# Shared core for the fence-aware markdown chunkers that previously lived as +# near-duplicates in gateway/stream_consumer.py, gateway/platforms/yuanbao.py +# (MarkdownProcessor — the richest version, which this core is derived from), +# and gateway/platforms/weixin.py. Each caller keeps its own knobs: +# +# * stream_consumer: newline-preferred splitting + close/reopen fence +# balancing (``prefer_paragraphs=False, balance_fences=True``) +# * yuanbao: atomic-block extraction + paragraph-boundary splitting, fences +# kept intact as atoms (``prefer_paragraphs=True, balance_fences=False``) +# * weixin: keeps its own block splitter (anchored ``_FENCE_RE``, per-line +# rstrip semantics) but reuses ``greedy_pack_blocks`` for packing. +# +# The typing helpers below use ``Optional``/``Callable`` from ``typing`` to +# match the module's existing import style. + + +def text_has_unclosed_fence(text: str) -> bool: + """Return True when *text* ends inside an unclosed ``` code fence. + + Scans line by line, toggling in/out state on lines starting with ```. + An odd number of toggles means the trailing fence is unclosed. + """ + in_fence = False + for line in text.split('\n'): + if line.startswith('```'): + in_fence = not in_fence + return in_fence + + +def text_ends_with_table_row(text: str) -> bool: + """True when the last non-empty line starts and ends with ``|``.""" + trimmed = text.rstrip() + if not trimmed: + return False + last_line = trimmed.split('\n')[-1].strip() + return last_line.startswith('|') and last_line.endswith('|') + + +def is_fence_atom(text: str) -> bool: + """True when an atomic block is a code block (starts with ```).""" + return text.lstrip().startswith('```') + + +def is_table_atom(text: str) -> bool: + """True when an atomic block is a table (first line is ``|...|``).""" + first_line = text.split('\n')[0].strip() + return first_line.startswith('|') and first_line.endswith('|') + + +_SENTENCE_END_NEWLINE_RE = re.compile(r'[。！？.!?]\n') + + +def split_at_paragraph_boundary(text, max_chars, len_fn=None): + """Find the nearest paragraph boundary within *max_chars*; return (head, tail). + + Split priority: + 1. Blank line (paragraph boundary) + 2. Newline after sentence-ending punctuation (CJK and ASCII) + 3. Last newline + 4. Force split at the *max_chars* window boundary + + ``head + tail == text`` always holds. *len_fn* allows measuring in + custom units (e.g. UTF-16 code units); a binary search finds the largest + prefix that fits when it is provided. + """ + _len = len_fn or len + if _len(text) <= max_chars: + return text, '' + + if _len is len: + window = text[:max_chars] + else: + lo, hi = 0, len(text) + while lo < hi: + mid = (lo + hi + 1) // 2 + if _len(text[:mid]) <= max_chars: + lo = mid + else: + hi = mid - 1 + window = text[:lo] + + # 1. Prefer the last blank line (\n\n) as paragraph boundary + pos = window.rfind('\n\n') + if pos > 0: + return text[:pos + 2], text[pos + 2:] + + # 2. Then the last newline following sentence-ending punctuation + best_pos = -1 + for m in _SENTENCE_END_NEWLINE_RE.finditer(window): + best_pos = m.end() + if best_pos > 0: + return text[:best_pos], text[best_pos:] + + # 3. Fallback: last newline + pos = window.rfind('\n') + if pos > 0: + return text[:pos + 1], text[pos + 1:] + + # 4. No valid split point: force split at the window boundary + cut = len(window) + return text[:cut], text[cut:] + + +def split_markdown_atoms(text: str) -> "list[str]": + """Split markdown into indivisible "atomic blocks". + + Atoms are: fenced code blocks (``` ... ``` inclusive), tables + (consecutive ``|...|`` lines), and plain paragraphs separated by blank + lines. Blank lines are separators and belong to no atom. + """ + lines = text.split('\n') + atoms: "list[str]" = [] + + current_lines: "list[str]" = [] + in_fence = False + + def _is_table_line(line: str) -> bool: + stripped = line.strip() + return stripped.startswith('|') and stripped.endswith('|') + + def _flush_current() -> None: + if current_lines: + atom = '\n'.join(current_lines) + if atom.strip(): + atoms.append(atom) + current_lines.clear() + + for line in lines: + if in_fence: + current_lines.append(line) + if line.startswith('```') and len(current_lines) > 1: + in_fence = False + _flush_current() + elif line.startswith('```'): + _flush_current() + in_fence = True + current_lines.append(line) + elif _is_table_line(line): + if current_lines and not _is_table_line(current_lines[-1]): + _flush_current() + current_lines.append(line) + elif line.strip() == '': + _flush_current() + else: + if current_lines and _is_table_line(current_lines[-1]): + _flush_current() + current_lines.append(line) + + _flush_current() + + return atoms + + +def infer_block_separator(prev_chunk: str, next_chunk: str) -> str: + """Infer the separator (``'\\n'`` or ``'\\n\\n'``) between two chunks. + + Single newline when the boundary sits at a code fence or a continued + table; paragraph separator otherwise. + """ + prev_trimmed = prev_chunk.rstrip() + next_trimmed = next_chunk.lstrip() + + if prev_trimmed.endswith('```') or next_trimmed.startswith('```'): + return '\n' + + if text_ends_with_table_row(prev_chunk): + first_line = next_trimmed.split('\n')[0].strip() if next_trimmed else '' + if first_line.startswith('|') and first_line.endswith('|'): + return '\n' + + return '\n\n' + + +def merge_streaming_fences(chunks: "list[str]") -> "list[str]": + """Stream-aware fence merge: rejoin chunks truncated mid-fence. + + While chunk *i* has an unclosed fence and a successor exists, merge the + successor into it using :func:`infer_block_separator`. + """ + if not chunks: + return [] + + result: "list[str]" = [] + i = 0 + while i < len(chunks): + current = chunks[i] + while text_has_unclosed_fence(current) and i + 1 < len(chunks): + sep = infer_block_separator(current, chunks[i + 1]) + current = current + sep + chunks[i + 1] + i += 1 + result.append(current) + i += 1 + + return result + + +def balance_fences_across_chunks(chunks: "list[str]") -> "list[str]": + """Close orphaned ``` fences at each chunk boundary and reopen on the next. + + When a split lands inside a triple-backtick code block, close the fence + at the end of the head chunk and reopen it (with the original language + tag) at the start of the next, so every delivered chunk is + fence-balanced on its own. + """ + if len(chunks) <= 1: + return chunks + out: "list[str]" = [] + carry_lang = None + for chunk in chunks: + prefix = f"```{carry_lang}\n" if carry_lang is not None else "" + in_code = carry_lang is not None + lang = carry_lang or "" + for line in chunk.split("\n"): + stripped = line.strip() + if stripped.startswith("```"): + if in_code: + in_code = False + lang = "" + else: + in_code = True + tag = stripped[3:].strip() + lang = tag.split()[0] if tag else "" + body = prefix + chunk + if in_code: + body += "\n```" + carry_lang = lang + else: + carry_lang = None + out.append(body) + return out + + +def greedy_pack_blocks(blocks, max_length, len_fn=None, sep="\n\n", overflow=None): + """Greedily pack pre-split *blocks* into chunks of at most *max_length*. + + Blocks are joined with *sep* while they fit. A block that alone exceeds + the limit is passed to *overflow(block)* (which must return a list of + chunks) when provided, else emitted as-is. + """ + _len = len_fn or len + packed: "list[str]" = [] + current = "" + for block in blocks: + candidate = block if not current else f"{current}{sep}{block}" + if _len(candidate) <= max_length: + current = candidate + continue + if current: + packed.append(current) + current = "" + if _len(block) <= max_length: + current = block + continue + if overflow is not None: + packed.extend(overflow(block)) + else: + packed.append(block) + if current: + packed.append(current) + return packed + + +def split_text_fence_aware( + text, + limit, + len_fn=None, + *, + prefer_paragraphs=True, + balance_fences=False, +): + """Split markdown text into chunks of at most *limit*, respecting fences. + + Two strategies, selected by ``prefer_paragraphs``: + + ``prefer_paragraphs=True`` (yuanbao-derived, the richest): + Extract atomic blocks (code fences, tables, paragraphs), greedily merge + them up to *limit*, split still-oversized non-atomic chunks at + paragraph boundaries, then re-merge small neighbours. Code blocks and + tables are never split in the middle; a single atom larger than + *limit* is emitted oversize rather than broken. + + ``prefer_paragraphs=False`` (stream_consumer-derived): + Newline-preferred hard splitting with headroom reserved for fence + markers when the text contains ```. + + ``balance_fences=True`` post-processes the chunks so a split inside a + code block closes the fence on the head chunk and reopens it (with the + language tag) on the tail — required by callers whose chunks are + delivered as independent messages that each must render standalone. + """ + _len = len_fn or len + + if not text: + return [] + + if prefer_paragraphs: + chunks = _chunk_markdown_paragraphs(text, limit, len_fn) + else: + chunks = _chunk_newline_preferred(text, limit, _len) + + if balance_fences: + chunks = balance_fences_across_chunks(chunks) + return chunks + + +def _chunk_markdown_paragraphs(text, max_chars, len_fn=None): + """Yuanbao-derived paragraph/atom chunking pipeline (see module docs).""" + _len = len_fn or len + + if _len(text) <= max_chars: + return [text] + + # Phase 1: Extract atomic blocks + atoms = split_markdown_atoms(text) + + # Phase 2: Greedy merge + chunks: "list[str]" = [] + indivisible_set: "set[int]" = set() + current_parts: "list[str]" = [] + current_len = 0 + + def _flush_parts() -> None: + if current_parts: + chunks.append('\n\n'.join(current_parts)) + + for atom in atoms: + atom_len = _len(atom) + sep_len = 2 if current_parts else 0 + projected_len = current_len + sep_len + atom_len + + if projected_len > max_chars and current_parts: + _flush_parts() + current_parts = [] + current_len = 0 + sep_len = 0 + + if (not current_parts + and atom_len > max_chars + and (is_fence_atom(atom) or is_table_atom(atom))): + indivisible_set.add(len(chunks)) + chunks.append(atom) + continue + + current_parts.append(atom) + current_len += sep_len + atom_len + + _flush_parts() + + # Phase 3: Split still-oversized chunks at paragraph boundaries + result: "list[str]" = [] + for idx, chunk in enumerate(chunks): + if _len(chunk) <= max_chars: + result.append(chunk) + continue + + if idx in indivisible_set: + result.append(chunk) + continue + + if text_has_unclosed_fence(chunk): + result.append(chunk) + continue + + remaining = chunk + while _len(remaining) > max_chars: + head, remaining = split_at_paragraph_boundary( + remaining, max_chars, len_fn=len_fn, + ) + if not head: + head, remaining = remaining[:max_chars], remaining[max_chars:] + if head: + result.append(head) + if remaining: + result.append(remaining) + + # Phase 4: Merge small trailing/leading chunks with neighbours + if len(result) > 1: + merged: "list[str]" = [result[0]] + for chunk in result[1:]: + prev = merged[-1] + combined = prev + '\n\n' + chunk + if _len(combined) <= max_chars: + merged[-1] = combined + else: + merged.append(chunk) + result = merged + + return [c for c in result if c] + + +def _chunk_newline_preferred(text, limit, len_fn): + """Stream-consumer-derived newline-preferred splitting (no balancing).""" + if len_fn(text) <= limit: + return [text] + # Reserve headroom for the close/reopen fence markers a balancing pass + # may add, so balanced chunks stay within the platform limit. + split_limit = limit + if "```" in text: + split_limit = max(limit - 16, limit // 2, 1) + # Local import: gateway.platforms.base is heavyweight and pulls config; + # helpers must stay import-light for adapters that import it first. + from gateway.platforms.base import _custom_unit_to_cp + + chunks: "list[str]" = [] + remaining = text + while len_fn(remaining) > split_limit: + _cp_budget = _custom_unit_to_cp(remaining, split_limit, len_fn) + split_at = remaining.rfind("\n", 0, _cp_budget) + if split_at < _cp_budget // 2: + split_at = _cp_budget + chunks.append(remaining[:split_at]) + remaining = remaining[split_at:].lstrip("\n") + if remaining: + chunks.append(remaining) + return chunks diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index bd25319fc8b4a..b44ce1ee698fb 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -56,7 +56,7 @@ except ImportError: # pragma: no cover - dependency gate CRYPTO_AVAILABLE = False from gateway.config import Platform, PlatformConfig -from gateway.platforms.helpers import MessageDeduplicator +from gateway.platforms.helpers import MessageDeduplicator, greedy_pack_blocks from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, @@ -670,12 +670,10 @@ def _mime_from_filename(filename: str) -> str: def _split_table_row(line: str) -> List[str]: - row = line.strip() - if row.startswith("|"): - row = row[1:] - if row.endswith("|"): - row = row[:-1] - return [cell.strip() for cell in row.split("|")] + """Delegate to the canonical table-row splitter in agent.markdown_tables.""" + from agent.markdown_tables import split_table_row + + return split_table_row(line) def _normalize_markdown_blocks(content: str) -> str: @@ -868,24 +866,14 @@ def _should_split_short_chat_block_for_weixin(block: str) -> bool: def _pack_markdown_blocks_for_weixin(content: str, max_length: int) -> List[str]: if len(content) <= max_length: return [content] - - packed: List[str] = [] - current = "" - for block in _split_markdown_blocks(content): - candidate = block if not current else f"{current}\n\n{block}" - if len(candidate) <= max_length: - current = candidate - continue - if current: - packed.append(current) - current = "" - if len(block) <= max_length: - current = block - continue - packed.extend(BasePlatformAdapter.truncate_message(block, max_length)) - if current: - packed.append(current) - return packed + # Block extraction stays weixin-local (_split_markdown_blocks uses the + # anchored _FENCE_RE + per-line rstrip semantics); the greedy packing + # loop is the shared core's. + return greedy_pack_blocks( + _split_markdown_blocks(content), + max_length, + overflow=lambda block: BasePlatformAdapter.truncate_message(block, max_length), + ) def _split_text_for_weixin_delivery( diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py index 9b3d34fe06ad2..96f8580179df3 100644 --- a/gateway/platforms/yuanbao.py +++ b/gateway/platforms/yuanbao.py @@ -60,6 +60,7 @@ from gateway.platforms.base import ( cache_image_from_bytes, cache_video_from_bytes, ) +from gateway.platforms import helpers as _mdchunk from gateway.platforms.helpers import MessageDeduplicator from gateway.platforms.yuanbao_media import ( download_url as media_download_url, @@ -202,45 +203,23 @@ class MarkdownProcessor: """ # -- Fence detection --------------------------------------------------- + # All chunking primitives below are thin delegates to the shared + # fence-aware chunker core in gateway.platforms.helpers, which was + # extracted from this class (the richest of the four duplicate + # implementations). The MarkdownProcessor method names are kept for + # the existing call sites and tests. @staticmethod def has_unclosed_fence(text: str) -> bool: - """ - Detect whether the text has unclosed code block fences. - - Scan line by line, toggling in/out state when encountering a line starting with ```. - An odd number of toggles indicates an unclosed fence. - - Args: - text: Markdown text to check - - Returns: - Returns True if the text ends with an unclosed fence, otherwise False - """ - in_fence = False - for line in text.split('\n'): - if line.startswith('```'): - in_fence = not in_fence - return in_fence + """Detect whether the text has unclosed code block fences.""" + return _mdchunk.text_has_unclosed_fence(text) # -- Table detection --------------------------------------------------- @staticmethod def ends_with_table_row(text: str) -> bool: - """ - Detect whether the text ends with a table row (last non-empty line starts and ends with |). - - Args: - text: Text to check - - Returns: - Returns True if the last non-empty line is a table row - """ - trimmed = text.rstrip() - if not trimmed: - return False - last_line = trimmed.split('\n')[-1].strip() - return last_line.startswith('|') and last_line.endswith('|') + """Detect whether the text ends with a table row.""" + return _mdchunk.text_ends_with_table_row(text) # -- Paragraph boundary splitting -------------------------------------- @@ -250,135 +229,25 @@ class MarkdownProcessor: max_chars: int, len_fn: Optional[Callable[[str], int]] = None, ) -> tuple[str, str]: - """ - Find the nearest paragraph boundary split point within max_chars, return (head, tail). - - Split priority: - 1. Blank line (paragraph boundary) - 2. Newline after period/question mark/exclamation mark (Chinese and English) - 3. Last newline - 4. Force split at max_chars - - Args: - text: Text to split - max_chars: Maximum character count limit - len_fn: Optional custom length function (e.g. UTF-16 length); defaults to built-in len - - Returns: - (head, tail) tuple, head is the front part, tail is the back part, satisfying head + tail == text - """ - _len = len_fn or len - if _len(text) <= max_chars: - return text, '' - - # Build a character-index window that fits within max_chars. - # When len_fn != len we cannot simply slice [:max_chars], so we - # binary-search for the largest prefix that fits. - if _len is len: - window = text[:max_chars] - else: - lo, hi = 0, len(text) - while lo < hi: - mid = (lo + hi + 1) // 2 - if _len(text[:mid]) <= max_chars: - lo = mid - else: - hi = mid - 1 - window = text[:lo] - - # 1. Prefer the last blank line (\n\n) as paragraph boundary - pos = window.rfind('\n\n') - if pos > 0: - return text[:pos + 2], text[pos + 2:] - - # 2. Then find the last newline after a sentence-ending punctuation - sentence_end_re = re.compile(r'[。！？.!?]\n') - best_pos = -1 - for m in sentence_end_re.finditer(window): - best_pos = m.end() - if best_pos > 0: - return text[:best_pos], text[best_pos:] - - # 3. Fallback: find the last newline - pos = window.rfind('\n') - if pos > 0: - return text[:pos + 1], text[pos + 1:] - - # 4. No valid split point found, force split at window boundary - cut = len(window) - return text[:cut], text[cut:] + """Find the nearest paragraph boundary within max_chars; return (head, tail).""" + return _mdchunk.split_at_paragraph_boundary(text, max_chars, len_fn=len_fn) # -- Atomic block helpers (private) ------------------------------------ @staticmethod def is_fence_atom(text: str) -> bool: """Determine whether an atomic block is a code block (starts with ```).""" - return text.lstrip().startswith('```') + return _mdchunk.is_fence_atom(text) @staticmethod def is_table_atom(text: str) -> bool: """Determine whether an atomic block is a table (first line starts with |).""" - first_line = text.split('\n')[0].strip() - return first_line.startswith('|') and first_line.endswith('|') + return _mdchunk.is_table_atom(text) @staticmethod def split_into_atoms(text: str) -> list[str]: - """ - Split text into a list of "atomic blocks", each being an indivisible logical unit: - - - Code block (fence): from opening ``` to closing ``` (including fence lines) - - Table: consecutive |...| lines forming a whole segment - - Normal paragraph: plain text segments separated by blank lines - - Blank lines serve as separators and are not included in any atomic block. - - Args: - text: Markdown text to split - - Returns: - List of atomic block strings (all non-empty) - """ - lines = text.split('\n') - atoms: list[str] = [] - - current_lines: list[str] = [] - in_fence = False - - def _is_table_line(line: str) -> bool: - stripped = line.strip() - return stripped.startswith('|') and stripped.endswith('|') - - def _flush_current() -> None: - if current_lines: - atom = '\n'.join(current_lines) - if atom.strip(): - atoms.append(atom) - current_lines.clear() - - for line in lines: - if in_fence: - current_lines.append(line) - if line.startswith('```') and len(current_lines) > 1: - in_fence = False - _flush_current() - elif line.startswith('```'): - _flush_current() - in_fence = True - current_lines.append(line) - elif _is_table_line(line): - if current_lines and not _is_table_line(current_lines[-1]): - _flush_current() - current_lines.append(line) - elif line.strip() == '': - _flush_current() - else: - if current_lines and _is_table_line(current_lines[-1]): - _flush_current() - current_lines.append(line) - - _flush_current() - - return atoms + """Split text into a list of indivisible "atomic blocks".""" + return _mdchunk.split_markdown_atoms(text) # -- Core: chunk splitting --------------------------------------------- @@ -392,177 +261,34 @@ class MarkdownProcessor: """ Split Markdown text into multiple chunks by max_chars. - Guarantees: + Guarantees (provided by the shared core, prefer_paragraphs mode): - Each chunk <= max_chars characters (unless a single code block/table itself exceeds the limit) - Code blocks (```...```) are not split in the middle - Table rows are not split in the middle (tables output as atomic blocks) - Split at paragraph boundaries (blank lines, after periods, etc.) - Small trailing/leading chunks are merged with neighbours when possible - - Args: - text: Markdown text to split - max_chars: Max characters per chunk, default 4000 - len_fn: Optional custom length function (e.g. UTF-16 length); defaults to built-in len - - Returns: - List of text chunks after splitting (non-empty) """ - _len = len_fn or len - - if not text: - return [] - - if _len(text) <= max_chars: - return [text] - - # Phase 1: Extract atomic blocks - atoms = cls.split_into_atoms(text) - - # Phase 2: Greedy merge - chunks: list[str] = [] - indivisible_set: set[int] = set() - current_parts: list[str] = [] - current_len = 0 - - def _flush_parts() -> None: - if current_parts: - chunks.append('\n\n'.join(current_parts)) - - for atom in atoms: - atom_len = _len(atom) - sep_len = 2 if current_parts else 0 - projected_len = current_len + sep_len + atom_len - - if projected_len > max_chars and current_parts: - _flush_parts() - current_parts = [] - current_len = 0 - sep_len = 0 - - if (not current_parts - and atom_len > max_chars - and (cls.is_fence_atom(atom) or cls.is_table_atom(atom))): - indivisible_set.add(len(chunks)) - chunks.append(atom) - continue - - current_parts.append(atom) - current_len += sep_len + atom_len - - _flush_parts() - - # Phase 3: Post-processing — split still-oversized chunks at paragraph boundaries - result: list[str] = [] - for idx, chunk in enumerate(chunks): - if _len(chunk) <= max_chars: - result.append(chunk) - continue - - if idx in indivisible_set: - result.append(chunk) - continue - - if cls.has_unclosed_fence(chunk): - result.append(chunk) - continue - - remaining = chunk - while _len(remaining) > max_chars: - head, remaining = cls.split_at_paragraph_boundary( - remaining, max_chars, len_fn=len_fn, - ) - if not head: - head, remaining = remaining[:max_chars], remaining[max_chars:] - if head: - result.append(head) - if remaining: - result.append(remaining) - - # Phase 4: Merge small trailing/leading chunks with neighbours - if len(result) > 1: - merged: list[str] = [result[0]] - for chunk in result[1:]: - prev = merged[-1] - combined = prev + '\n\n' + chunk - if _len(combined) <= max_chars: - merged[-1] = combined - else: - merged.append(chunk) - result = merged - - return [c for c in result if c] + return _mdchunk.split_text_fence_aware( + text, + max_chars, + len_fn, + prefer_paragraphs=True, + balance_fences=False, + ) # -- Block separator inference ----------------------------------------- @classmethod def infer_block_separator(cls, prev_chunk: str, next_chunk: str) -> str: - """ - Infer the separator to use between two split chunks. - - Rules (aligned with TS markdown-stream.ts): - - Previous chunk ends with code fence or next chunk starts with fence → single newline '\\n' - - Previous chunk ends with table row and next chunk starts with table row → single newline '\\n' (continued table) - - Otherwise → double newline '\\n\\n' (paragraph separator) - - Args: - prev_chunk: Previous chunk - next_chunk: Next chunk - - Returns: - '\\n' or '\\n\\n' - """ - prev_trimmed = prev_chunk.rstrip() - next_trimmed = next_chunk.lstrip() - - # Previous chunk ends with fence or next chunk starts with fence - if prev_trimmed.endswith('```') or next_trimmed.startswith('```'): - return '\n' - - # Table continuation - if cls.ends_with_table_row(prev_chunk): - first_line = next_trimmed.split('\n')[0].strip() if next_trimmed else '' - if first_line.startswith('|') and first_line.endswith('|'): - return '\n' - - return '\n\n' + """Infer the separator ('\\n' or '\\n\\n') between two split chunks.""" + return _mdchunk.infer_block_separator(prev_chunk, next_chunk) # -- Streaming fence merge --------------------------------------------- @classmethod def merge_block_streaming_fences(cls, chunks: list[str]) -> list[str]: - """ - Stream-aware fence-conscious chunk merging. - - When streaming output produces multiple chunks truncated in the middle of a fence, - attempt to merge adjacent chunks to complete the fence. - - Rules: - - If chunk i has an unclosed fence and chunk i+1 starts with ```, - merge i+1 into i (until the fence is closed or no more chunks). - - Use infer_block_separator to infer the separator during merging. - - Args: - chunks: Original chunk list - - Returns: - Merged chunk list (length <= original length) - """ - if not chunks: - return [] - - result: list[str] = [] - i = 0 - while i < len(chunks): - current = chunks[i] - # If current chunk has unclosed fence, try merging subsequent chunks - while cls.has_unclosed_fence(current) and i + 1 < len(chunks): - sep = cls.infer_block_separator(current, chunks[i + 1]) - current = current + sep + chunks[i + 1] - i += 1 - result.append(current) - i += 1 - - return result + """Stream-aware fence-conscious chunk merging (see shared core).""" + return _mdchunk.merge_streaming_fences(chunks) # -- Outer fence stripping --------------------------------------------- diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index b140df2b326f4..0e601be801606 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -1181,40 +1181,13 @@ class GatewayStreamConsumer: def _balance_fences_across_chunks(chunks: "list[str]") -> "list[str]": """Close orphaned ``` fences at each chunk boundary and reopen on the next. - When a split lands inside a triple-backtick code block, the head chunk - would render everything after the orphaned fence as code, and the tail - chunk's content would lose its code formatting. Mirror - ``BasePlatformAdapter.truncate_message``'s contract: close the fence at - the end of the chunk and reopen it (with the original language tag) at - the start of the next one, so EVERY delivered chunk is fence-balanced - on its own. + Thin delegate to the shared fence-chunker core in + :mod:`gateway.platforms.helpers` (``balance_fences_across_chunks``); + kept as a method for the existing call sites and tests. """ - if len(chunks) <= 1: - return chunks - out: "list[str]" = [] - carry_lang: "Optional[str]" = None - for chunk in chunks: - prefix = f"```{carry_lang}\n" if carry_lang is not None else "" - in_code = carry_lang is not None - lang = carry_lang or "" - for line in chunk.split("\n"): - stripped = line.strip() - if stripped.startswith("```"): - if in_code: - in_code = False - lang = "" - else: - in_code = True - tag = stripped[3:].strip() - lang = tag.split()[0] if tag else "" - body = prefix + chunk - if in_code: - body += "\n```" - carry_lang = lang - else: - carry_lang = None - out.append(body) - return out + from gateway.platforms.helpers import balance_fences_across_chunks + + return balance_fences_across_chunks(chunks) @staticmethod def _split_text_chunks( @@ -1227,26 +1200,20 @@ class GatewayStreamConsumer: Chunks are fence-balanced: a split inside a ``` code block closes the fence on the head chunk and reopens it on the tail, so no chunk leaves the rest of a message rendering as one giant code block. + + Delegates to the shared fence-chunker core + (:func:`gateway.platforms.helpers.split_text_fence_aware`) with this + consumer's knobs: newline-preferred splitting + fence balancing. """ - if len_fn(text) <= limit: - return [text] - # Reserve headroom for the close/reopen fence markers the balancing - # pass may add, so balanced chunks stay within the platform limit. - split_limit = limit - if "```" in text: - split_limit = max(limit - 16, limit // 2, 1) - chunks: list[str] = [] - remaining = text - while len_fn(remaining) > split_limit: - _cp_budget = _custom_unit_to_cp(remaining, split_limit, len_fn) - split_at = remaining.rfind("\n", 0, _cp_budget) - if split_at < _cp_budget // 2: - split_at = _cp_budget - chunks.append(remaining[:split_at]) - remaining = remaining[split_at:].lstrip("\n") - if remaining: - chunks.append(remaining) - return GatewayStreamConsumer._balance_fences_across_chunks(chunks) + from gateway.platforms.helpers import split_text_fence_aware + + return split_text_fence_aware( + text, + limit, + len_fn, + prefer_paragraphs=False, + balance_fences=True, + ) def _truncate_for_stream( self, diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 1bb2ca961923d..4c2d76a27598f 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -4496,7 +4496,13 @@ async def speak_text(payload: TTSSpeakRequest, profile: Optional[str] = None): def _split_text_for_speak_stream(text: str, cap: int) -> list: - """Split *text* into provider-cap-sized pieces on sentence boundaries.""" + """Split *text* into provider-cap-sized pieces on sentence boundaries. + + Deliberately NOT unified with gateway.platforms.helpers' + split_text_fence_aware: this splitter reflows whitespace (sentences are + re-joined with single spaces) and has no fence/markdown semantics, so + expressing it as knobs on the fence-aware core would change behavior. + """ from tools.tts_streaming import SENTENCE_BOUNDARY_RE as _SENTENCE_BOUNDARY_RE cap = cap if cap and cap > 0 else 4000 diff --git a/tests/gateway/test_fence_chunker.py b/tests/gateway/test_fence_chunker.py new file mode 100644 index 0000000000000..b03d5cca8bbe6 --- /dev/null +++ b/tests/gateway/test_fence_chunker.py @@ -0,0 +1,248 @@ +"""Invariant tests for the shared fence-aware markdown chunker core. + +The core lives in ``gateway.platforms.helpers`` and was extracted from the +Yuanbao ``MarkdownProcessor`` (the richest of the four formerly duplicated +implementations). These tests assert the core's *contract* (invariants), +not exact snapshots, so they survive internal tweaks that preserve behavior. +""" + +import pytest + +from gateway.platforms.helpers import ( + balance_fences_across_chunks, + greedy_pack_blocks, + infer_block_separator, + merge_streaming_fences, + split_at_paragraph_boundary, + split_markdown_atoms, + split_markdown_table_row, + split_text_fence_aware, + text_has_unclosed_fence, +) + + +def utf16_len(s: str) -> int: + return sum(2 if ord(c) > 0xFFFF else 1 for c in s) + + +LONG_PARAS = ("This is a sentence that goes on for a while. " * 8 + "\n\n") * 6 +FENCED = ( + "Header line\n\n```js\n" + + "\n".join(f"console.log({i}); // padding padding" for i in range(30)) + + "\n```\n\nTail paragraph after the code block ends here." +) +UNCLOSED = ( + "Some text before.\n\n```bash\necho one\n" + + "echo more stuff here to pad the line out\n" * 10 +) +TABLE = ( + "Intro paragraph.\n\n| col1 | col2 |\n|------|------|\n" + + "\n".join(f"| value{i} | data{i} |" for i in range(40)) + + "\n\nDone." +) +CJK = "这是一个中文段落，包含了很多字符。用于测试宽字符处理。这句话结束了。\n\n" * 10 +MIXED = ( + "Start.\n\n```sql\nSELECT * FROM t;\n```\n\n| x | y |\n|---|---|\n| 1 | 2 |\n\n" + "End paragraph with some extra words to pad things out." +) + +SAMPLES = [LONG_PARAS, FENCED, UNCLOSED, TABLE, CJK, MIXED, "x" * 500] + + +# ── split_text_fence_aware (paragraph mode: yuanbao-derived) ───────────────── + + +@pytest.mark.parametrize("text", SAMPLES) +@pytest.mark.parametrize("limit", [80, 200, 400]) +def test_paragraph_mode_chunks_within_limit_unless_atomic(text, limit): + chunks = split_text_fence_aware(text, limit, prefer_paragraphs=True) + atoms = split_markdown_atoms(text) + oversize_atom = any(len(a) > limit for a in atoms) + for chunk in chunks: + # A chunk may exceed the limit only when a single indivisible atom + # (code block / table) itself exceeds it. + if len(chunk) > limit: + assert oversize_atom, ( + f"chunk of {len(chunk)} > {limit} without an oversize atom" + ) + + +@pytest.mark.parametrize("text", [FENCED, MIXED]) +def test_paragraph_mode_never_splits_closed_fences(text): + for limit in (80, 150, 300): + chunks = split_text_fence_aware(text, limit, prefer_paragraphs=True) + for chunk in chunks: + assert not text_has_unclosed_fence(chunk), ( + f"fence split mid-block at limit={limit}: {chunk!r}" + ) + + +@pytest.mark.parametrize("text", SAMPLES) +def test_paragraph_mode_no_empty_chunks(text): + for limit in (80, 400): + chunks = split_text_fence_aware(text, limit, prefer_paragraphs=True) + assert all(c for c in chunks) + + +def test_paragraph_mode_short_text_single_chunk(): + assert split_text_fence_aware("hello", 100) == ["hello"] + assert split_text_fence_aware("", 100) == [] + + +def test_paragraph_mode_utf16_len_fn(): + chunks = split_text_fence_aware(CJK, 120, utf16_len, prefer_paragraphs=True) + assert chunks + assert all(utf16_len(c) <= 120 for c in chunks) + + +# ── split_text_fence_aware (newline mode + balancing: stream_consumer) ─────── + + +@pytest.mark.parametrize("text", SAMPLES) +def test_newline_mode_balanced_fences_every_chunk(text): + chunks = split_text_fence_aware( + text, 100, prefer_paragraphs=False, balance_fences=True + ) + for chunk in chunks: + # Every delivered chunk must render standalone: even fence count. + assert chunk.count("\n```") % 2 == 0 or not text_has_unclosed_fence(chunk) + assert not text_has_unclosed_fence(chunk) + + +def test_newline_mode_balancing_reopens_language_tag(): + text = "before\n\n```python\n" + "print(1)\n" * 20 + "```\nafter" + chunks = split_text_fence_aware( + text, 80, prefer_paragraphs=False, balance_fences=True + ) + assert len(chunks) > 1 + # Some tail chunk must reopen the python fence. + assert any(c.startswith("```python\n") for c in chunks[1:]) + + +def test_newline_mode_content_preserved_without_fences(): + text = "\n".join(f"line {i} with several words in it" for i in range(40)) + chunks = split_text_fence_aware(text, 120, prefer_paragraphs=False) + joined = "\n".join(chunks) + # Newline-mode splitting only removes leading newlines at boundaries. + assert joined.replace("\n", "") == text.replace("\n", "") + + +# ── split_at_paragraph_boundary ────────────────────────────────────────────── + + +@pytest.mark.parametrize("text", SAMPLES) +def test_split_at_paragraph_boundary_head_plus_tail(text): + head, tail = split_at_paragraph_boundary(text, 100) + assert head + tail == text + assert len(head) <= 100 or "\n" not in text[:100] + + +def test_split_at_paragraph_boundary_prefers_blank_line(): + text = "para one\n\npara two\n\npara three " + "x" * 200 + head, _ = split_at_paragraph_boundary(text, 60) + assert head.endswith("\n\n") + + +def test_split_at_paragraph_boundary_cjk_sentence(): + text = "第一句话。\n第二句话！\n" + "第三句话没有结束标点一直写下去" * 20 + head, tail = split_at_paragraph_boundary(text, 30) + assert head + tail == text + assert head.endswith(("。\n", "！\n")) + + +# ── atoms ──────────────────────────────────────────────────────────────────── + + +def test_atoms_fence_kept_whole(): + atoms = split_markdown_atoms(FENCED) + fence_atoms = [a for a in atoms if a.lstrip().startswith("```")] + assert len(fence_atoms) == 1 + assert fence_atoms[0].rstrip().endswith("```") + + +def test_atoms_table_kept_whole(): + atoms = split_markdown_atoms(TABLE) + table_atoms = [a for a in atoms if a.split("\n")[0].strip().startswith("|")] + assert len(table_atoms) == 1 + assert table_atoms[0].count("\n") == 41 # header + rule + 40 rows + + +def test_atoms_nonempty_and_no_blank_lines(): + for text in SAMPLES: + for atom in split_markdown_atoms(text): + assert atom.strip() + + +# ── streaming merge + separators ───────────────────────────────────────────── + + +def test_merge_streaming_fences_rejoins_split_fence(): + chunks = ["intro\n```py\ncode line", "more code\n```\ntail"] + merged = merge_streaming_fences(chunks) + assert len(merged) == 1 + assert not text_has_unclosed_fence(merged[0]) + + +def test_merge_streaming_fences_leaves_balanced_alone(): + chunks = ["one", "```\nx\n```", "three"] + assert merge_streaming_fences(chunks) == chunks + assert merge_streaming_fences([]) == [] + + +def test_infer_block_separator_rules(): + assert infer_block_separator("text\n```", "next") == "\n" + assert infer_block_separator("text", "```py\nx") == "\n" + assert infer_block_separator("| a | b |", "| c | d |") == "\n" + assert infer_block_separator("plain", "plain") == "\n\n" + + +# ── balance_fences_across_chunks ───────────────────────────────────────────── + + +def test_balance_single_chunk_untouched(): + chunks = ["```py\nunclosed"] + assert balance_fences_across_chunks(chunks) == chunks + + +def test_balance_closes_and_reopens(): + out = balance_fences_across_chunks(["a\n```go\nx", "y\n```\nb"]) + assert out[0].endswith("\n```") + assert out[1].startswith("```go\n") + assert all(not text_has_unclosed_fence(c) for c in out) + + +# ── greedy_pack_blocks ─────────────────────────────────────────────────────── + + +def test_greedy_pack_respects_limit_and_order(): + blocks = [f"block {i} " + "w" * 30 for i in range(10)] + packed = greedy_pack_blocks(blocks, 90) + assert all(len(p) <= 90 for p in packed) + assert "\n\n".join(packed).replace("\n\n", "|").count("|") >= 0 + # Order/content preserved + assert "".join(packed).replace("\n\n", "") == "".join(blocks) + + +def test_greedy_pack_overflow_callback(): + calls = [] + + def overflow(block): + calls.append(block) + return [block[:50], block[50:]] + + packed = greedy_pack_blocks(["x" * 120], 60, overflow=overflow) + assert calls and packed == ["x" * 50, "x" * 70] + + +# ── canonical table-row splitter delegation ────────────────────────────────── + + +def test_table_row_splitters_are_unified(): + from agent.markdown_tables import split_table_row + from gateway.platforms.weixin import _split_table_row + + rows = ["| a | b | c |", "a | b | c", "|配置|状态|", " | x | ", "||"] + for row in rows: + expected = split_table_row(row) + assert split_markdown_table_row(row) == expected + assert _split_table_row(row) == expected