国产精品电影_久久视频免费_欧美日韩国产激情_成年人视频免费在线播放_日本久久亚洲电影_久久都是精品_66av99_九色精品美女在线_蜜臀a∨国产成人精品_冲田杏梨av在线_欧美精品在线一区二区三区_麻豆mv在线看

RAG—Chunking策略實戰

人工智能
在 RAG 系統中,即便采用性能卓越的 LLM 并反復打磨 Prompt,問答仍可能出現上下文缺失、事實性錯誤或拼接不連貫等問題。多數團隊會頻繁更換檢索算法與 Embedding模型,但收益常常有限。

一、背景

二、什么是分塊(Chunking)

三、為何要對內容做分塊處理

四、分塊策略詳解

    1. 基礎分塊

    2. 結構感知分塊

    3. 語義與主題分塊

    4. 高級分塊

    5. 混合分塊

五、結論

一、背 景

在 RAG 系統中,即便采用性能卓越的 LLM 并反復打磨 Prompt,問答仍可能出現上下文缺失、事實性錯誤或拼接不連貫等問題。多數團隊會頻繁更換檢索算法與 Embedding模型,但收益常常有限。真正的瓶頸,往往潛伏在數據入庫之前的一個細節——文檔分塊(chunking)。不當的分塊會破壞語義邊界,拆散關鍵線索并與噪聲糾纏,使被檢索的片段呈現“順序錯亂、信息殘缺”的面貌。在這樣的輸入下,再強大的模型也難以基于支離破碎的知識推理出完整、可靠的答案。某種意義上,分塊質量幾乎決定了RAG的性能上限——它決定知識是以連貫的上下文呈現,還是退化為無法拼合的碎片。

在實際場景中,最常見的錯誤是按固定長度生硬切割,忽略文檔的結構與語義:定義與信息被切開、表頭與數據分離、步驟說明被截斷、代碼與注釋脫節,結果就是召回命中卻無法支撐結論,甚至誘發幻覺與錯誤引用。相反,高質量的分塊應盡量貼合自然邊界(標題、段落、列表、表格、代碼塊等),以適度重疊保持上下文連續,并保留必要的來源與章節元數據,確保可追溯與重排可用。當分塊尊重文檔的敘事與結構時,檢索的相關性與答案的事實一致性往往顯著提升,遠勝于一味更換向量模型或調參;換言之,想要真正改善 RAG 的穩健性與上限,首先要把“知識如何被切開并呈現給模型”這件事做好。

PS:本文主要是針對中文文檔類型的嵌入進行實戰。

二、什么是分塊(Chunking)

分塊是將大塊文本分解成較小段落的過程,這使得文本數據更易于管理和處理。通過分塊,我們能夠更高效地進行內容嵌入(embedding),并顯著提升從向量數據庫中召回內容的相關性和準確性。

在實際操作中,分塊的好處是多方面的。首先,它能夠提高模型處理的效率,因為較小的文本段落更容易進行嵌入和檢索。

其次,分塊后的文本能夠更精確地匹配用戶查詢,從而提供更相關的搜索結果。這對于需要高精度信息檢索和內容生成的應用程序尤為重要。

通過優化內容的分塊和嵌入策略,我們可以最大化LLM在各種應用場景中的性能。分塊技術不僅提高了內容召回的準確性,還提升了整體系統的響應速度和用戶體驗。

因此,在構建和優化基于LLM的應用程序時,理解和應用分塊技術是不可或缺的步驟。

分塊過程中主要的兩個概念:chunk_size塊的大小,chunk_overlap重疊窗口。

圖片圖片

三、為何要對內容做分塊處理

  • 模型上下文窗口限制:LLM無法一次處理超長文本。分塊的目的在于將長文檔切成模型可穩定處理的中等粒度片段,并盡量對齊自然語義邊界(如標題、段落、句子、代碼塊),避免硬切導致關鍵信息被截斷或語義漂移。即便使用長上下文模型,過長輸入也會推高成本并稀釋信息密度,合理分塊仍是必需的前置約束。
  • 檢索的信噪比:塊過大時無關內容會稀釋信號、降低相似度判別力;塊過小時語境不足、容易“只命中詞不命中義”。合適的塊粒度可在召回與精度間取得更好平衡,既覆蓋用戶意圖,又不引入多余噪聲。在一定程度上提升檢索相關性的同時又能保證結果穩定性。
  • 語義連續性:跨段落或跨章節的語義關系常在邊界處被切斷。通過設置適度的 chunk_overlap,可保留跨塊線索、減少關鍵定義/條件被“切開”的風險。對于強結構文檔,優先讓邊界貼合標題層級與句子斷點;必要時在檢索階段做輕量鄰近擴展,以提升答案的連貫性與可追溯性,同時避免重復內容擠占上下文預算。

總之理想的分塊是在“上下文完整性”和“信息密度”之間取得動態平衡:chunk_size決定信息承載量,chunk_overlap 用于彌補邊界斷裂并維持語義連續。只要邊界對齊語義、粒度貼合內容,檢索與生成的質量就能提升。

圖片圖片

四、分塊策略詳解

基礎分塊

基于固定長度分塊

  • 分塊策略:按預設字符數 chunk_size 直接切分,不考慮文本結構。
  • 優點:實現最簡單、速度快、對任意文本通用。
  • 缺點:容易破壞語義邊界;塊過大容易引入較多噪聲,過小則會導致上下文不足。
  • 適用場景:結構性弱的純文本,或數據預處理初期的基線方案。
from langchain_text_splitters import CharacterTextSplitter


splitter = CharacterTextSplitter(
    separator="",        # 純按長度切
    chunk_size=600,      # 依據實驗與模型上限調整
    chunk_overlap=90,    # 15% 重疊
)
chunks = splitter.split_text(text)
  • 參數建議(僅限中文語料建議):

chunk_size:300–800 字優先嘗試;若嵌入模型最佳輸入為 512/1024 tokens,可折算為約 350/700 中文字符起步。

