快捷導(dǎo)航

Python針對(duì)不同文本長度的處理方案總結(jié)與對(duì)比

更新時(shí)間：2025年02月23日 10:29:34 作者：大霸王龍

這篇文章主要為大家詳細(xì)介紹了Python如何針對(duì)不同文本長度的處理方案,結(jié)合層次化編碼和檢索優(yōu)化策略,感興趣的小伙伴可以參考一下

1.短文本處理(<500tokens)

from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')  # 384維小型模型

def process_short(text):
    """直接全文本編碼"""
    return model.encode(text, convert_to_tensor=True)

# 示例
short_text = "自然語言處理的基礎(chǔ)概念"  # 長度約15 tokens
vector = process_short(short_text)

2. 中長文本處理 (500-2000 tokens)

from langchain_text_splitters import RecursiveCharacterTextSplitter

def process_medium(text):
    """重疊分塊策略"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", "。", "！", "？"]
    )
    chunks = splitter.split_text(text)
    return [model.encode(chunk) for chunk in chunks]

# 示例
medium_text = "機(jī)器學(xué)習(xí)發(fā)展歷史...（約1500字）"  # 約1800 tokens
chunk_vectors = process_medium(medium_text)

3. 長文本處理 (2000-20000 tokens)

import spacy

def process_long(text):
    """語義分塊+摘要增強(qiáng)"""
    # 加載語義分割模型
    nlp = spacy.load("zh_core_web_sm") 
    doc = nlp(text)
    
    # 按段落分割
    chunks = [sent.text for sent in doc.sents]
    
    # 生成章節(jié)摘要
    summary_model = SentenceTransformer('uer/sbert-base-chinese-nli')
    summaries = [summary_model.encode(chunk[:200]) for chunk in chunks]
    
    return chunks, summaries

# 示例
long_text = "人工智能技術(shù)白皮書...（約2萬字）"  # 約20000 tokens
text_chunks, summary_vecs = process_long(long_text)

4. 超長文本處理 (20000-200000 tokens)

import faiss
import numpy as np

class HierarchicalIndex:
    def __init__(self):
        # 兩級(jí)索引結(jié)構(gòu)
        self.summary_index = faiss.IndexFlatL2(384)
        self.chunk_index = faiss.IndexIVFPQ(
            faiss.IndexFlatL2(384), 384, 100, 16, 8
        )
        self.metadata = []

    def add_document(self, text):
        # 生成段落級(jí)摘要
        chunks, summaries = process_long(text)
        
        # 構(gòu)建索引
        summary_vecs = np.array(summaries).astype('float32')
        chunk_vecs = np.array([model.encode(c) for c in chunks]).astype('float32')
        
        self.summary_index.add(summary_vecs)
        self.chunk_index.add(chunk_vecs)
        self.metadata.extend(chunks)

    def search(self, query, k=5):
        # 先檢索摘要層
        query_vec = model.encode(query).astype('float32')
        _, sum_indices = self.summary_index.search(np.array([query_vec]), 10)
        
        # 精搜相關(guān)塊
        target_chunks = [self.chunk_index.reconstruct(i) for i in sum_indices]
        target_chunks = np.array(target_chunks).astype('float32')
        _, chunk_indices = self.chunk_index.search(target_chunks, k)
        
        return [self.metadata[i] for i in chunk_indices]

# 使用示例
hindex = HierarchicalIndex()
hindex.add_document("某領(lǐng)域技術(shù)文檔...（約15萬字）")  # 約200000 tokens
results = hindex.search("深度學(xué)習(xí)在醫(yī)療影像的應(yīng)用")

5. 海量文本處理 (>200000 tokens)

import dask.dataframe as dd
from dask.distributed import Client

def process_extreme(file_path):
    """分布式處理方案"""
    client = Client(n_workers=4)  # 啟動(dòng)Dask集群
    
    # 分塊讀取
    df = dd.read_parquet(file_path, chunksize=100000)  
    
    # 并行編碼
    df['vector'] = df['text'].map_partitions(
        lambda s: s.apply(model.encode),
        meta=('vector', object)
    )
    
    # 構(gòu)建分布式索引
    df.to_parquet("encoded_data.parquet", engine="pyarrow")
    
# 示例（處理100萬條文本）
process_extreme("massive_data.parquet")

性能優(yōu)化對(duì)照表=

文本長度	處理策略	索引類型	響應(yīng)時(shí)間	內(nèi)存消耗
<500	直接編碼	FlatIndex	<10ms	1MB
2000	重疊分塊	IVF+PQ	50-100ms	50MB
20000	語義分塊+摘要索引	二級(jí)索引	200-500ms	300MB
200000	層次化索引	IVFOPQ+ProductQuant	1-2s	2GB
>200000	分布式處理	分片索引	10s+	集群資源