s02

文档解析

知识层

多格式文本提取与智能分块

PDF/MD/TXT 多格式支持，句子边界感知分块

分块策略直接影响知识图谱质量——按句子边界切割并保留重叠上下文，确保实体关系不被截断

📄

如何从不同格式的文档中可靠提取文本？

用户上传的种子文档可能是 PDF、Markdown 或 TXT 格式，编码可能是 UTF-8、GBK 或其他。文本提取模块需要统一处理这些差异，并将长文本智能地分割成适合知识图谱灌入的小块，同时保证实体关系不被截断。

文件输入 → 文本分块输出6 个步骤

FileParser.extract_text

按扩展名分发到 PDF/MD/TXT 提取器

_read_text_with_fallback

四级编码回退: UTF-8 → charset_normalizer → chardet → replace

_extract_from_pdf

PyMuPDF 逐页提取文本

extract_from_multiple

多文档合并，失败容错

TextProcessor.preprocess

规范化空白字符，清理多余空行

split_text_into_chunks

句子边界感知分块 + 重叠上下文

GraphRAG 知识图谱

总览

文档解析模块是流水线的第一步，负责将用户上传的多格式文件转化为结构化文本块。它由两个核心功能组成：文件解析器（FileParser）和文本分块器（split_text_into_chunks）。

1/6

Key Source Code

多级编码回退策略：UTF-8 → charset_normalizer → chardet → replace 兜底

backend/app/utils/file_parser.py

def _read_text_with_fallback(file_path: str) -> str:
    data = Path(file_path).read_bytes()
    # 首先尝试 UTF-8
    try:
        return data.decode('utf-8')
    except UnicodeDecodeError:
        pass
    # 尝试 charset_normalizer
    encoding = None
    try:
        from charset_normalizer import from_bytes
        best = from_bytes(data).best()
        if best and best.encoding:
            encoding = best.encoding
    except Exception:
        pass
    # 回退到 chardet
    if not encoding:
        try:
            import chardet
            result = chardet.detect(data)
            encoding = result.get('encoding') if result else None
        except Exception:
            pass
    # 最终兜底
    if not encoding:
        encoding = 'utf-8'
    return data.decode(encoding, errors='replace')

python

句子边界感知分块：在中英文句号处切割，保留重叠上下文

backend/app/utils/file_parser.py

def split_text_into_chunks(text, chunk_size=500, overlap=50):
    if len(text) <= chunk_size:
        return [text] if text.strip() else []
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        if end < len(text):
            # 查找最近的句子结束符
            for sep in ['。', '！', '？', '.\n', '!\n', '?\n', '\n\n', '. ', '! ', '? ']:
                last_sep = text[start:end].rfind(sep)
                if last_sep != -1 and last_sep > chunk_size * 0.3:
                    end = start + last_sep + len(sep)
                    break
        chunk = text[start:end].strip()
        if chunk:
            chunks.append(chunk)
        start = end - overlap if end < len(text) else len(text)
    return chunks

python