chunk_overlap:10%–20% 起步;超過 30% 通常導致索引體積與檢索開銷顯著上升,對實際性能起負作用,最后的效果并不會得到明顯提升。

基于句子的分塊

  • 分塊策略:先按句子切分,再將若干句子聚合成滿足chunk_size的塊;保證最基本的語義完整性。
  • 優點:句子級完整性最好。對問句/答句映射友好。便于高質量引用。
  • 缺點:中文分句需特別處理。僅句子級切分可能導致塊過短,需后續聚合。
  • 適用場景:法律法規、新聞、公告、FAQ 等以句子為主的文本。
  • 中文分句注意事項:

不要直接用 NLTK 英文 Punkt:無法識別中文標點,分句會失敗或異常。

可以直接使用以下內容進行分句:

1)基于中文標點的正則:按“。?。?;”等切分,保留引號與省略號等邊界。

2)使用支持中文的 NLP 庫進行更精細的分句:

3)HanLP(推薦,工業級,支持繁多語言學特性)Stanza(清華/斯坦福合作,中文支持較好)spaCy + pkuseg 插件(或 zh-core-web-sm/med/lg 生態)

  • 示例(適配常見中文標點,基于正則的分句):
import re


def split_sentences_zh(text: str):
    # 在句末標點(。???;)后面帶可選引號的場景斷句
    pattern = re.compile(r'([^。!?;]*[。???;]+|[^。?。浚籡+$)')
    sentences = [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]
    return sentences


def sentence_chunk(text: str, chunk_size=600, overlap=80):
    sents = split_sentences_zh(text)
    chunks, buf = [], ""
    for s in sents:
        if len(buf) + len(s) <= chunk_size:
            buf += s
        else:
            if buf:
                chunks.append(buf)
            # 簡單重疊:從當前塊尾部截取 overlap 字符與下一句拼接
            buf = (buf[-overlap:] if overlap > 0 and len(buf) > overlap else "") + s
    if buf:
        chunks.append(buf)
    return chunks


chunks = sentence_chunk(text, chunk_size=600, overlap=90)

HanLP 分句示例:

from hanlp_common.constant import ROOT
import hanlp


tokenizer = hanlp.load('PKU_NAME_MERGED_SIX_MONTHS_CONVSEG')  # 或句法/句子級管線
# HanLP 高層 API 通常通過句法/語料管線獲得句子邊界,具體以所用版本 API 為準
# 將句子列表再做聚合為 chunk_size

基于遞歸字符分塊

  • 分塊策略:給定一組由“粗到細”的分隔符(如段落→換行→空格→字符),自上而下遞歸切分,在不超出 chunk_size 的前提下盡量保留自然語義邊界。
  • 優點:在“保持語義邊界”和“控制塊大小”之間取得穩健平衡,對大多數文本即插即用。
  • 缺點:分隔符配置不當會導致塊粒度失衡,極度格式化文本(表格/代碼)效果一般。
  • 適用場景:綜合性語料、說明文檔、報告、知識庫條目。
import re
from langchain_text_splitters import RecursiveCharacterTextSplitter


separators = [
    r"\n#{1,6}\s",                 # 標題
    r"\n\d+(?:\.\d+)*\s",          # 數字編號標題 1. / 2.3. 等
    "\n\n",                        # 段落
    "\n",                          # 行
    " ",                           # 空格
    "",                            # 兜底字符級
]
splitter = RecursiveCharacterTextSplitter(
    separators=separators,
    chunk_size=700,
    chunk_overlap=100,
    is_separator_regex=True,       # 告訴分割器上面包含正則
)
chunks = splitter.split_text(text)
  • 參數與分隔符建議(僅中文文檔建議):

chunk_size:400–800 字符;如果內容更技術化、長句多時可適當上調該數值。

chunk_overlap:10%–20%。

separators(由粗到細,按需裁剪):

1)章節/標題:正則 r"^#{1,6}\s"(Markdown 標題)、r"^\d+(.\d+)*\s"(編號標題)

2)段落:"\n\n"

3)換行:"\n"

4)空格:" "

5)兜底:""

總結

  • 調優流程:

固定檢索與重排,只動分塊參數。

用驗證集計算 Recall@k、nDCG、MRR、來源命中文檔覆蓋率、答案事實性(faithfulness)。

觀察塊長分布:若長尾太長,適當收緊chunk_size 或增加粗粒度分隔符;若過短,放寬chunk_size 或降低分隔符優先級。

  • 重疊的成本與收益:

收益:緩解邊界斷裂,提升答案連貫性與可追溯性。

成本:索引尺寸增長、召回重復塊增多、rerank 負載提升。通常不建議超過 20%–25%。

  • 組合技巧:

先遞歸分塊,再對“異常長句”或“跨段引用”場景加一點點額外 overlap。

對標題塊注入父級標題上下文,提高定位能力與可解釋性。

  • 何時切換策略:

若問答頻繁丟上下文或引用斷裂:增大overlap或改用句子/結構感知策略。

若召回含噪過多:減小 chunk_size 或引入更強的結構分隔符。

結構感知分塊

利用文檔固有結構(標題層級、列表、代碼塊、表格、對話輪次)作為分塊邊界,邏輯清晰、可追溯性強,能在保證上下文完整性的同時提升檢索信噪比。

結構化文本分塊
  • 分塊策略以標題層級(H1–H6、編號標題)或語義塊(段落、列表、表格、代碼塊)為此類型文檔的天然邊界,對過長的結構塊再做二次細分,對過短的進行相鄰合并。
  • 實施步驟

解析結構:Markdown 用解析器remark/markdown-it-py或正則識別標題與語塊;HTML用 DOMBeautifulSoup/Cheerio遍歷 Hx、p、li、pre、table 等。

生成章節:以標題為父節點,將其后的連續兄弟節點納入該章節,直至遇到同級或更高層級標題。

二次切分:章節超出 chunk_size時,優先按子標題/段落切,再不足時按句子或遞歸字符切分。

