Ingestion Pipeline · 切块与嵌入

一、Ingestion 三件事

LlamaIndex 把这三步抽象成 transformation 序列,统一塞进 IngestionPipeline:

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding

pipeline = IngestionPipeline(transformations=[
    SentenceSplitter(chunk_size=512, chunk_overlap=50),
    OpenAIEmbedding(model="text-embedding-3-small"),
])
nodes = pipeline.run(documents=docs)

每个 transformation 输入 list[Node](Document 也是 Node),输出 list[Node]——这种统一接口让切块器、metadata 抽取器、embedding 可以自由组合。

二、Splitter 家族:五种主力

SentenceSplitter — 默认选它

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=512,            # token 预算,不是字符
    chunk_overlap=50,          # 相邻 chunk 重叠 token 数,防断句丢上下文
    separator=" ",              # 分词符
    paragraph_separator="\n\n\n",
    secondary_chunking_regex="[^,.;。?!]+[,.;。?!]?",  # 句子识别
)

MarkdownNodeParser — 技术文档神器

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
nodes = parser.get_nodes_from_documents(md_docs)

# 每个 node.metadata 里会有:
# {"header_1": "API 参考", "header_2": "用户", "header_3": "创建用户"}
# 这些层级会自动拼进 embedding,检索时对"怎么创建用户"命中率拉满

SemanticSplitter — 按主题切

from llama_index.core.node_parser import SemanticSplitterNodeParser

splitter = SemanticSplitterNodeParser(
    embed_model=embed_model,
    buffer_size=1,                # 比较相邻几个句子
    breakpoint_percentile_threshold=95,  # 95 分位数以上的语义距离就切
)

原理:对每两个相邻句子算 embedding 距离,距离大就在那里切。长文里主题漂移明显时效果最好——但成本是每次 ingestion 都要先跑一遍 embedding 计算距离。

三、chunk_size 和 overlap 怎么选

这是 RAG 最被问烂的问题。标准答案没有,但决策框架有:

overlap 的作用

overlap 主要防"答案正好在切割线上"的情况——比如一个定义横跨两段,不 overlap 就两边都不完整。经验值:

• chunk_overlap = chunk_size * 0.1 —— 保守,基本不丢
• Markdown/结构化文档 overlap 可以 0——标题本身就给了上下文
• overlap 太大(>20%)——存储翻倍,检索时相邻 chunk 语义高度相似,浪费 top_k 名额

四、Metadata Extractor:给 chunk 加智能标签

切完的 Node 除了基础 metadata,还可以用 LLM 再抽一层——摘要、标题、关键词、问题候选:

from llama_index.core.extractors import (
    TitleExtractor,                  # 给每段生成标题
    SummaryExtractor,                # 段落摘要(可带左右邻居)
    QuestionsAnsweredExtractor,      # 生成 N 个"这段能回答的问题"
    KeywordExtractor,                # 抽关键词
)

pipeline = IngestionPipeline(transformations=[
    SentenceSplitter(chunk_size=512),
    TitleExtractor(nodes=5),                    # 每组 5 个节点共享一个标题
    QuestionsAnsweredExtractor(questions=3, llm=llm),
    embed_model,
])

真正的秘密在于 QuestionsAnsweredExtractor——它让每个 chunk 带 3 个"本段能回答的假想问题",这些问题被一起 embed,query-document 匹配变成了 query-question 匹配,命中率显著提升。代价是每个 chunk 多 3 次 LLM 调用。

五、Embedding 模型选型

选型决策树

1. 纯中文场景 → BGE-large-zh-v1.5 或 BGE-M3(自部署),退而求其次 text-embedding-3-small
2. 中英混合 → BGE-M3 或 text-embedding-3-small
3. 要高性价比 + 不想自部署 → text-embedding-3-small,没得说
4. 要最高精度 → text-embedding-3-large 或 voyage-3
5. 长文档(单 chunk > 512 token)→ Jina v3(8K)、OpenAI v3(8K),别用 BGE(512)
6. 代码检索 → voyage-code-3 或 jina-embeddings-v2-code
7. 完全离线 → Ollama + nomic-embed-text 或 BGE-large

接入示例

# OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
embed = OpenAIEmbedding(model="text-embedding-3-small", embed_batch_size=100)

# 本地 BGE(HuggingFace)
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
embed = HuggingFaceEmbedding(
    model_name="BAAI/bge-large-zh-v1.5",
    device="cuda",               # 或 mps / cpu
    embed_batch_size=32,
    query_instruction="为这个句子生成表示以用于检索相关文章:",
)

# Ollama 本地
from llama_index.embeddings.ollama import OllamaEmbedding
embed = OllamaEmbedding(model_name="nomic-embed-text", base_url="http://localhost:11434")

# 全局设置(推荐)
from llama_index.core import Settings
Settings.embed_model = embed

并发与 batch 调优

• embed_batch_size:一次送给模型的 chunk 数。OpenAI API 上 100-500 比较稳;本地 GPU 看显存,32-128 常见
• num_workers:IngestionPipeline 并行度,CPU 密集型(splitter)开 4-8,网络密集型(API embedding)开 16-32
• API 限流 → 给 embedding provider 加重试和指数退避

