以下參考課程 LLM Twin: Building Your Production-Ready AI Replica 撰寫

在過去兩天的大量學習和實踐中，我們深入探討了如何通過結構化的方法制定資料前處理步驟。隨後，我們整理了這些核心步驟——清理、分塊和嵌入，以及作者在實踐這些步驟背後的設計思路。

1. 資料清理：保持數據的一致性

在資料流處理中，資料可能來自不同的來源，往往包含多種格式、不規則字符和雜訊。資料清理是每個資料管道的第一個步驟，旨在統一資料格式、去除不需要的內容，並準備好資料進行進一步處理。

清理的步驟

資料清理通常包括以下幾個步驟：

• 移除不需要的格式：如粗體、斜體等。
• 清理多餘的空格和標點：保證資料乾淨且易於處理。
• 替換網址和特殊符號：例如將網址替換為佔位符。

def clean_text(text_content):
    cleaned_text = unbold_text(text_content)
    cleaned_text = unitalic_text(cleaned_text)
    cleaned_text = remove_emojis_and_symbols(cleaned_text)
    cleaned_text = clean(cleaned_text)
    cleaned_text = replace_unicode_quotes(cleaned_text)
    cleaned_text = clean_non_ascii_chars(cleaned_text)
    cleaned_text = replace_urls_with_placeholder(cleaned_text)
    return cleaned_text

這段程式碼中，每個函數負責一項具體的清理任務，如移除文字格式、清理符號、替換特殊字符等。這樣的分步清理可以有效地準備資料，使其符合後續處理和機器學習模型的要求。

2. 分塊：處理大文本的方法

當處理大段文本時，這些資料往往過於龐大，無法直接輸入嵌入模型。因此，我們需要將大文本分成小塊，這樣不僅提高嵌入處理的效率，也可以避免過度消耗內存。

分塊邏輯

我們採用了一個雙層分割策略：

• 字符級分割：首先，將文本按段落進行初步分割。
• 基於模型的分割：接下來，使用變換器模型進行更精細的分割，確保每個分塊的長度符合模型的輸入要求。

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    SentenceTransformersTokenTextSplitter,
)

def chunk_text(text):
    # 按段落進行初步分割
    character_splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n"],
        chunk_size=500,
        chunk_overlap=0
    )
    text_split = character_splitter.split_text(text)

    # 基於變換器進行細粒度分割
    token_splitter = SentenceTransformersTokenTextSplitter(
        chunk_overlap=50,
        tokens_per_chunk=settings.EMBEDDING_MODEL_MAX_INPUT_LENGTH,
        model_name=settings.EMBEDDING_MODEL_ID,
    )

    chunks = []
    for section in text_split:
        chunks.extend(token_splitter.split_text(section))

    return chunks

• RecursiveCharacterTextSplitter 會根據段落來初步分割文本。
• SentenceTransformersTokenTextSplitter 則進行更精細的分割，確保每個分塊的長度不會超過嵌入模型的輸入限制。

這種雙層策略能夠平衡文本分割的粒度，使得每個分塊既不過大也不過小，並為後續的嵌入處理提供最佳輸入長度。

3. 嵌入：將文本轉換為向量

文本嵌入是將清理和分塊後的文本轉換為數字向量的過程，這些向量可以用於機器學習模型的訓練、分類和檢索等工作。選擇合適的嵌入模型至關重要，因為不同的文本類型需要不同的模型來進行最佳處理。

嵌入模型選擇

1. 一般文本嵌入：對於文章和貼文的文本，我們使用 SentenceTransformer，這是一個輕量級嵌入模型，適合於即時處理。

from sentence_transformers import SentenceTransformer

def embedd_text(text):
    model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
    return model.encode(text)

2. 程式碼嵌入：程式碼的結構和語境通常與一般文本不同，因此我們使用 INSTRUCTOR 模型來處理程式碼的嵌入。

from InstructorEmbedding import INSTRUCTOR

def embedd_repositories(text):
    model = INSTRUCTOR("hkunlp/instructor-xl")
    instruction = "Represent the structure of the repository"
    return model.encode([instruction, text])

4. 整合資料前處理到資料流管道

在資料流處理過程中，這些前述的資料前處理步驟（清理、分塊、嵌入）需要整合到資料流管道中。結合之前的文章中介紹過的分派器，我們將清理和嵌入過程具體應用到 Bytewax 資料流中。

stream = op.map("clean dispatch", stream, CleaningDispatcher.dispatch_cleaner)
stream = op.flat_map("chunk dispatch", stream, ChunkingDispatcher.dispatch_chunker)
stream = op.map("embedded chunk dispatch", stream, EmbeddingDispatcher.dispatch_embedder)

• CleaningDispatcher.dispatch_cleaner：處理資料的清理步驟。
• ChunkingDispatcher.dispatch_chunker：將清理後的資料進行分塊處理。
• EmbeddingDispatcher.dispatch_embedder：將分塊後的文本嵌入為向量。

這樣的資料流管道每一步都能確保資料的連續性和一致性，最終輸出為適合機器學習模型使用的向量資料。

ref.

• SOTA Python Streaming Pipelines for Fine-tuning LLMs and RAG — in Real-Time!