合并短塊:低于 min_chunk_chars 的塊與相鄰塊合并,優先與同一父標題下的前后塊。

上下文重疊:優先用“結構重疊”(父級標題路徑、前一小節標題+摘要),再輔以小比例字符overlap(10%–15%)。

寫入 metadata。

  • 示例代碼
import re
from typing import List, Dict


heading_pat = re.compile(r'^(#{1,6})\s+(.*)$')  # 標題
fence_pat = re.compile(r'^```')                 # fenced code fence


def split_markdown_structure(text: str, chunk_size=900, min_chunk=250, overlap_ratio=0.1) -> List[Dict]:
    lines = text.splitlines()
    sections = []
    in_code = False
    current = {"level": 0, "title": "", "content": [], "path": []}
    
    path_stack = []  # [(level, title)]
    
    for ln in lines:
        if fence_pat.match(ln):
            in_code = not in_code
        m = heading_pat.match(ln) if not in_code else None
        if m:
            if current["content"]:
                sections.append(current)
            level = len(m.group(1))
            title = m.group(2).strip()


            while path_stack and path_stack[-1][0] >= level:
                path_stack.pop()
            path_stack.append((level, title))
            breadcrumbs = [t for _, t in path_stack]
            current = {"level": level, "title": title, "content": [], "path": breadcrumbs}
        else:
            current["content"].append(ln)
    
    if current["content"]:
        sections.append(current)
    
    # 通過二次拆分/合并將部分平鋪成塊
    chunks = []
    def emit_chunk(text_block: str, path: List[str], level: int):
        chunks.append({
            "text": text_block.strip(),
            "meta": {
                "section_title": path[-1] if path else "",
                "breadcrumbs": path,
                "section_level": level,
            }
        })
    
    for sec in sections:
        raw = "\n".join(sec["content"]).strip()
        if not raw:
            continue
        if len(raw) <= chunk_size:
            emit_chunk(raw, sec["path"], sec["level"])
        else:
            paras = [p.strip() for p in raw.split("\n\n") if p.strip()]
            buf = ""
            for p in paras:
                if len(buf) + len(p) + 2 <= chunk_size:
                    buf += (("\n\n" + p) if buf else p)
                else:
                    if buf:
                        emit_chunk(buf, sec["path"], sec["level"])
                    buf = p
            if buf:
                emit_chunk(buf, sec["path"], sec["level"])
    
    merged = []
    for ch in chunks:
        if not merged:
            merged.append(ch)
            continue
        if len(ch["text"]) < min_chunk and merged[-1]["meta"]["breadcrumbs"] == ch["meta"]["breadcrumbs"]:
            merged[-1]["text"] += "\n\n" + ch["text"]
        else:
            merged.append(ch)
    
    overlap = int(chunk_size * overlap_ratio)
    for ch in merged:
        bc = " > ".join(ch["meta"]["breadcrumbs"][-3:])
        prefix = f"[{bc}]\n" if bc else ""
        if prefix and not ch["text"].startswith(prefix):
            ch["text"] = prefix + ch["text"]
        # optional character overlap can在檢索階段用鄰接聚合替代,這里略
    
    return merged
  • 參數建議(中文文檔)

chunk_size:600–1000 字;技術文/長段落可取上限,繼續適當增加。

min_chunk_chars:200–300 字(小于則合并)。

chunk_overlap:10%–15%;若使用“父級標題路徑 + 摘要”作為結構重疊,可降至 5%–10%。

對話式分塊

  • 分塊策略

以“輪次/說話人”為邊界,優先按對話鄰接對和小段話題窗口聚合。重疊采用“輪次重疊”而非單純字符重疊,保證上下文流暢。

  • 適用場景

客服對話、訪談、會議紀要、技術支持工單等多輪交流。

  • 檢索期鄰

接聚合在檢索階段對對話塊做“鄰接擴展”:取被召回塊前后各 1–2 輪上下文(或相鄰塊拼接)作為最終送審上下文,以提高回答連貫性與可追溯性。

  • 與重排協同

可提升對“誰說的、在哪段說的”的判斷力。

  • 示例代碼:(按輪次滑動窗口分塊)
from typing import List, Dict


def chunk_dialogue(turns: List[Dict], max_turns=10, max_chars=900, overlap_turns=2):
    """
    turns: [{"speaker":"User","text":"..." , "ts_start":123, "ts_end":130}, ...]
    """
    chunks = []
    i = 0
    while i < len(turns):
        j = i
        char_count = 0
        speakers = set()
        while j < len(turns):
            t = turns[j]
            uttr_len = len(t["text"])
            # 若單條超長,允許在句級二次切分(此處略),但不跨 speaker
            if (j - i + 1) > max_turns or (char_count + uttr_len) > max_chars:
                break
            char_count += uttr_len
            speakers.add(t["speaker"])
            j += 1
        
        if j > i:
            window = turns[i:j]
        elif i < len(turns):
            window = [turns[i]]
        else:
            break
        text = "\n".join([f'{t["speaker"]}: {t["text"]}' for t in window])
        meta = {
            "speakers": list(speakers),
            "turns_range": (i, j - 1),
            "ts_start": window[0].get("ts_start"),
            "ts_end": window[-1].get("ts_end"),
        }
        chunks.append({"text": text, "meta": meta})
        
        # 按輪次重疊回退
        if j >= len(turns):
            break
        next_start = i + len(window) - overlap_turns
        i = max(next_start, i + 1)  # 確保至少前進1步
    return chunks
  • 參數建議

max_turns_per_chunk:6–12 輪起步;語速快信息密度高可取 8–10。

max_chars_per_chunk:600–1000 字;若存在長段獨白,優先句級再切,不跨說話人。

overlap_turns:1–2 輪;保證上一問下一答的連續性。

keep_pairing:不要拆開明顯的問答對;若 chunk 臨界,寧可擴一輪或后移切分點。

