这是一个基于 LangChain 框架和通义千问(Qwen)大语言模型构建的检索增强生成(RAG)智能问答系统的教程。该系统是基于最基础的功能点,如大语言模型调用,文档处理,嵌入模型向量化,向量存储和检索,智能对话构建而成,最后用streamlit生成一个web界面,将功能可视化。本篇为教程第三章,之前章节请参考:

https://blog.csdn.net/zx79122564/article/details/157181513?spm=1011.2124.3001.6209

https://blog.csdn.net/zx79122564/article/details/157181554?spm=1011.2124.3001.6209

第三章 文本分割器

本节是一个专门为 Qwen 模型和中文文本优化的文本分割器,参考如下代码层级,创建文件text_splitter.py

text_splitter.py文件的完整代码如下:

# 文件: text_splitter.py
from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)
import tiktoken
from document_loader import load_documents
import os

class QwenTextSplitter:
    """针对Qwen和中文优化的文本分割器"""
    
    # chunk_size=500:每个文本块的大小(字符数);chunk_overlap=50:块之间的重叠字符数
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        # 适合中文的分割符优先级
        self.separators = ["\\n\\n", "\\n", "。", ";", ",", " ", ""]

    def recursive_split(self, documents):
        """递归字符分割(推荐)"""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=self.separators,
            length_function=len,
            is_separator_regex=False
        )
        return splitter.split_documents(documents)

    def chinese_aware_split(self, documents):
        """中文感知分割"""
        try:
            # 使用spacy进行更好的中文分割
            splitter = SpacyTextSplitter(
                chunk_size=self.chunk_size,
                chunk_overlap=self.chunk_overlap,
                separator="\\n\\n",
                pipeline="zh_core_web_sm"
            )
            return splitter.split_documents(documents)
        except:
            print("⚠️  Spacy不可用,使用递归分割")
            return self.recursive_split(documents)

    def token_based_split(self, documents):
        """基于Token的分割(适合Qwen的token限制)"""
        # Qwen-Plus支持128K上下文,但建议控制每个chunk的token数
        def tiktoken_len(text):
            try:
                encoding = tiktoken.get_encoding("cl100k_base")
                return len(encoding.encode(text))
            except:
                return len(text) // 3  # 粗略估计

        splitter = RecursiveCharacterTextSplitter(
            chunk_size=400,  # tokens
            chunk_overlap=50,
            separators=self.separators,
            length_function=tiktoken_len,
            is_separator_regex=False
        )
        return splitter.split_documents(documents)

    def analyze_chunks(self, chunks):
        """分析分割结果"""
        print("\\n📊 分割结果分析:")
        print(f"  块数量: {len(chunks)}")

        # 统计长度分布
        lengths = [len(c.page_content) for c in chunks]
        print(f"  平均长度: {sum(lengths)//len(lengths)} 字符")
        print(f"  最短: {min(lengths)} 字符")
        print(f"  最长: {max(lengths)} 字符")

        # 显示示例
        print(f"\\n📦 示例块:")
        for i in range(min(3, len(chunks))):
            chunk = chunks[i]
            content = chunk.page_content
            print(f"\\n--- 块 {i+1} ({len(content)}字符) ---")
            print(content[:100] + "..." if len(content) > 100 else content)

        return chunks