六、Pipeline 缓存:快速迭代的关键

调整参数时,每次重跑全量嵌入既慢又烧钱。IngestionPipeline 有个内置 transformation 缓存——同样的输入 + 同样的 transformation hash 会命中缓存:

from llama_index.core.ingestion import IngestionPipeline, IngestionCache
from llama_index.storage.kvstore.redis import RedisKVStore

cache = IngestionCache(
    cache=RedisKVStore.from_host_and_port("localhost", 6379),
    collection="llama_cache",
)

pipeline = IngestionPipeline(
    transformations=[SentenceSplitter(chunk_size=512), embed_model],
    cache=cache,
)
nodes = pipeline.run(documents=docs)
pipeline.persist("./pipeline_storage")

命中机制:transformation 对每个输入 Node 算 hash,命中直接返回缓存输出。改 chunk_size 会导致 SentenceSplitter 的配置 hash 变,但下游 embedding 如果输入文本没变还是能命中。

七、增量索引:DocstoreStrategy 详解

Ch3 提过 UPSERTS,这里讲清楚所有策略:

from llama_index.core.ingestion import DocstoreStrategy
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[...],
    docstore=SimpleDocumentStore(),
    docstore_strategy=DocstoreStrategy.UPSERTS,  # 看下表
    vector_store=vector_store,
)

增量流程示例

# 第一次跑:全量
pipeline.run(documents=all_docs)
pipeline.persist("./pipeline")

# 第二天重新拉 Notion,得到新的 docs_today(可能有新增、修改、删除)
pipeline = IngestionPipeline.load("./pipeline")
# 策略设为 UPSERTS_AND_DELETE
pipeline.docstore_strategy = DocstoreStrategy.UPSERTS_AND_DELETE
pipeline.run(documents=docs_today)
# 自动:新增/修改 → 重嵌入;今天没出现的老 doc_id → 从 vector store 删除

八、并行与分布式

nodes = pipeline.run(
    documents=docs,
    num_workers=8,                  # 多进程
    show_progress=True,
)

几个要点:

• num_workers > 1 使用 Python multiprocessing——确保 transformation 对象能 pickle
• API 嵌入时 worker 多未必快,瓶颈是外部限流——把 embed_batch_size 加大往往比加 worker 更有效
• 上百万文档级别 → 用 llama_deploy 或自己写 Kafka/Celery 队列(Ch12 讲)

九、端到端一个完整 Pipeline

from llama_index.core import Settings, SimpleDirectoryReader
from llama_index.core.ingestion import IngestionPipeline, DocstoreStrategy, IngestionCache
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor, QuestionsAnsweredExtractor
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient

Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

vector_store = QdrantVectorStore(
    client=QdrantClient(url="http://localhost:6333"),
    collection_name="knowledge_v1",
)

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=50),
        TitleExtractor(nodes=5),
        QuestionsAnsweredExtractor(questions=3),
        Settings.embed_model,
    ],
    docstore=SimpleDocumentStore.from_persist_dir("./pipeline") if os.path.exists("./pipeline") else SimpleDocumentStore(),
    docstore_strategy=DocstoreStrategy.UPSERTS_AND_DELETE,
    vector_store=vector_store,
    cache=IngestionCache(),
)

docs = SimpleDirectoryReader("./data", filename_as_id=True).load_data()
nodes = pipeline.run(documents=docs, num_workers=4, show_progress=True)
pipeline.persist("./pipeline")

print(f"入库 {len(nodes)} 个 node")

十、常见坑

1. chunk_size 用字符数当 token:SentenceSplitter 是按 token,1 字符 ≠ 1 token,英文 1 token ≈ 4 字符、中文 1 token ≈ 1-2 字符。
2. MarkdownNodeParser 的 metadata 没设 excluded_embed_metadata_keys:header_1/2/3 本身会被拼进 embedding 是好事,但如果太长(比如中文标题拉一整行)会稀释内容。
3. SemanticSplitter 用了 OpenAI embedding:每次 ingestion 都多一批调用,用本地 BGE 跑语义切块,用 OpenAI embed 入库 才是经济做法。
4. Extractor 忘了限制范围:TitleExtractor/QuestionsAnsweredExtractor 默认对所有 chunk 调 LLM——海量数据几百刀一次。先抽样跑一批评估效果。
5. 不开 Pipeline cache:调参时每次重跑全量 embed,慢 + 贵。
6. embed_batch_size 太小:默认 10,OpenAI 其实支持 100+ 一批,改一下吞吐翻 10 倍。
7. BGE 中文没设 query_instruction:检索准确率掉一截。
8. 换了 embedding 模型不重建索引:维度和语义空间全变了,查询会胡乱召回——必须重嵌入。
9. num_workers 开太多触发限流:OpenAI RPM/TPM 上限,并发失败率上升反而慢。
10. Docstore 没 persist:下次跑没老 hash,UPSERTS 退化成全量重建。

十一、本章小结