總結

  • 首選用結構邊界做第一次切分,再用句級/遞歸策略做二次細分。
  • 優先使用“結構重疊”(父標題路徑、上段標題+摘要、相鄰發言)替代大比例字符重疊。
  • 為每個塊寫好 metadata,可顯著提升檢索質量與可解釋性。
  • 對 PDF/HTML 先去噪(頁眉頁腳、導航、廣告等),避免把噪聲索引進庫。

語義與主題分塊

該方法不依賴文檔的物理結構,而是依據語義連續性與話題轉移來決定切分點,尤其適合希望“塊內高度內聚、塊間清晰分界”的知識庫與研究類文本。

語義分塊

  • 分塊策略

對文本先做句級切分,計算句子或短段的向量表示;

當相鄰語義的相似度顯著下降(發生“語義突變”)時設為切分點。

  • 適用場景

專題化、論證結構明顯的文檔:

白皮書、論文、技術手冊、FAQ 聚合頁;

需要高內聚檢索與高可追溯性。

  • 使用流程

句級切分:先用中文分句(標點/中文分句模型)得到句子序列。

向量化:對每個句子編碼,開啟歸一化(normalize)以便用余弦相似度。

突變檢測:

1)簡單粗暴的方法:sim(i, i-1) 低于閾值則切分。

2)穩健的方法:與“前后窗口的均值向量”比較,計算新穎度 novelty = 1 - cos(emb_i, mean_emb_window),新穎度高于閾值則切分。

3)平滑的方法:對相似度/新穎度做移動平均,降低抖動。

約束與修正:設置最小/最大塊長,避免過碎或過長,必要時進行相鄰塊合并。

  • 與檢索/重排的協同召回時可做“鄰接擴展”(把被命中的塊前后各追加一兩句),再做重排序。語義分塊的高內聚可讓 重排序更精準地區分相近候選。
  • 代碼示例
from typing import List, Dict, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer
import re


def split_sentences_zh(text: str) -> List[str]:
    # 簡易中文分句,可替換為 HanLP/Stanza 更穩健的實現
    pattern = re.compile(r'([^。???;]*[。???;]+|[^。!?;]+$)')
    return [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]


def rolling_mean(vecs: np.ndarray, i: int, w: int) -> np.ndarray:
    s = max(0, i - w)
    e = min(len(vecs), i + w + 1)
    return vecs[s:e].mean(axis=0)


def semantic_chunk(
    text: str,
    model_name: str = "BAAI/bge-m3",
    window_size: int = 2,
    min_chars: int = 350,
    max_chars: int = 1100,
    lambda_std: float = 0.8,
    overlap_chars: int = 80,
) -> List[Dict]:
    sents = split_sentences_zh(text)
    if not sents:
        return []
    
    model = SentenceTransformer(model_name)
    emb = model.encode(sents, normalize_embeddings=True, batch_size=64, show_progress_bar=False)
    emb = np.asarray(emb)
    
    # 基于窗口均值的“新穎度”分數
    novelties = []
    for i in range(len(sents)):
        ref = rolling_mean(emb, i-1, window_size) if i > 0 else emb[0]
        ref = ref / (np.linalg.norm(ref) + 1e-8)
        novelty = 1.0 - float(np.dot(emb[i], ref))
        novelties.append(novelty)
    novelties = np.array(novelties)
    
    # 相對閾值:μ + λσ
    mu, sigma = float(novelties.mean()), float(novelties.std() + 1e-8)
    threshold = mu + lambda_std * sigma
    
    chunks, buf, start_idx = [], "", 0
    def flush(end_idx: int):
        nonlocal buf, start_idx
        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {"start_sent": start_idx, "end_sent": end_idx-1}
            })
        buf, start_idx = "", end_idx
    
    for i, s in enumerate(sents):
        # 若超長則先沖洗
        if len(buf) + len(s) > max_chars and len(buf) >= min_chars:
            flush(i)
            # 結構化重疊:附加上一個塊的尾部
            if overlap_chars > 0 and len(s) < overlap_chars:
                buf = s
                continue
        
        buf += s
        
        # 達到最小長度后遇到突變則切分
        if len(buf) >= min_chars and novelties[i] > threshold:
            flush(i + 1)
    
    if buf:
        flush(len(sents))
    
    return chunks
  • 參數調優說明(僅作參考)

閾值的含義:語義變化敏感度控制器,越低越容易切、越高越保守。

設定方式:

1)絕對閾值:例如使用余弦相似度,若 sim < 0.75 則切分(需按語料校準)。

2)相對閾值:對全篇的相似度/新穎度分布估計均值μ與標準差σ,使用 μ ± λσ 作為閾值,更穩健。

初始的配置建議(僅限于中文技術/說明文檔):

       1)窗口大小 window_size:2–4 句

       2)最小/最大塊長:min_chunk_chars=300–400,max_chunk_chars=1000–1200

       3)閾值策略:novelty > μ + 0.8σ 或相似度 < μ - 0.8σ(先粗調后微調)

       4)overlap:10% 左右或按“附加上一句”做輕量輪次重疊

主題的分塊

  • 分塊策略

利用主題模型或聚類算法在“宏觀話題”發生切換時進行切分,更多的關注章節級、段落級的主題邊界。該類分塊策略主要適合長篇、多主題材料。

  • 適用場景

報告、書籍、長調研文檔、綜合評審;

當文檔內部確有較穩定的“話題塊”。

  • 使用流程(最好用“句向量 + 聚類 + 序列平滑”而非純 LDA)

句級切分并編碼:首先通過向量模型得到句向量,normalize。

文檔內或語料級聚類:

1)文檔內小規模:MiniBatchKMeans(k=3–8 先驗)或 SpectralClustering。

2)語料級統一主題:在大量文檔上聚類(或用 HDBSCAN+UMAP),再將每篇文檔的句子映射到最近主題中心。

序列平滑與解碼:

      1) 對句子的主題標簽做滑窗多數投票或一階馬爾可夫平滑,避免頻繁抖動。

      2)當主題標簽穩定變化并滿足最小塊長時,設為切分點。

主題命名:用 KeyBERT/TF-IDF 在每個塊內抽關鍵詞,或用小模型生成一句話主題摘要,寫入 metadata。

