feat(chunking): 新增语义分块预设，支持基于嵌入和聚类的智能切分

- 添加语义分块预设选项，使用嵌入模型和聚类算法进行语义切分
- 实现标题路径增强、表格转换和图像标题处理等语义增强功能
- 新增NLTK相关依赖并持久化数据包，避免重复下载
- 添加完整的单元测试验证语义切分逻辑

backend/package/pyproject.toml

@@ -76,6 +76,12 @@ dependencies = [
     "typer>=0.16.0",
     "unstructured>=0.17.2",
     "wcmatch>=8.0.0",
+    # Markdown 解析语义切分相关依赖
+    "markdown-it-py>=3.0.0",
+    "mdit-py-plugins>=0.4.0",
+    "scikit-learn>=1.3.0",
+    "nltk>=3.8.1",
+    "beautifulsoup4>=4.12.0",
 ]
 
 [tool.setuptools]

backend/package/yuxi/knowledge/chunking/ragflow_like/dispatcher.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from yuxi.knowledge.chunking.ragflow_like.parsers import book, general, laws, qa
+from yuxi.knowledge.chunking.ragflow_like.parsers import book, general, laws, qa, semantic
 from yuxi.knowledge.chunking.ragflow_like.presets import map_to_internal_parser_id, normalize_chunk_preset_id
 
 
