為什么 Chunking 決定了 LLM 的性能？窗口、檢索與成本全解析 精華
一個實用的文本分割指南，包含代碼、圖表，以及對Chonkie、LangChain和LlamaIndex的輕量介紹
上下文窗口變大了。有些模型一次能處理整章內容。這看似自由，但并未消除權衡。分塊依然決定模型讀什么，檢索返回什么，以及每次調用你得花多少錢。
分塊說起來簡單，做起來容易出錯。你需要把長文本切成模型或嵌入器能處理的片段。聽起來像是在調整大小，但實際上是關于相關性。好的分塊要小到足夠具體，大到能獨立存在。做到這一點，檢索就像記憶一樣自然。做不到，你會得到模糊的匹配、半吊子答案，甚至模型開始瞎猜。
這是一篇實用指南。我們用Chonkie舉例，因為它封裝了工程師們實際使用的那些不那么光鮮的部分。在合適的地方，我們會提到LangChain和LlamaIndex的分塊器，以及像late chunking這樣的長上下文策略——先嵌入再切分，以保留每個向量中的全局線索。你會看到這種方法在哪些場景有用，哪些場景沒用，還有它會花多少成本。
長窗口改變的是策略，不是原則。即使有百萬token的窗口，你也不會想把所有東西都塞進prompt。你想要的是幾個精準的片段，因為它們包含答案。分塊就是讓你先把這些片段創建出來。
窗口、檢索和成本
分塊處于三個核心問題的交匯處。模型只能讀取有限的token。你的檢索器必須拉到正確的段落。你發送的每個token都要花時間和金錢。分塊是平衡這三者的控制旋鈕。
上下文窗口設定了硬性上限。你永遠不會把整篇文檔直接發送。你會發送一個prompt包裝、一個問題和幾個片段。這些片段只有在你提前分好塊的情況下才會存在。這就是分塊的作用。它把長文本切成適合的片段，并塑造每個片段，讓模型單獨讀取時也能理解。
檢索關注的是焦點。如果一個分塊包含三個不相關的想法，向量會把它們混淆。查詢可能匹配到錯誤的部分，帶來一堆無關內容。如果分塊太小，就像缺了線索的謎語。最好的分塊就像教科書里的短段落：一個核心想法，足夠細節來支持主題，不帶跑題的廢話。
成本通過重疊和top-k潛入。重疊保護了邊界處的上下文，但會增加token數量。取五個分塊而不是三個，模型要讀更多內容，你得為這些token付費，還可能稀釋相關性。正確的設置不是口號，而是經過權衡的選擇。
一個簡單的分塊方法是從窗口倒推。計算你的system prompt、guardrails、指令和安全緩沖區的token數。剩下的就是檢索文本的工作預算。將這個預算分配給計劃展示的分塊數量。如果是一次性流式處理，留出空間給模型的回答。
這里有個小工具幫你老實計算token。它用了tiktoken，匹配很多OpenAI兼容模型的計數方式。如果用其他技術棧，可以換成Hugging Face的tokenizer。
# python 3.11
# pip install tiktoken
import tiktoken
from math import ceil
# 設置tokenizer匹配你的模型。常用選項：
# "cl100k_base" (廣泛使用)
# "o200k_base" (較新的200k token系列)
ENCODING_NAME = "cl100k_base"
enc = tiktoken.get_encoding(ENCODING_NAME)
deftokens(text: str) -> int:
returnlen(enc.encode(text))
defchunk_budget(context_max: int, prompt_text: str, buffer_tokens: int = 256) -> int:
used = tokens(prompt_text) + buffer_tokens
returnmax(0, context_max - used)
defchunks_needed(doc_tokens: int, chunk_size: int, overlap: int) -> int:
step = max(1, chunk_size - overlap)
return ceil(max(0, doc_tokens - overlap) / step)
defoverlap_overhead(doc_tokens: int, chunk_size: int, overlap: int) -> float:
n = chunks_needed(doc_tokens, chunk_size, overlap)
return (n * chunk_size) / max(1, doc_tokens)如果你的文檔有5萬token，窗口留給檢索文本3千token，想用三個分塊，每個分塊大約一千token。如果你設置15%的重疊，有效讀取量會增加。上面這個工具會告訴你開銷比例，幫你看清權衡。
字符和token不是一個單位。字符分割器速度快，適合粗略切割，但會錯過精確的token限制。token分割器匹配模型對文本的看法，讓你在窗口內榨取最大價值。你可以用字符做原型，到了生產預算時再換成token。Chonkie、LangChain和LlamaIndex都提供這兩種方式。實驗結束后，用token-aware的路徑。
檢索質量在分塊連貫時提升。你可以用簡單測試驗證。挑一組有已知答案的問題。建兩個不同分塊大小和重疊的索引。對每個問題，取出top-3分塊，檢查答案文本是否出現在檢索結果中。跟蹤recall@3，記錄拉進prompt的額外token數。當recall上升而token下降，你找到了更好的設置。當recall上升但token暴增，判斷收益是否值得成本。你不需要大數據集，十幾個來自你領域的真實問題就能揭示很多。
窗口大小不盡相同。嵌入器的限制可能比你的聊天模型小。如果嵌入器只能接受短文本，你得先分塊再嵌入。如果用late chunking，順序相反。你先把整個文檔通過長上下文編碼器，得到token級向量，再聚合成分塊向量。每個分塊向量現在都帶有全文的線索。對于證據跨越局部邊界的查詢，檢索效果會更好。代價是計算和內存。你需要一個能處理全文的模型，還得在回答問題前先跑一遍。
Contextual retrieval是個輕量變種。你還是先分塊，但在嵌入或存儲元數據時，給每個分塊附上小結、標題或線索錨點。比如，一個關于Berlin的分塊會帶上“Berlin, population context, European cities”的元數據。查詢時，這能幫向量空間或重排器區分相似項。它不會修復不連貫的分塊，但能讓好的分塊更常勝出。
窗口變大時，容易想塞更多分塊。別沖動。模型按順序讀取，上下文前部會得到更多關注。干凈的top-3通常比雜亂的top-8強。如果需要更廣的覆蓋，用兩步計劃。先取更寬的集合，用cross-encoder重排，再把top幾項傳給聊天模型。Chonkie通過生成更好的原始片段融入這個計劃。LangChain、LlamaIndex或你自己的代碼可以處理重排。
這里有個緊湊的滑動窗口代碼，尊重token計數，返回起始偏移量，方便后續映射回源位置。
from typing importList, Tuple
defsliding_chunks(text: str, max_tokens: int, overlap_tokens: int) -> List[Tuple[int, int, str]]:
ids = enc.encode(text)
n = len(ids)
step = max(1, max_tokens - overlap_tokens)
out = []
i = 0
while i < n:
j = min(n, i + max_tokens)
piece_ids = ids[i:j]
out.append((i, j, enc.decode(piece_ids)))
if j == n:
break
i += step
return out真正管用的分塊策略
沒有一種分割器適合所有語料庫。最佳選擇取決于你存儲什么、查詢什么，以及模型怎么計數文本。從最簡單可行的開始，只有測試告訴你必須升級時才往上走。
固定大小分塊帶重疊
這是主力選手。你按token切到目標大小，保留小部分重疊，確保邊界事實不被孤立。它快、可靠、易于預算，但也鈍。它不知道句子或段落的邊界。如果你的文檔格式統一或生成式，鈍點也沒啥。
# python 3.11
# pip install tiktoken
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
defto_tokens(s: str) -> list[int]:
return enc.encode(s)
deffrom_tokens(ids: list[int]) -> str:
return enc.decode(ids)
deffixed_chunks(text: str, max_tokens: int, overlap: int) -> list[str]:
ids = to_tokens(text)
step = max(1, max_tokens - overlap)
out = []
i = 0
while i < len(ids):
j = min(len(ids), i + max_tokens)
out.append(from_tokens(ids[i:j]))
if j == len(ids):
break
i += step
return out當你需要嚴格控制token預算時用這個。保持重疊小。如果檢索里反復出現相同句子，說明重疊設太高了。
自然邊界分塊
讀者按句子和段落思考。模型在分塊包含完整想法時表現更好。自然邊界分割器會把完整句子打包，直到下一個句子會超token預算。它保持連貫性，又不失預算控制。
import re
sent_re = re.compile(r'(?<!\w\.\w.)(?<![A-Z][a-z]\.)(?<=[.!?])\s+')
defsentence_pack(text: str, max_tokens: int) -> list[str]:
sentences = sent_re.split(text.strip())
out, cur = [], []
cur_len = 0
for s in sentences:
s_tokens = len(to_tokens(s))
if s_tokens > max_tokens:
out.extend(fixed_chunks(s, max_tokens, overlap=0))
continue
if cur_len + s_tokens <= max_tokens:
cur.append(s)
cur_len += s_tokens
else:
out.append(" ".join(cur))
cur, cur_len = [s], s_tokens
if cur:
out.append(" ".join(cur))
return out用在散文、文檔和報告上。它避免句子被切斷，給嵌入器更清晰的信號。如果段落短，就把段落當單位，打包思路不變。
遞歸分塊
真實文檔有結構。你可以不用LLM就尊重它。遞歸分割器先嘗試大分界，再回退到小分界，最后才用固定窗口。標題、雙換行、句子、token，這個順序盡量保持相關上下文。
import re
defrecursive_chunks(text: str, max_tokens: int, overlap: int = 0) -> list[str]:
iflen(to_tokens(text)) <= max_tokens:
return [text]
if"\n#"in text:
parts = re.split(r'\n(?=#)', text)
elif"\n\n"in text:
parts = text.split("\n\n")
else:
parts = sentence_pack(text, max_tokens)
out, buf = [], ""
for p in parts:
candidate = buf + ("\n\n"if buf else"") + p
iflen(to_tokens(candidate)) <= max_tokens:
buf = candidate
else:
if buf:
out.append(buf)
iflen(to_tokens(p)) <= max_tokens:
buf = p
else:
out.extend(fixed_chunks(p, max_tokens, overlap))
buf = ""
if buf:
out.append(buf)
return out if out else fixed_chunks(text, max_tokens, overlap)這和Chonkie的RecursiveChunker或LangChain的遞歸分割器實際效果一致。適合混合格式文檔，是個強默認選擇。
語義分塊
意義可以引導分割點。你嵌入候選單位（如句子），計算相鄰單位的相似度，當相似度下降或token預算超限時開始新分塊。結果是一組討論同一主題的句子，這正是你想要的分塊。
# 你提供一個返回numpy數組的embed()函數
import numpy as np
defsemantic_chunks(text: str, max_tokens: int, sim_threshold: float = 0.62) -> list[str]:
sentences = sent_re.split(text.strip())
vecs = [embed(s) for s in sentences] # shape (n, d)
out, cur, cur_len = [], [], 0
defcos(a, b):
na = a / (np.linalg.norm(a) + 1e-9)
nb = b / (np.linalg.norm(b) + 1e-9)
returnfloat((na * nb).sum())
for i, s inenumerate(sentences):
v = vecs[i]
next_v = vecs[i + 1] if i + 1 < len(vecs) elseNone
s_tokens = len(to_tokens(s))
start_new = False
if cur and next_v isnotNoneand cos(v, next_v) < sim_threshold:
start_new = True
elif cur_len + s_tokens > max_tokens:
start_new = True
if start_new and cur:
out.append(" ".join(cur))
cur, cur_len = [], 0
cur.append(s)
cur_len += s_tokens
if cur:
out.append(" ".join(cur))
return out當長段落內主題切換或標題不可靠時，用語義分塊。索引時需要一次嵌入，成本高，但檢索通常會以更干凈的結果回報。
Late Chunking
有時候你希望每個分塊向量記住文檔的其余部分。Late chunking反轉順序。你先用長上下文編碼器跑一遍全文，拿到token級向量，再聚合成分塊向量。每個分塊向量都帶有全局線索，避免單獨嵌入切片時丟失信息。
Late chunking適合答案從分散線索中抽取的場景。它對硬件要求高。你需要一個能吃下整個文件的模型和足夠的內存來處理這一輪。很多團隊把這個留給高價值語料庫，其他地方用經典的先分塊后嵌入。
LLM引導的分塊
你也可以讓小模型標記邊界。給它文本和預算，要求返回保持想法完整的偏移量。用你的tokenizer驗證后再信任。
{
"instruction": "將文本分成不超過800 token的連貫段落。返回JSON格式的start:end字符偏移量。",
"text": "…你的文檔在這兒…"
}這種方法對雜亂的散文或會議記錄很精確。但它慢且每次調用都有成本，所以大多團隊用它處理關鍵文檔或作為一次性預處理器。
代碼感知分塊
源代碼需要不同的刀。函數、類和docstring是自然單位。像Tree-sitter這樣的樹解析器可以遍歷文件，給你符號對應的范圍。然后你把這些范圍打包到token預算里。盡量別把函數體切開。模型在完整單位存在時回答代碼問題更好。
Chonkie和朋友們的定位
Chonkie用直白的命名暴露這些概念：TokenChunker用于固定路徑，SentenceChunker用于自然邊界，RecursiveChunker用于結構化路徑，SemanticChunker用于語義分割，LateChunker用于先嵌入后分割，還有實驗性的neural chunkers供你想要學習式選擇時使用。LangChain和LlamaIndex提供類似的分塊器。
這里有個TypeScript版本，方便JavaScript團隊復制token-aware模式，不用猜模型怎么計數文本。
// TypeScript: 帶重疊的token-aware滑動分塊
// npm i tiktoken
import { get_encoding } from"tiktoken";
const enc = get_encoding("cl100k_base"); // 或 "o200k_base" 匹配你的模型
typeChunk = { startTok: number; endTok: number; text: string };
exportfunctionslidingChunks(text: string, maxTokens: number, overlapTokens: number): Chunk[] {
const ids = enc.encode(text);
const step = Math.max(1, maxTokens - overlapTokens);
constout: Chunk[] = [];
for (let i = 0; i < ids.length; i += step) {
const j = Math.min(ids.length, i + maxTokens);
out.push({ startTok: i, endTok: j, text: enc.decode(ids.slice(i, j)) });
if (j === ids.length) break;
}
return out;
}標簽可能變，權衡不變。
選一個分塊方式，測recall@固定k，記錄每次查詢拉取的token。如果recall上升而token穩定，你選對了。如果recall上升但token激增，判斷收益是否值得延遲和賬單。
從原始文本到索引分塊
好的pipeline把雜亂輸入變成干凈、可檢索的片段。Chonkie的CHOMP理念和大多數團隊的做法無縫對接。先規范化文本，再分割，再潤色和豐富，最后存儲。不同庫，節奏相同。
名字聽起來好玩，但工作很標準。Document是你源文本。Chef是預處理，修空格、壞OCR或怪Unicode。Chunker是你選的分割器。Refinery是后處理，合并零散片段、標記元數據、附加嵌入。Friends是分塊的去處，要么是向量存儲，要么是文件導出。如果你喜歡中性標簽，換成預處理、分割、精煉、輸出，一一對應。
預處理是小勝累積的地方。從爬取的PDF里去掉重復頁眉頁腳。規范化引號和破折號。壓縮重復空格。目標是穩定的分割和嵌入。兩個略有不同的相同段落在向量空間里不會很好碰撞。
分割是重頭戲。把參數集中一處，每次運行都記錄。以后換模型或預算時，你會想知道哪個索引用了哪些設置。在索引旁留個簡短清單，記下tokenizer名稱、分塊大小和重疊。
精煉是質量控制。合并無法獨立的小尾巴分塊。附上源元數據，如文件名、章節標題、起止字符偏移和token范圍。如果打算搜索元數據，再加一行簡短摘要。這里也做嵌入，因為你想一次搞定然后繼續。
輸出故意簡單。把向量和元數據推到存儲里?；蛘邔慗SONL，在別處索引。不管怎樣，記錄索引版本和生成它的嵌入模型。檢索出問題時，沒法追溯存儲內容會很麻煩。
這里有個緊湊的Python示例，模仿Chonkie風格的pipeline感覺。API形狀有代表性，換成你用的真實接口。
# python 3.11
# 模仿典型的chonkie/lc風格流程；適配你的具體庫
from dataclasses import dataclass
from typing import Iterable, Dict, Any, List
import hashlib, time
import tiktoken # token-aware
@dataclass
classChunk:
text: str
doc_id: str
chunk_id: str
start_char: int
end_char: int
start_tok: int
end_tok: int
meta: Dict[str, Any]
vector: list[float] | None = None
defnormalize_text(raw: str) -> str:
cleaned = " ".join(raw.replace("\u00A0", " ").split())
return cleaned
defchunk_sliding( text: str,
tokenizer,
chunk_size: int,
overlap: int,
doc_id: str = "doc-001") -> List[Chunk]:
"""帶重疊的token窗口分割；返回token/字符范圍以映射回源"""
ids = tokenizer.encode(text)
out: List[Chunk] = []
i = 0
step = max(1, chunk_size - overlap)
while i < len(ids):
j = min(len(ids), i + chunk_size)
piece_ids = ids[i:j]
piece = tokenizer.decode(piece_ids)
start_char = len(tokenizer.decode(ids[:i]))
end_char = start_char + len(piece)
cid = hashlib.md5(f"{doc_id}:{i}:{j}".encode()).hexdigest()[:10]
out.append(Chunk(
text=piece,
doc_id=doc_id,
chunk_id=f"{doc_id}-{cid}",
start_char=start_char,
end_char=end_char,
start_tok=i,
end_tok=j,
meta={"ver": 1}
))
if j == len(ids):
break
i += step
return out
defembed_all(chunks: Iterable[Chunk], embedder) -> List[Chunk]:
texts = [c.text for c in chunks]
vecs = embedder.embed_documents(texts) # list[list[float]]
out: List[Chunk] = []
for c, v inzip(chunks, vecs):
c.vector = v
out.append(c)
return out
defupsert(chunks: Iterable[Chunk], vectordb):
payload = [{
"id": c.chunk_id,
"vector": c.vector,
"metadata": {
"doc_id": c.doc_id,
"start_char": c.start_char,
"end_char": c.end_char,
"start_tok": c.start_tok,
"end_tok": c.end_tok,
**c.meta
},
"text": c.text
} for c in chunks]
vectordb.upsert(payload)
# 連接起來（token-aware）
classTikTokenizer:
def__init__(self, name: str = "cl100k_base"):
self.enc = tiktoken.get_encoding(name)
defencode(self, s: str):
returnself.enc.encode(s)
defdecode(self, ids):
returnself.enc.decode(ids)
classDummyEmbedder:
defembed_documents(self, texts):
return [[hash(t) % 997 / 997.0for _ inrange(8)] for t in texts]
classDummyDB:
defupsert(self, rows):
print(f"已插入 {len(rows)} 個分塊")
tok = TikTokenizer("cl100k_base") # 如果你的模型用它，換成 "o200k_base"
embedder = DummyEmbedder()
db = DummyDB()
raw = open("document.txt").read()
clean = normalize_text(raw)
chunks = chunk_sliding(clean, tokenizer=tok, chunk_size=1200, overlap=180, doc_id="doc-001")
vectored = embed_all(chunks, embedder)
upsert(vectored, db)
print(f"在 {time.strftime('%Y-%m-%d %H:%M:%S')} 索引了 {len(vectored)} 個分塊")索引衛生值得注意。重疊有幫助，但會在接縫處制造重復。檢索后常看到相鄰的兩個命中共享一句。構建prompt前按字符范圍去重。保留更長的或得分更高的。這里有個跨存儲工作的小工具。
def dedupe_overlaps(hits):
# 每個hit: {"doc_id","start_char","end_char","text", "score"?}
hits = sorted(hits, key=lambda h: (-(h.get("score", 0.0)), h["doc_id"], h["start_char"]))
kept = []
seen = {}
for h in hits:
key = h["doc_id"]
cur = seen.get(key, [])
overlaps = [k for k in cur ifnot (h["end_char"] <= k["start_char"] or h["start_char"] >= k["end_char"])]
if overlaps:
best = max([*overlaps, h], key=lambda x: (x["end_char"] - x["start_char"], x.get("score", 0.0)))
if best is h:
for k in overlaps:
cur.remove(k)
cur.append(h)
else:
cur.append(h)
seen[key] = cur
for v in seen.values():
kept.extend(v)
kept.sort(key=lambda h: (h["doc_id"], h["start_char"]))
return kept在索引旁留個簡短清單。記錄tokenizer、嵌入模型、分塊大小、重疊和日期。生產中感覺不對時，你會有個清晰的發貨記錄。
付諸實踐：問答、摘要和搜索
一個pipeline只有在回答問題、把長報告壓成可行動的內容，或比人更快找到段落時才證明自己。以下模式匹配許多團隊現在的發貨方式。部件相同，旋鈕相同，只有預算和庫變。
基于事實的問答
問答是戴上耳機的檢索。你索引一次，取幾個聚焦分塊，把這些分塊和簡單指令、明確請求交給模型。你不會把整本書塞進上下文，你喂一小盤。
這里有個緊湊的Python檢索步驟，假設你已有索引。它取top-k，去掉重疊回聲，構建基于事實的prompt。LLM調用是個占位符，接入你用的客戶端。
from typing importList, Dict, Any
defbuild_prompt(question: str, hits: List[Dict[str, Any]]) -> str:
parts = []
for i, h inenumerate(hits, 1):
parts.append(f"[{i}] {h['text']}\n(來源: {h['doc_id']}:{h['start_char']}-{h['end_char']})")
context = "\n\n".join(parts)
return (
"僅用以下片段回答。引用來源如[1]、[2]。如果答案不在其中，說你不知道。\n\n"
f"{context}\n\n問題: {question}\n回答:"
)
defanswer(question: str, vectordb, k: int = 3, budget_tokens: int = 2200) -> str:
raw_hits = vectordb.similarity_search(question, top_k=8) # 每個包含text, doc_id, start_char, end_char, score
hits = dedupe_overlaps(raw_hits)
kept, used = [], 0
for h in hits:
t = len(enc.encode(h["text"]))
if used + t > budget_tokens:
continue
kept.append(h)
used += t
iflen(kept) == k:
break
prompt = build_prompt(question, kept)
# llm_response = llm.complete(prompt, max_tokens=400)
# return llm_response.text
return prompt # 演示用保持k小。給檢索文本設動態token預算。攜帶偏移量，答案能引用來源，你也能在閱讀器里高亮。如果語料庫噪點多，加第二輪用cross-encoder重排前十個命中，再挑最后三個。
可擴展到章節的摘要
長文檔先累死人，再累死模型。好的摘要讓你掃一眼、做決定、按需深挖。你可以用map-reduce，或邊走邊精煉。兩者都行。精煉適合線性報告，map-reduce適合章節分明的龐大文檔。
這是一個小精煉循環。它把運行摘要帶到下一步，讓模型始終看到全局和下一個分塊。
def refine_summary(chunks: list[str], llm, part_tokens: int = 800) -> str:
summary = "開始簡潔的執行摘要。捕捉關鍵點、數字和決定。"
for i, ch inenumerate(chunks, 1):
ch_ids = enc.encode(ch)[:part_tokens]
ch = enc.decode(ch_ids)
prompt = (
f"當前摘要:\n{summary}\n\n"
f"新一節 ({i}/{len(chunks)}):\n{ch}\n\n"
"更新摘要。保持簡潔。保留數字和名字。如新信息修正早期內容，改正它。"
)
# resp = llm.complete(prompt, max_tokens=300)
# summary = resp.text.strip()
summary = f"[占位符更新第{i}節] {summary[:120]}"
return summary你不必追求一次完美覆蓋。你需要一個掃一眼可信的摘要。如果文檔含大量表格或代碼，分塊時把這些塊當原子單位，模型在重寫前能看到完整單位。
像記憶一樣的搜索
簡單的語義搜索是種安慰。你輸入想找的，得到大概想要的段落。好的分塊讓它感覺干脆，而不是糊成一團。字符適合初建，token在乎預算時更好。
這里有個小型端到端流程，包含常見模塊。它分塊、嵌入、索引，然后跑查詢。換上你的真實tokenizer和嵌入器，換上你的存儲。
import tiktoken
# 本地tokenizer讓代碼塊自包含
enc = tiktoken.get_encoding("cl100k_base")
classMiniIndex:
def__init__(self, embedder):
self.embedder = embedder
self.rows = [] # [{"id","vector","text","meta"}]
defadd(self, rows):
self.rows.extend(rows)
defsimilarity_search(self, q: str, top_k: int = 3):
import numpy as np
qv = np.array(self.embedder.embed_query(q))
scored = []
for r inself.rows:
v = np.array(r["vector"])
score = float(qv @ v / (np.linalg.norm(qv) * np.linalg.norm(v) + 1e-9))
scored.append((score, r))
scored.sort(key=lambda x: x[0], reverse=True)
return [r for _, r in scored[:top_k]]
# 使用文章前面定義的sliding_chunks(text, chunk_size, overlap)
defindex_corpus(docs, embedder, chunk_size=800, overlap=100):
idx = MiniIndex(embedder)
for d in docs:
chunks = sliding_chunks(d["text"], chunk_size, overlap)
rows = []
for (si, ei, ch_text) in chunks:
vec = embedder.embed_documents([ch_text])[0]
rows.append({
"id": f"{d['id']}-{si}:{ei}",
"vector": vec,
"text": ch_text,
"meta": {"doc_id": d["id"], "start_tok": si, "end_tok": ei}
})
idx.add(rows)
return idx
classDummyEmbedder:
defembed_documents(self, texts):
return [[(hash(t) % 997) / 997.0for _ inrange(384)] for t in texts]
defembed_query(self, text):
return [(hash(text) % 997) / 997.0for _ inrange(384)]
docs = [{"id": "manual", "text": open("manual.txt").read()}]
embedder = DummyEmbedder()
idx = index_corpus(docs, embedder)
hits = idx.similarity_search("factory reset steps", top_k=3)
for h in hits:
print(h["meta"]["doc_id"], h["text"][:120].replace("\n", " "), "…")生產中你不會打印。你會展示一個整潔的結果卡，帶標題、短摘錄和鏈接，滾動到閱讀器里的精確位置。這個緊湊循環讓系統感覺活了。分塊器默默完成了關鍵工作。
何時升級到Late Chunking或重排
如果你的問題常從遠處線索中抽取，late chunking能幫上忙，因為每個分塊向量生來就帶有全局視野。如果top-k返回近似命中，cross-encoder重排器能救回正確順序。兩者都加時間和復雜性，所以用在答案必須更準而不是更快的地方。其他情況保留簡單路徑。
閉環：一個你真會用的簡單測試計劃
空談無用，測量為王。目標不是排行榜，是確信你的分塊設置能帶來好recall，不膨脹上下文。
你需要三樣東西：一小套有已知答案的真實問題；一個可重復的方式，用特定分塊設置建索引；一個循環，逐個提問，取top k，檢查正確答案是否出現，記錄會發送給模型的token數。
這里有個緊湊的測試框架。它對一組分塊大小和重疊網格測試，用小真實集報告recall@3和平均發送的上下文token數。接入真實嵌入器和向量存儲時，替換相應部分。
# python 3.11
# pip install tiktoken
import tiktoken, time, itertools, numpy as np
from typing importList, Dict, Any, Tuple
enc = tiktoken.get_encoding("cl100k_base")
classDummyEmbedder:
defembed_documents(self, texts: List[str]) -> List[List[float]]:
return [[(hash(t) % 997) / 997.0for _ inrange(384)] for t in texts]
defembed_query(self, text: str) -> List[float]:
return [(hash(text) % 997) / 997.0for _ inrange(384)]
classMiniIndex:
def__init__(self, embedder):
self.embedder = embedder
self.rows: List[Dict[str, Any]] = []
defadd(self, rows: List[Dict[str, Any]]):
self.rows.extend(rows)
defsimilarity_search(self, q: str, top_k: int = 10) -> List[Dict[str, Any]]:
qv = np.array(self.embedder.embed_query(q))
scored = []
for r inself.rows:
v = np.array(r["vector"])
score = float(qv @ v / (np.linalg.norm(qv) * np.linalg.norm(v) + 1e-9))
scored.append((score, r))
scored.sort(key=lambda x: x[0], reverse=True)
return [r for _, r in scored[:top_k]]
deftokens(s: str) -> int:
returnlen(enc.encode(s))
defsliding_chunks(text: str, max_tokens: int, overlap_tokens: int) -> List[Tuple[int, int, str]]:
ids = enc.encode(text)
n = len(ids)
step = max(1, max_tokens - overlap_tokens)
out, i = [], 0
while i < n:
j = min(n, i + max_tokens)
piece_ids = ids[i:j]
out.append((i, j, enc.decode(piece_ids)))
if j == n:
break
i += step
return out
defindex_with_settings(doc_id: str, text: str, embedder, chunk_size: int, overlap: int) -> MiniIndex:
idx = MiniIndex(embedder)
rows = []
for si, ei, ch_text in sliding_chunks(text, chunk_size, overlap):
vec = embedder.embed_documents([ch_text])[0]
rows.append({"id": f"{doc_id}-{si}:{ei}", "vector": vec, "text": ch_text,
"meta": {"doc_id": doc_id, "start_tok": si, "end_tok": ei, "tok_len": tokens(ch_text)}})
idx.add(rows)
return idx
defdedupe_overlaps(hits: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
hits = sorted(hits, key=lambda h: (h["meta"]["doc_id"], h["meta"]["start_tok"]))
kept, last_end = [], -1
for h in hits:
if h["meta"]["start_tok"] >= last_end:
kept.append(h)
last_end = h["meta"]["end_tok"]
return kept
defevaluate(truth: List[Dict[str, str]], chunk_sizes: List[int], overlaps: List[int]) -> List[Dict[str, Any]]:
results = []
embedder = DummyEmbedder()
for cs, ov in itertools.product(chunk_sizes, overlaps):
start = time.time()
recalls, token_budgets = [], []
for item in truth:
idx = index_with_settings(item["doc_id"], item["text"], embedder, cs, ov)
hits = dedupe_overlaps(idx.similarity_search(item["question"], top_k=8))
topk = hits[:3]
recall = any(item["answer"].strip() in h["text"] for h in topk)
recalls.append(1if recall else0)
token_budgets.append(sum(tokens(h["text"]) for h in topk))
dur_ms = int((time.time() - start) * 1000)
results.append({
"chunk_size": cs,
"overlap": ov,
"recall_at_3": sum(recalls) / max(1, len(recalls)),
"avg_tokens_retrieved": sum(token_budgets) / max(1, len(token_budgets)),
"build_eval_ms": dur_ms
})
returnsorted(results, key=lambda r: (-r["recall_at_3"], r["avg_tokens_retrieved"]))像看交易一樣讀輸出。如果兩個設置recall打平，選token少的。如果一個設置recall大勝但只加了少量token開銷，拿下勝利。如果一個設置recall勝但token翻倍，判斷你的用戶和預算能不能承受。換嵌入器或重排器時，重跑這個網格。保持真實集小而誠實，十到二十個反映真實使用的問題夠你不自欺。
你可以用同樣習慣做摘要。用生產中的分塊方式拆長報告。跑精煉循環。檢查重要實體和數字是否在過程中保留。數你用的token。如果關鍵數字老丟，說明分塊錯了或預算太緊。先修分塊，再調prompt。
最后在索引旁留個簡短清單。記錄tokenizer、嵌入模型、分塊大小、重疊和日期。生產中感覺不對時，你會有個清晰的發貨記錄。
本文轉載自??AI大模型觀察站??，作者：AI研究生