約束:min/max_chars,保留代碼/表格等原子塊,必要時與結構邊界結合使用。

  • 代碼示例(KMeans 文檔內聚類 + 序列平滑)
from typing import List, Dict
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans
import re


def split_sentences_zh(text: str) -> List[str]:
    pattern = re.compile(r'([^。???;]*[。???;]+|[^。???;]+$)')
    return [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]


def topic_chunk(
    text: str,
    k_topics: int = 5,
    min_chars: int = 500,
    max_chars: int = 1400,
    smooth_window: int = 2,
    model_name: str = "BAAI/bge-m3"
) -> List[Dict]:
    sents = split_sentences_zh(text)
    if not sents:
        return []
    
    model = SentenceTransformer(model_name)
    emb = model.encode(sents, normalize_embeddings=True, batch_size=64, show_progress_bar=False)
    emb = np.asarray(emb)
    
    km = KMeans(n_clusters=k_topics, n_init="auto", random_state=42)
    labels = km.fit_predict(emb)
    
    # 簡單序列平滑:滑窗多數投票
    smoothed = labels.copy()
    for i in range(len(labels)):
        s = max(0, i - smooth_window)
        e = min(len(labels), i + smooth_window + 1)
        window = labels[s:e]
        vals, counts = np.unique(window, return_counts=True)
        smoothed[i] = int(vals[np.argmax(counts)])
    
    chunks, buf, start_idx, cur_label = [], "", 0, smoothed[0]
    def flush(end_idx: int):
        nonlocal buf, start_idx
        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {"start_sent": start_idx, "end_sent": end_idx-1, "topic": int(cur_label)}
            })
        buf, start_idx = "", end_idx
    
    for i, s in enumerate(sents):
        switched = smoothed[i] != cur_label
        over_max = len(buf) + len(s) > max_chars
        under_min = len(buf) < min_chars
        
        # 嘗試延后切分,保證最小塊長
        if switched and not under_min:
            flush(i)
            cur_label = smoothed[i]
        
        if over_max and not under_min:
            flush(i)
        
        buf += s
    
    if buf:
        flush(len(sents))
    
    return chunks
  • 一些參數對結果的影響

k(主題數):難以精準預設,可通過輪廓系數(silhouette)/肘部法初篩,再結合領域先驗與人工校正。

HDBSCAN:min_cluster_size 影響較大,過小會碎片化,過大則合并不同話題。

min_topic_span_sents:如 5–8 句,防止標簽抖動導致過密切分。

小文檔不宜用:樣本太少時主題不可分,優先用語義分塊或結構分塊。

高級分塊

小-大分塊

  • 分塊策略用“小粒度塊”(如句子/短句)做高精度召回,定位到最相關的微片段;再將其“所在的大粒度塊”(如段落/小節)作為上下文送入 LLM,以兼顧精確性與上下文完整性。
  • 使用流程

構建索引(離線):

1)Sentence/短句索引(索引A):單位為句子或子句。

2)段落/小節存儲(存儲B):保留原始大塊文本與結構信息。

檢索(在線):

      1)用索引A召回 top_k_small 個小塊(向量檢索)。

      2)將小塊按 parent_id 分組,計算組內分數(max/mean/加權),選出 top_m_big 個父塊候選。

      3)對“查詢-父塊文本”做交叉編碼重排,提升相關性排序的穩定性。

      4)上下文組裝:在每個父塊中高亮或優先保留命中小句附近的上下文(鄰近N句或窗口字符 w),在整體 token 預算內拼接多塊。

      5)代碼示例(偽代碼)

# 離線:構建小塊索引,并保存 parent_id -> 大塊文本 的映射
# 在線檢索:
small_hits = small_index.search(embed(query), top_k=30)
groups = group_by_parent(small_hits)
scored_parents = score_groups(groups, agg="max")
candidates = top_m(scored_parents, m=3)


# 交叉編碼重排
rerank_inputs = [(query, parent_text(pid)) for pid in candidates]
reranked = cross_encoder_rerank(rerank_inputs)


# 組裝上下文:對每個父塊,僅保留命中句及其鄰近窗口,并加上標題路徑
contexts = []
for pid, _ in reranked:
    hits = groups[pid]
    context = build_local_window(parent_text(pid), hits, window_sents=1)
    contexts.append(prefix_with_breadcrumbs(pid) + context)


final_context = pack_under_budget(contexts, token_budget=3000)    # 留出回答空間

父子段分塊

  • 分塊策略

將文檔按章節/段落等結構單元切成“父塊”(Parent),再在每個父塊內切出“子塊”(通常為句子/短段或者篤固定塊)。然后為“子塊”建向量索引以做高精度召回。當檢索時先召回子塊,再按 parent_id 聚合并擴展到父塊或父塊中的局部窗口,兼顧最后召回內容的精準與上下文完整性。

  • 適用場景

結構清晰的說明文、手冊、白皮書、法規、FAQ 聚合頁;

需要“句級證據準確 + 段/小節級上下文完整”的問答。

  • 使用流程

結構粗切(父塊)

1)按標題層級/段落/代碼塊切出父塊。

2)父塊寫入 breadcrumbs(H1/H2/...)、anchor、block_type、start/end_offset。

精細切分(子塊)

      1)在父塊內部以句子/子句/固定塊為單位切分(可用遞歸分塊兜底),小比例 overlap(或附加上一句內容)。

      2)為每個子塊記錄child_offset、sent_index_range、parent_id。

建索引與存儲

      1)子塊向量索引A:先編碼,normalize 后建索引。

      2) 父塊存儲B:保存原文與結構元信息,此處可以選建一個父塊級向量索引用于粗排或回退。

檢索與組裝

      1 ) 用索引A召回 top_k_child 子塊。

      2 ) 按 parent_id 分組并聚合打分(max/mean/命中密度),選出 top_m_parent 父塊候選。

      3 ) 對 (query, parent_text 或 parent_window) 交叉編碼重排。

      4)上下文裁剪:對每個父塊僅保留“命中子塊±鄰近窗口”(±1–2 句或 80–200 字),加上標題路徑前綴,控制整體 token 預算。