@@ -42,6 +42,8 @@ def _dispatch_markdown_parser(
         return book.chunk_markdown(markdown_content, parser_config)
     if parser_id == "laws":
         return laws.chunk_markdown(filename, markdown_content, parser_config)
+    if parser_id == "semantic":
+        return semantic.chunk_markdown(markdown_content, parser_config)
 
     return general.chunk_markdown(markdown_content, parser_config)
 

backend/package/yuxi/knowledge/chunking/ragflow_like/parsers/semantic.py

@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+from markdown_it import MarkdownIt
+from mdit_py_plugins.dollarmath import dollarmath_plugin
+from yuxi.utils.logging_config import logger
+from yuxi.knowledge.chunking.ragflow_like.nlp import count_tokens
+
+from ..utils.md_parser_utils import (
+    infer_heading_level, 
+    get_title_path, 
+    extract_table_block, 
+    split_text_by_length_and_newline
+)
+from ..utils.table_utils import html_table_to_key_value
+
+def _flush_content(
+    result: list, 
+    current_content: list, 
+    title_stack: list, 
+    max_length: int, 
+    embed_fn: Any,
+    special_element: str = None, 
+    allow_split: bool = False
+    ) -> None:
+    if not current_content:
+        return
+    
+    content = '\n'.join(current_content).strip()
+    if not content:
+        current_content.clear()
+        return
+
+    level = next((i + 1 for i in range(5, -1, -1) if title_stack[i]), 1)
+    title_path = get_title_path(title_stack)
+    
+    if special_element and not allow_split:
+        header = f"{'#' * level} {title_path}|{special_element}" if title_path else f"{'#' * level} {special_element}"
+        result.extend([header, content, '-' * 10])
+    else:
+        if count_tokens(content) > max_length:
+            chunks = split_text_by_length_and_newline(content, max_length, embed_fn=embed_fn, token_count_fn=count_tokens)
+            for idx, chunk in enumerate(chunks, 1):
+                base_header = f"{'#' * level} {title_path}" if title_path else f"{'#' * level}"
+                if special_element:
+                    header = f"{base_header}|{special_element}|Part {idx}"
+                else:
+                    header = f"{base_header}|Part {idx}"
+                result.extend([header, chunk, '-' * 10])
+        else:
+            base_header = f"{'#' * level} {title_path}" if title_path else f"{'#' * level}"
+            if special_element:
+                header = f"{base_header}|{special_element}"
+            else:
+                header = base_header
+            
+            if header:
+                result.append(header)
+                result.append("")
+            result.extend([content, '-' * 10])
+    
+    current_content.clear()
+
+def _handle_image_caption(tokens, i, result, current_content, title_stack, max_length, embed_fn):
+    token = tokens[i]
+    if token.type != 'paragraph_open':
+        return False, i
+        
+    inline_token = tokens[i + 1]
+    if inline_token.type != 'inline':
+        return False, i
+        
+    content = inline_token.content.strip()
+    image_pattern = r'^!\[.*?\]\(.*?\)\s*$'
+    caption_pattern = r'^(?:Figure|图|Fig\.|表|Table)\s*[\d\w\.]+'
+
+    img_match = re.search(r'^(!\[.*?\]\(.*?\))', content)
+    if img_match:
+        rest = content[img_match.end():].strip()
+        if rest and re.match(caption_pattern, rest, re.IGNORECASE):
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            current_content.append(content)
+            caption_title = rest.split('\n')[0].strip()
+            _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=caption_title)
+            return True, i + 3
+
+    if re.match(image_pattern, content):
+        next_p_idx = i + 3
+        if next_p_idx + 1 < len(tokens) and tokens[next_p_idx].type == 'paragraph_open':
+            next_inline = tokens[next_p_idx + 1]
+            if next_inline.type == 'inline':
+                next_content = next_inline.content.strip()
+                if re.match(caption_pattern, next_content, re.IGNORECASE):
+                    _flush_content(result, current_content, title_stack, max_length, embed_fn)
+                    current_content.append(content)
+                    current_content.append(next_content)
+                    _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=next_content)
+                    return True, i + 6
+
+    if current_content and re.match(caption_pattern, content, re.IGNORECASE):
+        last_item = current_content[-1].strip()
+        if re.match(image_pattern, last_item):
+            image_tag = current_content.pop()
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            current_content.append(image_tag)
+            current_content.append(content)
+            _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=content)
+            return True, i + 3
+
+    return False, i
+
+def chunk_markdown(
+    markdown_content: str, 
+    parser_config: dict[str, Any] | None = None,
+    embed_fn: Any | None = None
+) -> list[str]:
+    """
+    语义化切分 Markdown 内容。
+    
+    Args:
+        markdown_content: 待切分的 Markdown 文本
+        parser_config: 切分参数，如 chunk_token_num
+        embed_fn: 可选。传入用于生成向量的函数。如果不传，将从系统配置中加载模型。
+                  通过注入此参数可以避免在单元测试中加载重型资源。
+    """
+    parser_config = parser_config or {}
+    max_length = int(parser_config.get("chunk_token_num", 512))
+    logger.info(f"语义切分开始: max_length={max_length}, content_length={len(markdown_content)}")
+    
+    # 延迟加载重型资源，仅在没有注入 embed_fn 时触发
+    if embed_fn is None:
+        try:
+            from yuxi.config.app import config
+            from yuxi.models.embed import select_embedding_model
+
+            embed_model_id = parser_config.get("embed_model_id") or config.embed_model
+            logger.info(f"语义切分加载Embedding模型: {embed_model_id}")
+            embed_model = select_embedding_model(embed_model_id)
+            embed_fn = embed_model.encode
+        except Exception as e:
+            logger.error(f"加载 Embedding 模型失败: {e}。将退化为简单切分。")
+            embed_fn = None
+
+    md = MarkdownIt('commonmark').enable('table')
+    md.use(dollarmath_plugin, allow_space=True, allow_digits=True)
+    
+    tokens: list = md.parse(markdown_content)
+    original_lines: list = markdown_content.split('\n')
+    
+    result: list = []
+    current_content: list = []
+    title_stack: list = [''] * 6
+
+    i = 0
+    while i < len(tokens):
+        token = tokens[i]
+        if token.type == 'heading_open':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            level = int(token.tag[1:]) if token.tag and len(token.tag) > 1 else 1
+            inline_token = tokens[i + 1]
+            if inline_token.type == 'inline':
+                full_title = inline_token.content.strip()
+                title_stack[level - 1] = full_title
+                for j in range(level, 6):
+                    title_stack[j] = ''
+            i += 3
+            continue
+        elif token.type == 'table_open':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            j, table_content = extract_table_block(tokens, i, original_lines)
+            current_content.append(table_content)
+            _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element='Table')
+            i = j + 1 if j < len(tokens) else len(tokens)
+            continue
+        elif token.type == 'paragraph_open':
+            handled, new_i = _handle_image_caption(tokens, i, result, current_content, title_stack, max_length, embed_fn)
+            if handled:
+                i = new_i
+                continue
+            inline_token = tokens[i + 1]
+            if inline_token.type == 'inline':
+                current_content.append(inline_token.content.strip())
+            i += 3
+            continue
+        elif token.type == 'fence':
+            current_content.append(f"```\n{token.content}\n```")
+            i += 1
+            continue
+        elif token.type == 'ordered_list_open':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            list_content = []
+            j = i + 1
+            list_item_counter = 1
+            while j < len(tokens) and tokens[j].type != 'ordered_list_close':
+                if tokens[j].type == 'list_item_open':
+                    k = j + 1
+                    while k < len(tokens) and tokens[k].type != 'list_item_close':
+                        if tokens[k].type == 'paragraph_open' and k + 1 < len(tokens) and tokens[k + 1].type == 'inline':
+                            list_content.append(f"{list_item_counter}. {tokens[k + 1].content.strip()}")
+                            list_item_counter += 1
+                        k += 1
+                j += 1
+            if list_content:
+                current_content.extend(list_content)
+                _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=token.type)
+            i = j + 1
+            continue
+        elif token.type == 'bullet_list_open':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            list_content = []
+            j = i + 1
+            while j < len(tokens) and tokens[j].type != 'bullet_list_close':
+                if tokens[j].type == 'list_item_open':
+                    k = j + 1
+                    while k < len(tokens) and tokens[k].type != 'list_item_close':
+                        if tokens[k].type == 'paragraph_open' and k + 1 < len(tokens) and tokens[k + 1].type == 'inline':
+                            list_content.append(f"- {tokens[k + 1].content.strip()}")
+                        k += 1
+                j += 1
+            if list_content:
+                current_content.extend(list_content)
+                _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=token.type)
+            i = j + 1
+            continue
+        elif token.type == 'html_block':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            content = token.content.strip()
+            is_converted_table = False
+            if '<table' in content.lower():
+                try:
+                    kv_list = html_table_to_key_value(content)
+                    if kv_list:
+                        content = '\n'.join([f"- {item}" for item in kv_list])
+                        is_converted_table = True
+                except Exception as e:
+                    logger.warning(f"HTML表格转KV失败: {e}")
+            
+            current_content.append(content)
+            if is_converted_table:
+                _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element='Table KV', allow_split=True)
+            else:
+                _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element=token.type)
+            i += 1
+            continue
+        elif token.type in ['list_item_close', 'ordered_list_close', 'bullet_list_close', 'list_item_open']:
+            i += 1
+            continue
+        elif token.type == 'math_block':
+            _flush_content(result, current_content, title_stack, max_length, embed_fn)
+            current_content.append(f"$ {token.content} $")
+            _flush_content(result, current_content, title_stack, max_length, embed_fn, special_element='Math Block')
+            i += 1
+            continue
+        else:
+            i += 1
+
+    _flush_content(result, current_content, title_stack, max_length, embed_fn)
+
+    chunks = []
+    current_chunk_parts = []
+    for item in result:
+        if item == '-' * 10:
+            if current_chunk_parts:
+                chunks.append('\n'.join(current_chunk_parts).strip())
+                current_chunk_parts = []
+        else:
+            current_chunk_parts.append(item)
+    
+    if current_chunk_parts:
+        chunks.append('\n'.join(current_chunk_parts).strip())
+
+    logger.info(f"语义切分完成: chunks={len(chunks)}")
+    return chunks

