AI-NEWS

Appearance

分块与 Embedding ​

🎯 学习目标

  • 权衡 chunk 大小、overlap 与语义边界
  • 完成从切分到 embedding 入库的可运行代码
  • 理解 query 与 document 必须使用同一 embedding 模型
  • 会用余弦相似度解释 Top-K 检索结果

引言 ​

Chunk 太大检索不精确，太小丢上下文；Embedding 模型换错则全库失效。本节给出完整可跑通的切分 + embedding + 检索示例，你可以换成自己的语料直接试验。

章节正文 ​

第 1 步：切分策略怎么选 ​

常用策略：

  1. 固定长度 + overlap：按字符或 Token 切，overlap 64–128 避免句边界断档
  2. 递归字符切分：优先 \n\n、\n、空格，保留段落结构
  3. 结构感知：Markdown 按标题、HTML 按 section

经验起点：中文 FAQ chunk_size=400–600 字符，overlap=10–15%。技术长文可 800–1200。务必用典型问题做 recall 测试再调参。

第 2 步：RecursiveCharacterTextSplitter 完整示例 ​

python
from dataclasses import dataclass

@dataclass
class Document:
    page_content: str
    metadata: dict

def recursive_split(text: str, chunk_size: int = 500, chunk_overlap: int = 80) -> list[str]:
    separators = ["\n\n", "\n", "。", " ", ""]
    chunks = [text]
    for sep in separators:
        if sep == "":
            break
        new_chunks = []
        for piece in chunks:
            if len(piece) <= chunk_size:
                new_chunks.append(piece)
                continue
            parts = piece.split(sep) if sep else [piece]
            buf = ""
            for part in parts:
                candidate = buf + (sep if buf else "") + part
                if len(candidate) <= chunk_size:
                    buf = candidate
                else:
                    if buf:
                        new_chunks.append(buf)
                    buf = part
            if buf:
                new_chunks.append(buf)
        chunks = new_chunks
    # overlap：相邻 chunk 共享尾部/头部（简化实现）
    if chunk_overlap and len(chunks) > 1:
        overlapped = [chunks[0]]
        for i in range(1, len(chunks)):
            tail = chunks[i - 1][-chunk_overlap:]
            overlapped.append(tail + chunks[i])
        chunks = overlapped
    return [c.strip() for c in chunks if c.strip()]

def split_documents(docs: list[Document], chunk_size=500, chunk_overlap=80) -> list[Document]:
    out = []
    for doc in docs:
        for i, chunk in enumerate(recursive_split(doc.page_content, chunk_size, chunk_overlap)):
            meta = {**doc.metadata, "chunk_index": i}
            out.append(Document(page_content=chunk, metadata=meta))
    return out

第 3 步：Embedding 与向量入库 ​

python
import math
from openai import OpenAI

client = OpenAI()

def embed_texts(texts: list[str], model="text-embedding-3-small") -> list[list[float]]:
    resp = client.embeddings.create(model=model, input=texts)
    return [d.embedding for d in resp.data]

def cosine_similarity(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(x * x for x in b))
    return dot / (na * nb + 1e-9)

class InMemoryVectorStore:
    def __init__(self):
        self.docs: list[Document] = []
        self.vectors: list[list[float]] = []

    def upsert(self, docs: list[Document], vectors: list[list[float]]):
        self.docs.extend(docs)
        self.vectors.extend(vectors)

    def search(self, query_vec: list[float], top_k: int = 5):
        scored = [(cosine_similarity(query_vec, v), d) for v, d in zip(self.vectors, self.docs)]
        scored.sort(key=lambda x: x[0], reverse=True)
        return scored[:top_k]

无 OpenAI 时可换 BAAI/bge-small-zh-v1.5 等开源模型，文档与 query 同一模型。

第 4 步：端到端：切分 → 入库 → 检索 ​

python
def build_index(raw_docs: list[Document]) -> InMemoryVectorStore:
    chunks = split_documents(raw_docs, chunk_size=500, chunk_overlap=80)
    vectors = embed_texts([c.page_content for c in chunks])
    store = InMemoryVectorStore()
    store.upsert(chunks, vectors)
    return store

def rag_retrieve(store: InMemoryVectorStore, question: str, top_k=5):
    q_vec = embed_texts([question])[0]
    return store.search(q_vec, top_k=top_k)

# --- 使用 ---
raw = [Document(page_content=open("faq.md").read(), metadata={"source": "faq.md"})]
store = build_index(raw)
hits = rag_retrieve(store, "报销流程是什么？")
for score, doc in hits:
    print(score, doc.metadata, doc.page_content[:80])

打印 Top-K 分数分布：若第一名与第二名差距极小，可能需要 rerank（4.6）。

第 5 步：调参检查清单 ​

  • chunk 是否切断关键定义（如「上述政策自 2025-01-01 起」与正文分离）？
  • overlap 是否足够让边界句出现在两个 chunk？
  • embedding 是否对中文/代码/领域词友好？
  • 是否误把 HTML 标签、页眉页脚 embed 进去？

每次调参用同一组 10 个问题记录 MRR@K 或人工「能否找到正确段落」命中率。

动手练习 ​

  1. 对两份 Markdown 跑 split_documents，打印 chunk 数与平均长度，调整 chunk_size 观察变化。
  2. 实现 InMemoryVectorStore.search，对 3 个问题打印 Top-3 与 cosine 分数。
  3. 故意用不同 embedding 模型 encode query vs document，观察检索如何失效。
  4. 在 metadata 保留 chunk_index，检索后在 UI mock 中展示「来源文件 + 块序号」。
  5. 选 5 个 golden questions，记录当前 chunk 参数下的 hit@3，作为 baseline。

常见问题 ​

Q：chunk_size 按字符还是 Token？

生产更应按 Token（与模型窗口一致），字符只是近似。中文 500 字符约 300–600 Token 视内容而定。用 tiktoken 或模型 tokenizer 估算更稳。

Q：OpenAI embedding 和开源 BGE 能混用吗？

不能混在同一索引。换模型必须 re-embed 全库。可建双索引 A/B 对比后再迁移。

Q：余弦相似度低就一定不相关吗？

不一定。不同模型校准不同；应用相对排序（Top-K）而非绝对阈值，或结合 rerank / Hybrid 检索。

本节小结 ​

切分保留语义边界与 overlap；Embedding 文档与 query 同源；入库后用 cosine Top-K 检索。4.4 的代码是你后续 Advanced RAG 的 baseline，请先跑通再优化。