LlamaIndex

Installation

pip install llama-index                          # meta package
pip install llama-index-llms-anthropic
pip install llama-index-embeddings-openai
pip install llama-index-vector-stores-chroma
pip install llama-parse

Global Settings

from llama_index.core import Settings
from llama_index.llms.anthropic import Anthropic
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = Anthropic(model="claude-sonnet-4-20250514")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 1024
Settings.chunk_overlap = 200

Documents and Nodes

from llama_index.core import Document
from llama_index.core.schema import TextNode, NodeRelationship, RelatedNodeInfo

doc = Document(
    text="...",
    metadata={"source": "manual.pdf", "page": 3},
    excluded_embed_metadata_keys=["source"],    # keep out of embedding input
    excluded_llm_metadata_keys=[],
)

node = TextNode(
    text="chunk text",
    metadata={"page": 3},
    relationships={NodeRelationship.SOURCE: RelatedNodeInfo(node_id=doc.doc_id)},
)

SimpleDirectoryReader

from llama_index.core import SimpleDirectoryReader

docs = SimpleDirectoryReader(
    input_dir="./data",
    recursive=True,
    required_exts=[".pdf", ".md", ".txt"],
    filename_as_id=True,
).load_data()

IngestionPipeline (with caching + dedup)

from llama_index.core.ingestion import IngestionPipeline, IngestionCache
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor, QuestionsAnsweredExtractor
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core.storage.docstore import SimpleDocumentStore
import chromadb

chroma_client = chromadb.PersistentClient(path="./chroma")
vector_store = ChromaVectorStore(chroma_collection=chroma_client.get_or_create_collection("kb"))

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=1024, chunk_overlap=200),
        TitleExtractor(nodes=5),
        QuestionsAnsweredExtractor(questions=3),
        Settings.embed_model,
    ],
    vector_store=vector_store,
    docstore=SimpleDocumentStore(),             # enables upsert + dedup
    cache=IngestionCache(),
)

nodes = pipeline.run(documents=docs, show_progress=True)
pipeline.persist("./pipeline_storage")

NodeParser Variants

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

sentence = SentenceSplitter(chunk_size=512, chunk_overlap=50)
window = SentenceWindowNodeParser.from_defaults(window_size=3,
    window_metadata_key="window", original_text_metadata_key="original_text")
semantic = SemanticSplitterNodeParser(
    buffer_size=1, breakpoint_percentile_threshold=95, embed_model=Settings.embed_model)
md = MarkdownNodeParser()
hierarchical = HierarchicalNodeParser.from_defaults(chunk_sizes=[2048, 512, 128])
code = CodeSplitter(language="python", chunk_lines=40, chunk_lines_overlap=10, max_chars=1500)

VectorStoreIndex and Query Engine

from llama_index.core import VectorStoreIndex, StorageContext

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(nodes=nodes, storage_context=storage_context)

query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact",        # compact | refine | tree_summarize | simple_summarize
    streaming=True,
)

streaming_response = query_engine.query("How does authentication work?")
for token in streaming_response.response_gen:
    print(token, end="")

Retriever + Response Synthesizer (explicit)

from llama_index.core import get_response_synthesizer
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.core.retrievers import VectorIndexRetriever

retriever = VectorIndexRetriever(index=index, similarity_top_k=10)
synth = get_response_synthesizer(response_mode="tree_summarize", streaming=True)
engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=synth,
    node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.7)],
)

Sub-Question Query Engine

from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata

tools = [
    QueryEngineTool(
        query_engine=finance_index.as_query_engine(),
        metadata=ToolMetadata(name="finance", description="Financial reports 2022-2024"),
    ),
    QueryEngineTool(
        query_engine=product_index.as_query_engine(),
        metadata=ToolMetadata(name="product", description="Product documentation"),
    ),
]

sub_q = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    use_async=True,
    verbose=True,
)
answer = sub_q.query("Compare 2023 revenue with product launch timeline")

Router Query Engine

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector, PydanticMultiSelector

router = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=tools,
    verbose=True,
)

Property Graph Index

from llama_index.core import PropertyGraphIndex
from llama_index.core.indices.property_graph import (
    SimpleLLMPathExtractor,
    ImplicitPathExtractor,
    SchemaLLMPathExtractor,
)

schema_extractor = SchemaLLMPathExtractor(
    llm=Settings.llm,
    possible_entities=["PERSON", "ORG", "PRODUCT"],
    possible_relations=["WORKS_AT", "MAKES", "COMPETES_WITH"],
)

graph_index = PropertyGraphIndex.from_documents(
    docs,
    kg_extractors=[schema_extractor, ImplicitPathExtractor()],
    show_progress=True,
)
graph_index.storage_context.persist(persist_dir="./graph_store")

graph_engine = graph_index.as_query_engine(include_text=True, similarity_top_k=5)

Agents with Tools (FunctionAgent / ReActAgent)

from llama_index.core.agent.workflow import FunctionAgent, AgentWorkflow
from llama_index.core.tools import FunctionTool

def get_weather(city: str) -> str:
    """Return current weather for a city."""
    return f"Sunny in {city}"

weather_tool = FunctionTool.from_defaults(fn=get_weather)

agent = FunctionAgent(
    tools=[weather_tool],
    llm=Settings.llm,
    system_prompt="You are a helpful assistant.",
)

import asyncio
async def main():
    result = await agent.run("What is the weather in Rome?")
    print(result)

asyncio.run(main())

LlamaParse Integration

from llama_parse import LlamaParse
from llama_index.core import VectorStoreIndex
import os

parser = LlamaParse(
    api_key=os.environ["LLAMA_CLOUD_API_KEY"],
    result_type="markdown",
    premium_mode=True,
)
docs = parser.load_data("10k.pdf")
index = VectorStoreIndex.from_documents(docs)

Observability (Callbacks)

from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler, TokenCountingHandler
import tiktoken

debug = LlamaDebugHandler(print_trace_on_end=True)
token_counter = TokenCountingHandler(
    tokenizer=tiktoken.encoding_for_model("gpt-4o").encode,
    verbose=True,
)
Settings.callback_manager = CallbackManager([debug, token_counter])

# After a query:
print("Embed tokens:", token_counter.total_embedding_token_count)
print("LLM prompt tokens:", token_counter.prompt_llm_token_count)
print("LLM completion tokens:", token_counter.completion_llm_token_count)

Arize Phoenix / OpenTelemetry

import llama_index.core
llama_index.core.set_global_handler("arize_phoenix")
# Launch Phoenix UI: import phoenix as px; px.launch_app()

Persistence and Reload

from llama_index.core import load_index_from_storage, StorageContext

index.storage_context.persist(persist_dir="./storage")

storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

Anti-Patterns

Production Checklist

• IngestionPipeline with docstore for incremental upserts
• Embedding + LLM pinned in Settings
• Streaming enabled for user-facing queries
• Node postprocessors: similarity cutoff + reranker (Cohere, Jina)
• Observability: callback manager with token counter + tracing
• Persisted storage (persist_dir) with content hashing
• Eval with RagasEvaluator or RelevancyEvaluator
• Fallback on SubQuestionQueryEngine failures (retry + single-engine)