打分與聚合策略

1)組分數:score_parent = α·max(child_scores) + (1-α)·mean(child_scores) + β·coverage(命中子塊數/父塊子塊總數)。

2)密度歸一化:density = sum(exp(score_i)) / length(parent_text),為避免長父塊因命中多而“天然占優”。

3)窗口合并:同一父塊內相鄰命中窗口若間距小于閾值則合并,減少重復與碎片。

  • 與“小-大分塊”的關系

1)小-大分塊是檢索工作流(小粒度召回→大粒度上下文);

2)父子段分塊是數據建模與索引設計(顯式維護 parent–child 映射)。

3)兩者強相關、常配合使用:父子映射讓小-大擴展更穩、更易去重與回鏈。

  • 示例
from typing import List, Dict, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer




embedder = SentenceTransformer("BAAI/bge-m3")


def search_parent_child(query: str, top_k_child=40, top_m_parent=3, window_chars=180):
    q = embedder.encode([query], normalize_embeddings=True)[0]
    hits = small_index.search(q, top_k=top_k_child)  # 返回 [(child_id, score), ...]
    # 分組
    groups: Dict[str, List[Tuple[str, float]]] = {}
    for cid, score in hits:
        p = child_parent_id[cid]
        groups.setdefault(p, []).append((cid, float(score)))
    
    # 聚合打分(max + coverage)
    scored = []
    for pid, items in groups.items():
        scores = np.array([s for _, s in items])
        agg = 0.7 * scores.max() + 0.3 * (len(items) / (len(parents[pid]["sent_spans"]) + 1e-6))
        scored.append((pid, float(agg)))
    scored.sort(key=lambda x: x[1], reverse=True)
    candidates = [pid for pid, _ in scored[:top_m_parent]]
    
    # 為每個父塊構造“命中窗口”
    contexts = []
    for pid in candidates:
        ptext = parents[pid]["text"]
        # 找到子塊命中區間并合并窗口
        spans = sorted([(children[cid]["start"], children[cid]["end"]) for cid, _ in groups[pid]])
        merged = []
        for s, e in spans:
            s = max(0, s - window_chars)
            e = min(len(ptext), e + window_chars)
            if not merged or s > merged[-1][1] + 50:
                merged.append([s, e])
            else:
                merged[-1][1] = max(merged[-1][1], e)
        windows = [ptext[s:e] for s, e in merged]
        prefix = " > ".join(parents[pid]["meta"].get("breadcrumbs", [])[-3:])
        contexts.append((pid, f"[{prefix}]\n" + "\n...\n".join(windows)))
    
    # 交叉編碼重排(此處用占位函數)
    reranked = cross_encoder_rerank(query, [c[1] for c in contexts])  # 返回 indices 順序
    ordered = [contexts[i] for i in reranked]
    return ordered  # [(parent_id, context_text), ...]
  • 調參建議(僅作參考,具體需要按照實際來)調參順序:先定父/子塊長度 → 標定 top_k_child 與聚合權重 → 調整窗口大小與合并閾值 → 最后接入交叉編碼重排并控制 token 預算。

代理式分塊

  • 分塊策略

使用一個小溫度、強約束的 LLM Agent 模擬“人類閱讀與編排”,根據語義、結構與任務目標動態決定分塊邊界,并輸出結構化邊界信息與理由(rationale 可選,不用于檢索)。

  • 適用場景

高度復雜、長篇、非結構化且混合格式(文本+代碼+表格)的文檔;

結構/語義/主題策略單獨使用難以取得理想邊界時。

  • 使用時的注意事項

規則護欄:

1)禁止在代碼塊、表格單元、引用塊中間切分,對圖片/公式作為原子單元處理。

2)保持標題鏈路完整,強制最小/最大塊長(min/max_chars / min/max_sents)。

目標對齊:在系統提示中明確“為了檢索問答/用于摘要/用于診斷”的目標,Agent 以任務優先級決定邊界與上下文冗余度。

1)結構化輸出:要求輸出 segments: [{start_offset, end_offset, title_path, reason}],不能接受自由文本。

2)自檢與回退:Agent 產出的邊界先過一遍約束校驗器(如長度、原子塊、順序等),不符合規則的內容則自動回退到遞歸/句級分塊。

成本控制:

     1 )長文分批閱讀(分段滑動窗口);

     2 ) 在每段末尾只輸出邊界草案,最終匯總并去重;

     3 )溫度低(≤0.3)、max_tokens 受控。

     4) 示例:Agent 輸出模式(偽 Prompt 片段)

系統:你是分塊器。目標:為RAG檢索創建高內聚、可追溯的塊。規則:
1) 不得在代碼/表格/公式中間切分;
2) 每塊400-1000字;
3) 保持標題路徑完整;
4) 盡量讓“定義+解釋”在同一塊;
5) 輸出JSON,含 start_offset/end_offset/title_path。


用戶:<文檔片段文本>
助手(示例輸出):
{
  "segments": [
    {"start": 0, "end": 812, "title_path": ["指南","安裝"], "reason": "完整步驟+注意事項"},
    {"start": 813, "end": 1620, "title_path": ["指南","配置"], "reason": "參數表與示例緊密相關"}
  ]
}
  • 集成的流程

粗切:先用結構感知/遞歸策略獲得初步塊,降低 Agent 處理跨度。

Agent 精修:對“疑難塊”(過長/多格式/主題混雜)調用 Agent 細化邊界。

質檢:規則校驗 + 語義稀疏度檢測(塊內相似度方差過大則再細分)。

寫入 metadata。

混合分塊

單一策略難覆蓋所有文檔與場景?;旌戏謮K通過“先粗后細、按需細化”,在效率、可追溯性與答案質量之間取得穩健平衡。

  • 分塊策略

