LlamaIndex NodeParser — SentenceSplitter, TokenTextSplitter, SemanticSplitter selection

from llama_index.core.node_parser import (
    SentenceSplitter,
    TokenTextSplitter,
    SentenceWindowNodeParser,
    MarkdownNodeParser,
    HTMLNodeParser,
    CodeSplitter,
    HierarchicalNodeParser,
)

# --- Prose (default choice) ---
sentence_splitter = SentenceSplitter(
    chunk_size=512,           # target tokens per node
    chunk_overlap=64,         # overlap between adjacent nodes
    paragraph_separator="\n\n",
    secondary_chunking_regex="[^,.;。]+[,.;。]?",
)

# --- Markdown with header awareness ---
markdown_parser = MarkdownNodeParser(
    include_metadata=True,     # attach header context to each chunk
    include_prev_next_rel=True, # link chunks for context expansion
)

# --- Source code (per-language AST-aware) ---
python_splitter = CodeSplitter(
    language="python",         # respects function/class boundaries
    chunk_lines=40,
    chunk_lines_overlap=8,
    max_chars=1500,
)

# --- Hierarchical (parent-child for context expansion) ---
# Creates chunks at three levels: 2048, 512, 128 tokens
hierarchical = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]
)

# --- Sentence-window (retrieval on sentence, context is window) ---
sentence_window = SentenceWindowNodeParser.from_defaults(
    window_size=3,             # retrieve 1 sentence, expand to ±3 for context
    window_metadata_key="window",
    original_text_metadata_key="original_text",
)

# --- Route by file type ---
def choose_parser(doc):
    ext = doc.metadata.get("file_type", "")
    if ext == "md":
        return markdown_parser
    elif ext == "py":
        return python_splitter
    elif ext in ("html", "htm"):
        return HTMLNodeParser()
    else:
        return sentence_splitter

# Apply
all_nodes = []
for doc in documents:
    parser = choose_parser(doc)
    all_nodes.extend(parser.get_nodes_from_documents([doc]))

from llama_index.core import VectorStoreIndex
from llama_index.core.node_parser import SentenceWindowNodeParser
from llama_index.core.postprocessor import MetadataReplacementPostProcessor

# 1) Parse into sentence-window nodes
node_parser = SentenceWindowNodeParser.from_defaults(
    window_size=3,                     # ±3 sentences of context
    window_metadata_key="window",      # stored in metadata for later
    original_text_metadata_key="original_text",
)
nodes = node_parser.get_nodes_from_documents(documents)

# 2) Build the index — each node is embedded on its SINGLE sentence
#    but carries the surrounding window in metadata
index = VectorStoreIndex(nodes=nodes)

# 3) At query time, replace node text with the window before sending to LLM
query_engine = index.as_query_engine(
    similarity_top_k=6,
    node_postprocessors=[
        MetadataReplacementPostProcessor(target_metadata_key="window"),
    ],
)

response = query_engine.query("What is the refund policy?")
print(response)

# Retrieval precision: 1 sentence, so embedding matches focused topic
# Context to LLM: 7 sentences, so the model has full context to answer
# Common finding: ~10-20% hit-rate lift vs plain chunking with the same top_k

from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding

# Semantic splitter needs an embedding model to compute sentence similarities
embed_model = OpenAIEmbedding(model="text-embedding-3-small")

semantic_splitter = SemanticSplitterNodeParser(
    buffer_size=1,                          # sentences to group before comparing
    breakpoint_percentile_threshold=95,     # split at top-5% similarity drops
    embed_model=embed_model,
)

# Ingest a document
nodes = semantic_splitter.get_nodes_from_documents(documents)
print(f"Semantic splitter produced {len(nodes)} nodes")

# --- Trade-offs ---
# + Chunks align with topic boundaries
# + Great for messy / unstructured text
# - Slow: embeds each sentence pair to measure boundary
# - Expensive: at scale, embedding cost during ingestion is 2-5x higher
# - Chunk sizes vary; may exceed limits without a cap

# Recommended pattern: semantic split THEN cap size
from llama_index.core.node_parser import SentenceSplitter

def safe_semantic_parse(documents, max_chunk_tokens=1500):
    nodes = semantic_splitter.get_nodes_from_documents(documents)

    # Cap over-sized chunks with a fallback splitter
    capper = SentenceSplitter(chunk_size=max_chunk_tokens, chunk_overlap=100)
    final_nodes = []
    for node in nodes:
        if len(node.text.split()) > max_chunk_tokens * 0.8:  # rough proxy
            # Re-split the oversized chunk
            sub_nodes = capper.get_nodes_from_documents(
                [Document(text=node.text, metadata=node.metadata)]
            )
            final_nodes.extend(sub_nodes)
        else:
            final_nodes.append(node)
    return final_nodes

nodes = safe_semantic_parse(documents, max_chunk_tokens=1500)