def main():
    print("🎯 第3章:文本分割(Qwen优化版)")
    print("=" * 50)

    # 加载文档
    docs = load_documents("docs/qwen_intro.txt")
    if not docs:
        print("❌ 无法加载文档")
        return

    # 创建分割器
    splitter = QwenTextSplitter(
        chunk_size=400,  # 适合Qwen的chunk大小
        chunk_overlap=50
    )

    print("\\n🔄 使用递归分割策略...")
    chunks = splitter.recursive_split(docs)
    splitter.analyze_chunks(chunks)

    # 测试Token-based分割
    print("\\n🔄 测试Token-based分割...")
    try:
        token_chunks = splitter.token_based_split(docs)
        print(f"  Token分割块数: {len(token_chunks)}")
    except Exception as e:
        print(f"  Token分割失败: {e}")

    # 保存结果
    import pickle
    if not os.path.exists("data"):
        os.makedirs("data")

    with open("data/document_chunks.pkl", "wb") as f:
        pickle.dump(chunks, f)

    print(f"\\n💾 分割结果已保存至 data/document_chunks.pkl")
    print(f"   共 {len(chunks)} 个文本块,可用于向量化")

    # 使用Qwen评估分割质量
    from qwen_client import QwenClient
    qwen = QwenClient()

    sample_chunk = chunks[0].page_content if chunks else ""
    if sample_chunk:
        evaluation = qwen.chat_completion([
            {"role": "system", "content": "你是一个文本分析专家"},
            {"role": "user", "content": f"这个文本块是否保持了语义完整性?请分析:\\n{sample_chunk}"}
        ])
        print(f"\\n🧠 Qwen分割质量评估:\\n{evaluation[:200]}...")


if __name__ == "__main__":
    main()

运行代码

python src/text_splitter.py

执行结果如下:

代码解析:

  1. 为什么需要做文本分割?
    突破模型固定长度限制:模型都有固定的上下文窗口(如4K、128K tokens),无法一次性处理超长文本
    显著降低计算成本与时间:即使模型能处理非常长的文本,所需的计算量、内存和API成本也会呈平方级增长。分割成小块可以显著降低成本,提高处理速度。
    增强信息检索与任务精度:注意力机制衰减,模型在处理长文本时,对中间部分信息的记忆和关注度会下降(“中间遗忘”问题)。分割后,模型可以在每个小片段内保持更强的注意力。
  2. 相关参数
1. chunk_size(块大小)
影响因素:
模型上下文窗口:Qwen-Plus 支持 128K tokens,但实际使用不宜过大
文本类型:技术文档、小说、新闻等不同文本有不同的最佳大小
检索效果:太小的块可能缺乏上下文,太大的块可能包含无关信息
推荐值:
# 不同场景的推荐值
场景配置 = {
    "技术文档": 400-600,      # 保持完整概念
    "小说/故事": 800-1200,    # 保持情节连贯
    "新闻/短文": 300-500,     # 单篇文章大小
    "代码/API文档": 200-400,  # 保持函数/类的完整性
    "学术论文": 600-1000,     # 保持段落完整性
}

2. chunk_overlap(重叠大小)
作用:
防止信息断裂:确保重要信息不被分割在两个块之间
上下文连贯:帮助模型理解跨块的内容关系
检索增强:提高相关文档片段的召回率

推荐值:
# 重叠比例建议
重叠比例 = chunk_size * 0.1  # 10% 的重叠
# 或者固定值:50-100 字符

3.langchain中三种常用分割方法:RecursiveCharacterTextSplitter, CharacterTextSplitter, TokenTextSplitter及中文感知分割SpacyTextSplitter,它们的核心区别在于“按什么单位来计算长度”“如何寻找切割点”

1. CharacterTextSplitter

核心理念:按字符数分割。

  • 如何工作:它最基本,直接按照你设定的 chunk_size(字符数)和 chunk_overlap(重叠字符数)进行切割。
  • 优点:极其简单、快速,不依赖任何外部库(如分词器)。
  • 缺点:非常“机械”。它几乎不关心文本结构,可能会在单词、句子甚至一个词的中间生硬地切断,破坏语义。
  • 典型使用场景:当你对分割质量要求不高,或者处理的是非标准、无清晰分隔符(如代码、日志)的文本时。

示例:设置 chunk_size=100, chunk_overlap=20

原文:“这是一个用于测试的句子。我们希望看到它如何被分割成块。”
分割:它可能会在“测试的句”后面直接切断,因为这里刚好接近100个字符,完全无视句号。

2. RecursiveCharacterTextSplitter

核心理念:递归地尝试不同分隔符,按字符数分割。