先用宏觀邊界(結構感知)做粗粒度切分,再對“過大或主題跨度大的塊”應用更精細的策略(遞歸、句子、語義/主題)。查詢時配合“小-大分塊”/“父子段分塊”的檢索組裝,以小精召回、以大保上下文。

  • 使用流程

粗切(離線):按標題/段落/代碼塊/表格等結構單元切分,清理噪聲(頁眉頁腳/導航)。

細化(離線):對超長或密度不均的塊,按規則選用遞歸/句子/語義分塊二次細分。

索引(離線):同時為“小塊索引(句/子句)”與“大塊存儲(段/小節)”生成數據與metadata。

檢索(在線):小塊高精度召回 → 按父塊聚合與重排→ 在父塊中抽取命中句鄰域作為上下文,控制整體 token 預算。

  • 策略選擇規則

若塊類型為代碼/表格/公式:保持原子,不在中間切分,直接與其解釋文字打包。

若為對話:按輪次/說話人做對話式分塊,overlap 使用“輪次重疊”。

若為普通說明文/Markdown章節:

1)長度 > max_coarse或句長方差高/標點稀疏:優先語義分塊(句向量+突變閾值)。

2)否則:遞歸字符分塊(標題/段落/換行/空格/字符)保持語義邊界。

對過短塊:與同一父標題相鄰塊合并,優先向后合并。

  • 質量-成本檔位(僅供參考)

fast:僅結構→遞歸。overlap 5%–10%,不跑語義分塊和主題分塊

balanced(推薦):結構→遞歸,對異常塊啟用語義分塊,小-大檢索,overlap 10%左右

quality:在 balanced 基礎上對疑難塊啟用 Agent 精修,更強的鄰接擴展與rerank

  • 簡潔調度器示例, 將結構粗切與若干細分器組合為一個“混合分塊”入口,關鍵是類型判斷與長度閾值控制。可以把前文已實現的結構/句子/語義/對話分塊函數掛入此調度器。
from typing import List, Dict


def hybrid_chunk(
    doc_text: str,
    parse_structure,          # 函數:返回 [{'type': 'text|code|table|dialogue', 'text': str, 'breadcrumbs': [...], 'anchor': str}]
    recursive_splitter,       # 函數:text -> [{'text': str}]
    sentence_splitter,        # 函數:text -> [{'text': str}]
    semantic_splitter,        # 函數:text -> [{'text': str}]
    dialogue_splitter,        # 函數:turns(list) -> [{'text': str}],若無對話則忽略
    max_coarse_len: int = 1100,
    min_chunk_len: int = 320,
    target_len: int = 750,
    overlap_ratio: float = 0.1,
) -> List[Dict]:
    """
    返回格式: [{'text': str, 'meta': {...}}]
    """
    blocks = parse_structure(doc_text)  # 先拿到結構塊
    chunks: List[Dict] = []
    
    def emit(t: str, meta_base: Dict):
        t = t.strip()
        if not t:
            return
        # 結構重疊前綴(標題路徑)
        bc = " > ".join(meta_base.get("breadcrumbs", [])[-3:])
        prefix = f"[{bc}]\n" if bc else ""
        chunks.append({
            "text": (prefix + t) if not t.startswith(prefix) else t,
            "meta": meta_base
        })
    
    for b in blocks:
        t = b["text"]
        btype = b.get("type", "text")
        
        # 原子塊:代碼/表格
        if btype in {"code", "table", "formula"}:
            emit(t, {**b, "splitter": "atomic"})
            continue
        
        # 對話塊
        if btype == "dialogue":
            for ck in dialogue_splitter(b.get("turns", [])):
                emit(ck["text"], {**b, "splitter": "dialogue"})
            continue
        
        # 普通文本:依據長度與“可讀性”啟用不同細分器
        if len(t) <= max_coarse_len:
            # 中短文本:遞歸 or 句子
            sub = recursive_splitter(t)
            # 合并過短子塊
            buf = ""
            for s in sub:
                txt = s["text"]
                if len(buf) + len(txt) < min_chunk_len:
                    buf += txt
                else:
                    emit(buf or txt, {**b, "splitter": "recursive"})
                    buf = "" if buf else ""
            if buf:
                emit(buf, {**b, "splitter": "recursive"})
        else:
            # 超長文本:語義分塊優先
            for ck in semantic_splitter(t):
                emit(ck["text"], {**b, "splitter": "semantic"})
    
    # 輕量字符重疊(可選)
    if overlap_ratio > 0:
        overlapped = []
        for i, ch in enumerate(chunks):
            overlapped.append(ch)
            if i + 1 < len(chunks) and ch["meta"].get("breadcrumbs") == chunks[i+1]["meta"].get("breadcrumbs"):
                # 在相鄰同章節塊間引入小比例重疊
                ov = int(len(ch["text"]) * overlap_ratio)
                if ov > 0:
                    head = ch["text"][-ov:]
                    chunks[i+1]["text"] = head + chunks[i+1]["text"]
        chunks = overlapped
    
    return chunks

五、結論

圖片圖片


責任編輯:武曉燕 來源: 得物技術
相關推薦

2025-05-27 08:35:00

2025-05-28 09:00:00

2024-08-05 10:23:36

2025-09-03 04:00:45

RAG系統分塊

2025-06-11 08:40:00

LangChainRAG人工智能

2024-10-29 11:54:25

2025-05-06 10:05:23

2025-04-03 16:02:14

2024-09-24 14:32:17

RAG高級優化Fusion

2025-10-27 08:25:01

2025-11-28 02:33:00

2025-02-06 13:50:06

2024-11-08 08:39:39

2025-03-11 08:00:00

LLM開發深度學習

2024-06-27 16:38:57

2025-05-14 01:40:00

RAG數據工具

2025-03-21 14:34:17

2024-08-12 08:28:53

2024-08-06 10:02:42

2025-03-27 10:22:02

點贊
收藏

51CTO技術棧公眾號