backend/package/yuxi/knowledge/chunking/ragflow_like/presets.py

@@ -9,12 +9,14 @@ CHUNK_PRESET_GENERAL = "general"
 CHUNK_PRESET_QA = "qa"
 CHUNK_PRESET_BOOK = "book"
 CHUNK_PRESET_LAWS = "laws"
+CHUNK_PRESET_SEMANTIC = "semantic"
 
 CHUNK_PRESET_IDS = {
     CHUNK_PRESET_GENERAL,
     CHUNK_PRESET_QA,
     CHUNK_PRESET_BOOK,
     CHUNK_PRESET_LAWS,
+    CHUNK_PRESET_SEMANTIC,
 }
 
 CHUNK_PRESET_DESCRIPTIONS: dict[str, str] = {
@@ -22,6 +24,7 @@ CHUNK_PRESET_DESCRIPTIONS: dict[str, str] = {
     CHUNK_PRESET_QA: "问答分块：优先抽取问题-回答结构，适合 FAQ、题库、问答手册。",
     CHUNK_PRESET_BOOK: "书籍分块：强化章节标题识别并做层级合并，适合教材、手册、长章节文档。",
     CHUNK_PRESET_LAWS: "法规分块：按法条层级组织与合并，适合法律法规、制度规范类文本。",
+    CHUNK_PRESET_SEMANTIC: "语义分块：利用嵌入和聚类算法进行语义切分，并自动增强标题上下文。",
 }
 
 CHUNK_ENGINE_VERSION = "ragflow_like_v1"
@@ -63,6 +66,14 @@ _PRESET_DEFAULTS: dict[str, dict[str, Any] | None] = {
     CHUNK_PRESET_QA: {"raptor": {"use_raptor": False}, "graphrag": {"use_graphrag": False}},
     CHUNK_PRESET_BOOK: {"raptor": {"use_raptor": False}, "graphrag": {"use_graphrag": False}},
     CHUNK_PRESET_LAWS: {"raptor": {"use_raptor": False}, "graphrag": {"use_graphrag": False}},
+    CHUNK_PRESET_SEMANTIC: {
+        "raptor": {"use_raptor": False},
+        "graphrag": {
+            "use_graphrag": True,
+            "entity_types": ["organization", "person", "geo", "event", "category"],
+            "method": "light",
+        },
+    },
 }
 
 
@@ -235,4 +246,9 @@ def get_chunk_preset_options() -> list[dict[str, str]]:
             "label": "Laws",
             "description": CHUNK_PRESET_DESCRIPTIONS[CHUNK_PRESET_LAWS],
         },