这是 CharacterTextSplitter 的智能升级版,也是目前最常用、默认推荐的方法。

  • 如何工作
  1. 它有一个分隔符优先级列表,例如:["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""](先尝试双换行,再单换行,再句号...最后是空字符,即按字符切)。
  2. 它首先尝试用最高优先级的(如\n\n)将文本分成大段。
  3. 如果某一段仍然超过 chunk_size,它就降级用下一个分隔符(如\n)继续分割这一小段。
  4. 如此递归下去,直到所有片段都小于设定的大小。
  • 优点
  • 能最大程度地尊重文本的自然边界(段落 -> 句子 -> 词语)。
  • 在保证块大小的同时,尽可能保持语义的完整性。
  • 同样不依赖外部NLP库,通用性强。
  • 缺点:仍然是基于字符长度,对于像中文、日文等语言,字符数与语义单位的对应关系不如英文单词明确。
  • 典型使用场景绝大多数通用文档(如TXT、PDF、网页文章)的预处理。是RAG应用中的首选和默认方法

示例:优先级 ["\n\n", "\n", "。"]chunk_size=150

它会先按 \n\n分段落,如果某一段超过150字,就再按 \n分,如果还超,就再按 分句。

3. TokenTextSplitter

核心理念:按模型的Token数分割。

  • 如何工作:它使用模型的分词器来计算文本的Token数量(例如,GPT系列使用tiktoken库)。然后按照你设定的 chunk_size(Token数)和 chunk_overlap(重叠Token数)进行切割。
  • 优点
    • 精准控制上下文窗口:确保分割后的文本块长度绝对不超过模型的Token限制。
    • 成本控制精确:由于LLM的API计费通常按Token数计算,用它分割可以精确预测成本。
  • 缺点
    • 需要依赖特定的分词器(如tiktokenHuggingFace tokenizer),配置稍复杂。
    • 切割点可能仍然在句子中间,因为它优先满足Token数限制。
  • 典型使用场景:当你需要精确准备输入给特定大模型(如OpenAI GPT、Claude)的文本时,或者需要严格核算Token使用量时。

关键概念:一个Token不等于一个字符或一个词。例如,英文中“tokenization”可能被分成“token”和“ization”两个Token;中文中“人工智能”可能被分成“人工”和“智能”两个Token。

4. SpacyTextSplitter

核心理念

利用句法分析找到更自然的句子边界,而不是简单的标点符号匹配。
与按固定字符数或简单标点分割不同,SpacyTextSplitter 使用 spaCy 库的自然语言处理能力来识别“真正的句子边界”,然后在这些边界处进行分割。

工作原理

  1. 加载 spaCy 语言模型:需要一个预训练的 NLP 模型来理解文本结构
  2. 句子边界检测:模型分析文本,识别完整的句子(考虑上下文,避免被缩写、小数点等迷惑)
  3. 控制块大小:将检测到的句子组合成块,确保每块的总长度(字符数或Token数)不超过设定值
  4. 保持语义连贯:尽可能不在句子中间断开,而是在完整的句子结束后分割

对中文的支持情况

重要提示:虽然称为“中文感知”,但实际效果取决于 spaCy 中文模型的质量。

优点(中文场景):

  • 智能处理标点:能区分中文句号(。)和英文句号(.),以及全角/半角符号
  • 考虑中文分句规则:理解中文中“;”、“,”有时并非句子结束,而“。”、“!”、“?”通常是句子结束
  • 处理特殊情况:能正确识别如“等等。”、“例如:”等不应用作分割的地方

局限性与挑战:

  1. 模型依赖性强:分割质量完全取决于 spaCy 中文模型(如 zh_core_web_smzh_core_web_trf)的性能
  2. 计算成本高:需要加载完整的 NLP 模型,比基于规则的分割器(如 RecursiveCharacterTextSplitter)慢得多
  3. 中文长句问题:中文常见长句用逗号连接,spaCy 可能不会在逗号处分句,导致块过大
  4. 安装复杂度:需要安装 spaCy 和对应的中文模型包

完成本节内容,即可完成对文章的切割,下一步继续做向量化和向量存储。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