国产日本亚洲| 99精品欧美一区二区三区| 久久久精品tv| 国产精品腿扒开做爽爽爽挤奶网站| 五月天激情综合网| 午夜欧洲一区| 国内在线观看一区二区三区| 国产无套内射久久久国产| 欧洲成人在线视频| 制服丝袜亚洲播放| 国产成人一级电影| 成人免费a**址| 岛国一区二区三区| 99精品偷自拍| 久久久蜜桃精品| 久久国产欧美日韩精品| 国产一区91| 香蕉亚洲视频| 成人免费毛片嘿嘿连载视频| 国产午夜精品福利| 亚洲一区二区中文在线| 欧美一二三区精品| 国产精品久久久亚洲| 热久久美女精品天天吊色| 国产精品视频网| 伊人色综合久久天天五月婷| 国产精品色一区二区三区| 欧美精品成人| 日本10禁啪啪无遮挡免费一区二区 | 在线观看日产精品| 亚洲午夜在线观看视频在线| 精品视频全国免费看| 欧美丰满少妇xxxbbb| 精品网站999www| 欧美在线视频免费播放| 国产女主播一区二区| 欧美一区二区三区| 日韩高清三区| 亚洲一级电影视频| 亚洲欧美日韩视频一区| 97成人在线视频| 91九色综合久久| 欧美午夜精品久久久久久蜜| 欧美三日本三级少妇三99| 日韩国产欧美亚洲| 国产婷婷视频在线| 香蕉视频官网在线观看日本一区二区| www.99.热| 91免费版在线观看| 欧美成人精品三级网站| 午夜亚洲福利在线老司机| 欧美日韩国产中文| 麻豆成人小视频| 污污的视频在线观看| 手机精品视频在线观看| 一个色综合导航| 欧美性受极品xxxx喷水| 色哟哟免费在线观看 | 欧美亚洲视频在线看网址| 六月婷婷激情综合| 成人豆花视频| 亚洲天堂av一区| 在线观看国产中文字幕| 日韩国产一级片| 99久久自偷自偷国产精品不卡| 在线看视频不卡| fc2ppv国产精品久久| 日韩精品成人在线观看| 欧美96一区二区免费视频| 亚洲国产精品麻豆| 亚洲伊人久久大香线蕉av| 成人黄色影片在线| 欧美一区二区综合| 久久免费资源| 亚洲午夜未删减在线观看| 国产东北露脸精品视频| 欧美人成网站| 精品国产一区一区二区三亚瑟| 欧美天堂一区| 日韩伦理精品| 毛片av在线| 亚洲视频在线观看免费| 少妇精品久久久久久久久久| 久久久久久久久免费视频| 亚洲中字在线| 精品中文视频在线| 91九色在线观看视频| 伊人精品影院| 亚洲男人影院| 日韩欧美www| 99热com| 国产传媒欧美日韩成人精品大片| 色又黄又爽网站www久久| 国内自拍中文字幕| 在线成人黄色| 国产精品27p| 国产一区二区在线观| 欧美日韩国产免费| 日韩精品久久久久久福利| 亚洲综合社区网| aaa大片免费观看| 国产电影精品久久禁18| 欧美精品一区二区免费| 成人动漫在线播放| 久久午夜电影| 日本乱人伦a精品| 伊人影院在线视频| 七七婷婷婷婷精品国产| 亚洲男人天堂2024| 人人九九精品| 91小视频在线观看| 日韩欧美亚洲日产国| 国产精品久久久久一区二区三区厕所 | 波多野结衣中文字幕在线| 成人爽a毛片| 亚洲国产精品久久不卡毛片| 色噜噜狠狠一区二区| 蜜桃一区二区三区四区| 日韩一区二区视频| 伊人狠狠色丁香综合尤物| 另类一区二区| 亚洲摸摸操操av| 国产喂奶挤奶一区二区三区| 亚洲欧洲三级| 国产精品免费看| 亚洲精品一区二| 成人网页在线观看| 99sesese| 欧美视频精品在线观看| 色大18成网站www在线观看| 亚洲成人精品视频在线观看| 欧美男男gaytwinkfreevideos| 久久国产精品一区二区三区四区| 极品少妇xxxx偷拍精品少妇| 午夜久久久久久久久久久| 久久99久久精品欧美| 中文字幕在线观看一区二区三区| 91免费观看国产| www.亚洲| 欧美亚洲禁片免费| 波多一区二区| 精品久久免费看| 毛片在线导航| 欧美成人免费播放| 加勒比色老久久爱综合网| a级黄色小视频| 亚洲国产另类 国产精品国产免费| 欧洲综合视频| 欧美性一级生活| 国产写真视频在线观看| 亚洲福利影片在线| 欧美日韩中出| 成人久久一区二区| 视频一区二区中文字幕| 日韩国产一级片| 亚洲v日本v欧美v久久精品| 青青草免费观看免费视频在线| 欧美一区日韩一区| 青草伊人久久| 国产伦精品一区二区| 国产成人啪午夜精品网站男同| 一级在线免费视频| 欧美无人高清视频在线观看| 国产99久久| 人妻av中文系列| 久久影院资源网| 久久一二三国产| 激情视频极品美女日韩| 9a蜜桃久久久久久免费| a在线播放不卡| 91在线品视觉盛宴免费| 57pao成人国产永久免费| 97精品超碰一区二区三区| 国产日本久久| 亚洲精品成人a8198a| 久久综合久久综合九色| 久久电影天堂| 成人免费看片'免费看| 一区二区三区视频免费| 99久久国产综合精品麻豆| 色是在线视频| 亚洲三区在线| 亚洲精品wwww| 亚洲精选在线| 免费大片在线观看| 日本一本a高清免费不卡| 国产精品国产精品国产专区不片 | 超碰在线最新| 亚洲成人xxx| 亚洲一区欧美| 欧美亚洲黄色片| 亚洲r级在线视频| 69堂精品视频在线播放| 成人国产精品色哟哟| 1024日韩| 在线看福利影| 欧洲亚洲一区| 欧美系列亚洲系列| 国产精品嫩草影院在线看| 男女无套免费视频网站动漫|