+        {
+            "value": CHUNK_PRESET_SEMANTIC,
+            "label": "Semantic",
+            "description": CHUNK_PRESET_DESCRIPTIONS[CHUNK_PRESET_SEMANTIC],
+        },
     ]

backend/package/yuxi/knowledge/chunking/ragflow_like/utils/md_parser_utils.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import re
+from typing import List, Callable, Any
+from .semantic_utils import semantic_chunking_with_auto_clusters
+
+def infer_heading_level(title: str) -> int:
+    """
+    根据标题文本推断其层级级别（1-6级）。
+    """
+    m = re.match(r'^\s*(\d+(?:\.\d+)*)[.)、]?\s*', title)
+    if m:
+        return max(1, min(len(m.group(1).split('.')), 6))
+    m_zh = re.match(r'^\s*[一二三四五六七八九十百千]+[、.]\s*', title)
+    if m_zh:
+        return 1
+    return 1
+
+def get_title_path(stack: List[str]) -> str:
+    """
+    根据标题栈生成标题路径，用"|"分隔。
+    """
+    return '|'.join([t for t in stack if t])
+
+def extract_table_block(tokens: List[Any], i: int, original_lines: List[str]) -> tuple[int, str]:
+    """
+    从token流和原始文本中提取完整的表格块。
+    """
+    token = tokens[i]
+    table_start = token.map[0] if token.map else 0
+    j = i + 1
+    while j < len(tokens) and tokens[j].type != 'table_close':
+        j += 1
+    if j < len(tokens):
+        end_token = tokens[j]
+        if end_token.map and end_token.map[1] is not None:
+            table_end = end_token.map[1]
+        else:
+            table_end = None
+            for k in range(j + 1, len(tokens)):
+                if tokens[k].map and tokens[k].map[0] is not None:
+                    table_end = tokens[k].map[0]
+                    break
+            if table_end is None:
+                table_end = table_start + 1
+                for line_idx in range(table_start, len(original_lines)):
+                    line = original_lines[line_idx].strip()
+                    if not line or not (line.startswith('|') or '|' in line):
+                        table_end = line_idx
+                        break
+    else:
+        table_end = table_start + 1
+        for line_idx in range(table_start, len(original_lines)):
+            line = original_lines[line_idx].strip()
+            if not line or not (line.startswith('|') or '|' in line):
+                table_end = line_idx
+                break
+    return j, '\n'.join(original_lines[table_start:table_end])
+
+def split_text_by_length_and_newline(
+    text: str, 
+    max_length: int,
+    embed_fn: Callable[[List[str]], Any],
+    token_count_fn: Callable[[str], int]
+) -> List[str]:
+    """
+    层次化文本切分策略。
+    """
+    chunks = []
+    
+    paragraphs = text.split('\n\n')
+    
+    for paragraph in paragraphs:
+        paragraph = paragraph.strip()
+        if not paragraph:
+            continue
+        
+        paragraph_token_count = token_count_fn(paragraph)
+
+        if paragraph_token_count <= max_length:
+            chunks.append(paragraph)
+            continue
+        
+        lines = paragraph.split('\n')
+        current_chunk_lines = []
+        current_chunk_tokens = 0
+        
+        for line in lines:
+            line = line.strip()
+            if not line:
+                continue
+            
+            line_token_count = token_count_fn(line)
+            added_tokens = line_token_count + (1 if current_chunk_lines else 0)
+            
+            if line_token_count > max_length:
+                if current_chunk_lines:
+                    chunks.append('\n'.join(current_chunk_lines))
+                    current_chunk_lines = []
+                    current_chunk_tokens = 0
+                
+                sub_chunks = semantic_chunking_with_auto_clusters(
+                    line, 
+                    embed_fn=embed_fn, 
+                    token_count_fn=token_count_fn, 
+                    max_chunk_size=max_length
+                )
+                chunks.extend(sub_chunks)
+            
+            elif current_chunk_tokens + added_tokens > max_length:
+                chunks.append('\n'.join(current_chunk_lines))
+                current_chunk_lines = [line]
+                current_chunk_tokens = line_token_count
+            
+            else:
+                current_chunk_lines.append(line)
+                current_chunk_tokens += added_tokens
+        
+        if current_chunk_lines:
+            chunks.append('\n'.join(current_chunk_lines))
+    
+    return chunks

backend/package/yuxi/knowledge/chunking/ragflow_like/utils/semantic_utils.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import re
+import nltk
+from nltk.tokenize import sent_tokenize
+from sklearn.cluster import AgglomerativeClustering
+from sklearn.metrics import silhouette_score
+from typing import Callable, List, Any
+
+# Ensure punkt_tab is available
+try:
+    nltk.data.find('tokenizers/punkt_tab')
+except LookupError:
+    nltk.download('punkt_tab')
+
+def split_sentences_chinese(text: str) -> List[str]:
+    """
+    Split sentences by Chinese punctuation while keeping the punctuation.
+    """
+    pattern = r'(?<=[。！？])(?![”’"])|(?<=[。！？][”’"])'
+    sentences = re.split(pattern, text)
+    return [s.strip() for s in sentences if s.strip()]
+
+def split_mixed_sentences(text: str) -> List[str]:
+    """
+    Handle both Chinese and English sentence splitting.
+    """
+    chunks = re.split(r'(\n+)', text)
+    sentences = []
+
+    for ch in chunks:
+        if not ch.strip():
+            continue
+        if re.search(r'[A-Za-z]', ch):
+            parts = sent_tokenize(ch)
+            sentences.extend([p.strip() for p in parts if p.strip()])
+        else:
+            sents = split_sentences_chinese(ch)
+            if sents:
+                sentences.extend([s.strip() for s in sents if s.strip()])
+            else:
+                parts = re.split(r'(?<=[。！？])', ch)
+                sentences.extend([p.strip() for p in parts if p.strip()])
+    return sentences
+
+def find_best_num_clusters(embeddings: Any, min_clusters: int = 2, max_clusters: int = 10) -> int:
+    """
+    Select best number of clusters using silhouette score.
+    """
+    if len(embeddings) <= min_clusters:
+        return len(embeddings)
+
+    best_score = -1
+    best_k = min_clusters
+
+    limit_k = min(max_clusters, len(embeddings))
+    for k in range(min_clusters, limit_k + 1):
+        labels = AgglomerativeClustering(n_clusters=k, metric='cosine', linkage='average').fit_predict(embeddings)
+        if len(set(labels)) <= 1:
+            continue
+        score = silhouette_score(embeddings, labels, metric='cosine')
+        if score > best_score:
+            best_score = score
+            best_k = k
+
+    return best_k
+
+def semantic_chunking_with_auto_clusters(
+    text: str, 
+    embed_fn: Callable[[List[str]], Any],
+    token_count_fn: Callable[[str], int],
+    max_chunk_size: int = 512
+) -> List[str]:
+    """
+    Semantic chunking with automatic cluster number selection.
+    """
+    sentences = split_mixed_sentences(text)
+    if len(sentences) < 2:
+        return [text.strip()]
+
+    # Vectorization
+    embeddings = embed_fn(sentences)
+    
+    # Pre-calculate token counts
+    sentence_token_counts = [token_count_fn(s) for s in sentences]
+    total_tokens = sum(sentence_token_counts)
+
+    # Determine number of clusters
+    best_k = max(total_tokens // max_chunk_size, 1) + 1
+    best_k = min(best_k, len(sentences))
+    
+    # Clustering
+    labels = AgglomerativeClustering(n_clusters=best_k, metric='cosine', linkage='average').fit_predict(embeddings)
+
+    chunks = []
+    current_chunk = ""
+    current_chunk_tokens = 0
+    current_label = labels[0]
+
+    for sentence, label, token_count in zip(sentences, labels, sentence_token_counts):
+        if label != current_label or current_chunk_tokens + token_count > max_chunk_size:
+            if current_chunk.strip():
+                chunks.append(current_chunk.strip())
+            current_chunk = sentence
+            current_chunk_tokens = token_count
+            current_label = label
+        else:
+            current_chunk += sentence
+            current_chunk_tokens += token_count
+
+    if current_chunk.strip():
+        chunks.append(current_chunk.strip())
+
+    return chunks

backend/package/yuxi/knowledge/chunking/ragflow_like/utils/table_utils.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from typing import List
+from bs4 import BeautifulSoup
+from yuxi.utils.logging_config import logger
+
+def html_table_to_markdown(html: str) -> str:
+    """
+    将HTML表格转换为Markdown格式的具体实现
+    """
+    soup = BeautifulSoup(html, 'html.parser')
+    table = soup.find('table')
+    if table is None:
+        return ''
+
+    rows = table.find_all('tr')
+    if not rows:
+        return ''
+
+    grid = []
+    
+    for r_idx, row in enumerate(rows):
+        while len(grid) <= r_idx:
+            grid.append([])
+            
+        cells = row.find_all(['td', 'th'])
+        c_idx = 0
+        
+        for cell in cells:
+            while c_idx < len(grid[r_idx]) and grid[r_idx][c_idx] is not None:
+                c_idx += 1
+                
+            text = cell.get_text(strip=True)
+            text = text.replace('\n', ' ')
+            
+            rowspan = int(cell.get('rowspan', 1))
+            colspan = int(cell.get('colspan', 1))
+            
+            for r in range(rowspan):
+                target_r = r_idx + r
+                while len(grid) <= target_r:
+                    grid.append([])
+                    
+                for c in range(colspan):
+                    target_c = c_idx + c
+                    while len(grid[target_r]) <= target_c:
+                        grid[target_r].append(None)
+                    
+                    grid[target_r][target_c] = text
+            
+            c_idx += colspan
+
+    if not grid:
+        return ''
+
+    markdown_lines = []
+    max_cols = max(len(r) for r in grid)
+    for r in grid:
+        while len(r) < max_cols:
+            r.append("")
+
+    header = grid[0]
+    header = [h if h is not None else "" for h in header]
+    markdown_lines.append('| ' + ' | '.join(header) + ' |')
+    markdown_lines.append('|' + '|'.join([' --- ' for _ in range(max_cols)]) + '|')
+
+    for row in grid[1:]:
+        row_clean = [cell if cell is not None else "" for cell in row]
+        line = '| ' + ' | '.join(row_clean) + ' |'
+        markdown_lines.append(line)
+
+    return '\n'.join(markdown_lines)
+
+
+def html_table_to_key_value(html: str) -> List[str]:
+    """
+    将HTML表格转换为键值对格式的列表
+    """
+    soup = BeautifulSoup(html, 'html.parser')
+    table = soup.find('table')
+    if table is None:
+        return []
+
+    rows = table.find_all('tr')
+    if not rows:
+        return []
+
+    grid = []
+    
+    for r_idx, row in enumerate(rows):
+        while len(grid) <= r_idx:
+            grid.append([])
+            
+        cells = row.find_all(['td', 'th'])
+        c_idx = 0
+        
+        for cell in cells:
+            while c_idx < len(grid[r_idx]) and grid[r_idx][c_idx] is not None:
+                c_idx += 1
+                
+            text = cell.get_text(strip=True)
+            rowspan = int(cell.get('rowspan', 1))
+            colspan = int(cell.get('colspan', 1))
+            
+            for r in range(rowspan):
+                target_r = r_idx + r
+                while len(grid) <= target_r:
+                    grid.append([])
+                    
+                for c in range(colspan):
+                    target_c = c_idx + c
+                    while len(grid[target_r]) <= target_c:
+                        grid[target_r].append(None)
+                    grid[target_r][target_c] = text
+            c_idx += colspan
+
+    if not grid:
+        return []
+        
+    headers = grid[0]
+    headers = [h if h is not None else "" for h in headers]
+    
+    kv_lines = []
+    for row_values in grid[1:]:
+        min_len = min(len(headers), len(row_values))
+        row_parts = []
+        for i in range(min_len):
+            key = headers[i]
+            val = row_values[i] if row_values[i] is not None else ""
+            if key:
+                row_parts.append(f"{key}：{val}")
+        if row_parts:
+            kv_lines.append("；".join(row_parts) + "；")
+            
+    return kv_lines

backend/test/unit/backends/test_semantic_chunking.py

@@ -0,0 +1,78 @@
+import pytest
+import numpy as np
+
+# 现在可以安全地导入了，因为顶层不再有重型依赖
+from yuxi.knowledge.chunking.ragflow_like.parsers import semantic
+
+@pytest.fixture
+def mock_embed_fn():
+    """模拟一个简单的嵌入函数"""
+    def encode(sentences):
+        if isinstance(sentences, str):
+            sentences = [sentences]
+        # 返回模拟的向量 (N, 384)
+        return np.random.rand(len(sentences), 384).astype(np.float32)
+    return encode
+
+@pytest.fixture
+def sample_markdown():
+    return """
+# 核心业务说明
+
+本节主要介绍公司的核心业务流程。
+
+## 第一部分：研发流程
+1. 需求分析：收集用户反馈。
+2. 系统设计：架构方案评审。
+3. 编码实现：遵循规范。
+
+## 第二部分：交付标准
+| 阶段 | 交付物 | 责任人 |
+| --- | --- | --- |
+| 交付1 | 源码包 | 开发 |
+| 交付2 | 测试报告 | 测试 |
+
+这里是一些额外的语义内容。
+语义内容应当被聚类在一起。
+这也是语义内容。
+"""
+
+def test_semantic_chunking_basic(mock_embed_fn, sample_markdown):
+    """测试基本的语义切分逻辑 (使用注入方式)"""
+    
+    # 配置切分参数
+    parser_config = {
+        "chunk_token_num": 100,  # 触发语义切分
+        "overlapped_percent": 0
+    }
+    
+    # 直接注入 mock_embed_fn，完全跳过 config 加载
+    chunks = semantic.chunk_markdown(
+        sample_markdown, 
+        parser_config=parser_config, 
+        embed_fn=mock_embed_fn
+    )
+    
+    # 验证结果
+    assert isinstance(chunks, list)
+    assert len(chunks) > 0
+    
+    # 验证标题路径增强是否生效
+    has_enhanced_title = any("# 核心业务说明|第一部分：研发流程" in chunk for chunk in chunks)
+    assert has_enhanced_title, "标题路径增强失效"
+    
+    # 验证表格是否被正确保留
+    has_table = any("交付1" in chunk and "源码包" in chunk for chunk in chunks)
+    assert has_table, "表格内容丢失"
+    
+    print(f"\n成功切分为 {len(chunks)} 个片段")
+
+def test_heading_inference():
+    """测试标题层级推断工具类"""
+    from yuxi.knowledge.chunking.ragflow_like.utils.md_parser_utils import infer_heading_level
+    
+    assert infer_heading_level("1. 简介") == 1
+    assert infer_heading_level("1.1 详细设计") == 2
+    assert infer_heading_level("1.2.3 核心逻辑") == 3
+    assert infer_heading_level("一、 背景") == 1
+    assert infer_heading_level("普通文本") == 1

docker-compose.yml

@@ -52,6 +52,7 @@ services:
       - ./.env:/app/.env
       - ${MODEL_DIR:-./models}:/models  # 使用默认值处理未定义的环境变量
       - /var/run/docker.sock:/var/run/docker.sock
+      - ~/nltk_data:/root/nltk_data  # 持久化语义切分使用的NLTK数据包，避免重复下载
     ports:
       - "5050:5050"
     networks:

web/src/utils/chunk_presets.js

@@ -18,6 +18,11 @@ export const CHUNK_PRESET_OPTIONS = [
     value: 'laws',
     label: 'Laws',
     description: '法规分块：按法条层级组织与合并，适合法律法规、制度规范类文本。'
+  },
+  {
+    value: 'semantic',
+    label: 'Semantic',
+    description: '语义分块：利用嵌入和聚类算法进行语义切分，并自动增强标题上下文。'
   }
 ]