Developing Llamaindex Systems

LlamaIndex Agentic Systems

Build production-grade agentic RAG systems with semantic ingestion, knowledge graphs, dynamic routing, and observability.

Quick Start

Build a working agent in 6 steps:

Step 1: Install Dependencies

pip install llama-index-core>=0.10.0 llama-index-llms-openai llama-index-embeddings-openai arize-phoenix

See scripts/requirements.txt for full pinned dependencies.

Step 2: Ingest with Semantic Chunking

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

embed_model = OpenAIEmbedding(model_name="text-embedding-3-small")
splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,
    embed_model=embed_model
)

docs = SimpleDirectoryReader(input_files=["data.pdf"]).load_data()
nodes = splitter.get_nodes_from_documents(docs)

Step 3: Build Index

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex(nodes, embed_model=embed_model)
index.storage_context.persist(persist_dir="./storage")

Step 4: Verify Index

# Confirm index built correctly
print(f"Indexed {len(index.docstore.docs)} document chunks")

# Preview a sample node
sample = list(index.docstore.docs.values())[0]
print(f"Sample chunk: {sample.text[:200]}...")

Step 5: Create Query Engine

query_engine = index.as_query_engine(similarity_top_k=5)
response = query_engine.query("What are the key concepts?")
print(response)

Step 6: Enable Observability

import phoenix as px
import llama_index.core

px.launch_app()
llama_index.core.set_global_handler("arize_phoenix")
# All subsequent queries are now traced

For production script, run: python scripts/ingest_semantic.py

---

Architecture Overview

Six pillars for agentic systems:

---

Decision Trees

Which Node Parser?

Is the content source code?
├─ Yes → CodeSplitter
│        language="python" (or typescript, javascript, java, go)
│        chunk_lines=40, chunk_lines_overlap=15
│        → See: references/ingestion.md#codesplitter
│
└─ No, it's documents:
    ├─ Need semantic coherence (legal, technical docs)?
    │   └─ Yes → SemanticSplitterNodeParser
    │            buffer_size=1 (sensitive), 3 (stable)
    │            breakpoint_percentile_threshold=95 (fewer), 70 (more)
    │            → See: references/ingestion.md#semanticsplitternodeparser
    │
    ├─ Prioritize speed → SentenceSplitter
    │        chunk_size=1024, chunk_overlap=20
    │        → See: references/ingestion.md#sentencesplitter
    │
    └─ Need fine-grained retrieval → SentenceWindowNodeParser
             window_size=3 (surrounding sentences in metadata)
             → See: references/ingestion.md#sentencewindownodeparser

Trade-off: Semantic chunking requires embedding calls during ingestion (cost + latency).

Which Retrieval Mode?

Query contains exact terms (function names, error codes, IDs)?
├─ Yes, exact match critical → BM25
│        retriever = BM25Retriever.from_defaults(nodes=nodes)
│        → See: references/retrieval-strategies.md#bm25retriever
│
├─ Conceptual/semantic query → Vector
│        retriever = index.as_retriever(similarity_top_k=5)
│        → See: references/context-rag.md
│
└─ Mixed or unknown query type → Hybrid (recommended default)
         alpha=0.5 (equal weight), 0.3 (favor BM25), 0.7 (favor vector)
         → See: references/retrieval-strategies.md#hybrid-search

Trade-off: Hybrid adds BM25 index overhead but provides most robust retrieval.

Which Graph Extractor?

Need document navigation only (prev/next/parent)?
├─ Yes → ImplicitPathExtractor (no LLM, zero cost)
│        → See: references/property-graphs.md#implicitpathextractor
│
└─ No, need semantic relationships:
    ├─ Fixed ontology required (regulated domain)?
    │   └─ Yes → SchemaLLMPathExtractor
    │            Pass schema: {"PERSON": ["WORKS_AT"], "COMPANY": ["LOCATED_IN"]}
    │            → See: references/property-graphs.md#schemallmpathextractor
    │
    └─ No, discovery/exploration:
        └─ SimpleLLMPathExtractor
           max_paths_per_chunk=10 (control noise)
           → See: references/property-graphs.md#simplellmpathextractor

Which Graph Retriever?

Need SQL-like aggregations (COUNT, SUM)?
├─ Yes, trusted environment → TextToCypherRetriever
│        Risk: LLM syntax errors, injection
│        → See: references/property-graphs.md#texttocypherretriever
│
├─ Yes, need safety → CypherTemplateRetriever
│        Pre-define: MATCH (p:Person {name: $name}) RETURN p
│        LLM only extracts parameters
│        → See: references/property-graphs.md#cyphertemplateretriever
│
└─ No, robustness priority → VectorContextRetriever
         Vector search → graph traversal (path_depth=2)
         Most reliable, no code generation
         → See: references/property-graphs.md#vectorcontextretriever

Which Agent Pattern?

Simple tool loop sufficient?
├─ Yes → ReAct Agent (FunctionCallingAgent)
│        Tools via FunctionTool or ToolSpec
│        → See: references/orchestration.md#react-agent-pattern
│
└─ No, need:
    ├─ Branching/cycles → Workflow
    │   → See: references/orchestration.md#branching
    ├─ Human-in-the-loop → Workflow (suspend/resume)
    │   → See: references/orchestration.md#human-in-the-loop
    ├─ Multi-agent handoff → Workflow + Concierge pattern
    │   → See: references/orchestration.md#concierge-multi-agent
    └─ Parallel execution → Workflow with multiple event emissions
        → See: references/orchestration.md#workflows

---

Common Patterns

Pattern 1: Metadata-Enriched Ingestion

from llama_index.core.extractors import TitleExtractor, SummaryExtractor, KeywordExtractor
from llama_index.core.ingestion import IngestionPipeline

pipeline = IngestionPipeline(
    transformations=[
        splitter,
        TitleExtractor(),
        SummaryExtractor(),
        KeywordExtractor(keywords=5),
        embed_model,
    ]
)
nodes = pipeline.run(documents=docs)

Pattern 2: PropertyGraphIndex with Hybrid Retrieval

from llama_index.core import PropertyGraphIndex
from llama_index.core.indices.property_graph import SimpleLLMPathExtractor

index = PropertyGraphIndex.from_documents(
    docs,
    embed_model=embed_model,
    kg_extractors=[SimpleLLMPathExtractor(max_paths_per_chunk=10)],
)

# Hybrid: vector search + graph traversal
retriever = index.as_retriever(include_text=True)

Pattern 3: Router with Multiple Engines

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.tools import QueryEngineTool

tools = [
    QueryEngineTool.from_defaults(
        query_engine=summary_engine,
        description="High-level summaries and overviews"
    ),
    QueryEngineTool.from_defaults(
        query_engine=detail_engine,
        description="Specific facts, numbers, and details"
    ),
]

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

Pattern 4: Event-Driven Workflow

from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Event

class QueryEvent(Event):
    query: str

class MyAgent(Workflow):
    @step
    async def classify(self, ev: StartEvent) -> QueryEvent:
        return QueryEvent(query=ev.get("query"))

    @step
    async def respond(self, ev: QueryEvent) -> StopEvent:
        result = self.query_engine.query(ev.query)
        return StopEvent(result=str(result))

# Run
agent = MyAgent(timeout=60)
result = await agent.run(query="What is X?")

Pattern 5: Reranking Pipeline

from llama_index.core.postprocessor import SimilarityPostprocessor, LLMRerank

query_engine = index.as_query_engine(
    similarity_top_k=10,  # Retrieve more
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),
        LLMRerank(top_n=3),  # Rerank to top 3
    ]
)

---

Script Reference

Adapt scripts by modifying configuration variables at the top of each file.

---

Reference Index

Load references based on task:

---

Troubleshooting

Agent says "I don't know" with relevant data

Diagnose:

# Open Phoenix UI at http://localhost:6006
# Navigate to Traces → Select query → Retrieval span → Retrieved Nodes

Fix:

# 1. Increase retrieval candidates
query_engine = index.as_query_engine(similarity_top_k=10)  # was 5

# 2. Add reranking to improve precision
from llama_index.core.postprocessor import LLMRerank
query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[LLMRerank(top_n=3)]
)

Verify: Re-run query, check Phoenix shows improved relevance scores (>0.7).

Semantic chunking too slow

Diagnose:

# Time the ingestion
import time
start = time.time()
nodes = splitter.get_nodes_from_documents(docs)
print(f"Chunking took {time.time() - start:.1f}s for {len(docs)} docs")

Fix:

# Option 1: Use local embeddings (no API calls)
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# Option 2: Hybrid strategy for large corpora
bulk_nodes = SentenceSplitter().get_nodes_from_documents(bulk_docs)
critical_nodes = SemanticSplitterNodeParser(...).get_nodes_from_documents(critical_docs)

Verify: Re-run with show_progress=True, confirm <1s per document.

Graph extraction producing noise

Diagnose:

# Check extracted triples
for node in index.property_graph_store.get_triplets():
    print(node)  # Look for irrelevant or duplicate relationships

Fix:

# Option 1: Reduce paths per chunk
SimpleLLMPathExtractor(max_paths_per_chunk=5)  # was 10

# Option 2: Use strict schema
SchemaLLMPathExtractor(
    possible_entities=["PERSON", "COMPANY"],
    possible_relations=["WORKS_AT", "FOUNDED"],
    strict=True
)

Verify: Re-index, confirm triplet count reduced and relationships are relevant.

Workflow step not triggering

Diagnose:

# Enable verbose mode
agent = MyWorkflow(timeout=60, verbose=True)
result = await agent.run(query="test")
# Check console for: [Step Name] Received event: EventType

Fix:

# Verify type hints match exactly
class MyEvent(Event):
    query: str

@step
async def my_step(self, ev: MyEvent) -> StopEvent:  # Type hint must be MyEvent
    ...

Verify: Verbose output shows [my_step] Received event: MyEvent.

Phoenix not showing traces

Diagnose:

import phoenix as px
session = px.launch_app()
print(f"Phoenix URL: {session.url}")  # Should print http://localhost:6006

Fix:

# MUST call BEFORE any LlamaIndex imports/operations
import phoenix as px
px.launch_app()

import llama_index.core
llama_index.core.set_global_handler("arize_phoenix")

# Now import and use LlamaIndex
from llama_index.core import VectorStoreIndex

Verify: Make a query, refresh Phoenix UI, trace appears within 5 seconds.

---

When Not to Use This Skill

This skill is specific to LlamaIndex in Python. Do not use for:

• LangChain projects — Different framework, different APIs
• Pure vector search without agents — Simpler solutions exist
• Non-Python environments — All examples are Python 3.9+
• Local-only / offline setups — Scripts default to OpenAI APIs; modification required for local models
• Simple Q&A bots — Overkill if you don't need graphs, routing, or workflows

If unsure: Check if your use case involves semantic chunking, knowledge graphs, query routing, or multi-step agents. If yes, this skill applies.

---

Glossary

Continuation Guide: llamaindex-agentic-systems

Created: 2025-12-28 Final Score: 92/100 (Grade A) Status: Complete — Ready for Use

---

Quick Reference

Skill Location

llamaindex-agentic-systems/
├── SKILL.md                        # Main entry point (~350 lines)
├── references/
│   ├── ingestion.md                # Semantic chunking (~245 lines)
│   ├── property-graphs.md          # Knowledge graphs (~280 lines)
│   ├── context-rag.md              # Query routing (~265 lines)
│   ├── orchestration.md            # Agents & Workflows (~310 lines)
│   └── observability.md            # Debugging & eval (~285 lines)
└── scripts/
    ├── requirements.txt            # Pinned dependencies
    ├── ingest_semantic.py          # Ingestion script (~280 lines)
    └── agent_workflow.py           # Agent template (~310 lines)

Total Lines: ~2,325

---

Installation

Option 1: Copy to Skills Directory

# User skills location
cp -r llamaindex-agentic-systems /mnt/skills/user/

# Verify
ls /mnt/skills/user/llamaindex-agentic-systems/SKILL.md

Option 2: Download and Extract

If provided as a .zip file:

unzip llamaindex-agentic-systems.zip -d /mnt/skills/user/

---

Activation Triggers

The skill activates when Claude detects queries involving:

Keywords:

• LlamaIndex, llama-index, llama_index
• SemanticSplitterNodeParser, IngestionPipeline
• PropertyGraphIndex, Neo4j (in LlamaIndex context)
• RouterQueryEngine, SubQuestionQueryEngine
• ReAct, Workflow, FunctionTool
• Arize Phoenix, LLMRerank

Task Patterns:

• "Build a LlamaIndex agent"
• "Set up semantic chunking"
• "Create a knowledge graph with LlamaIndex"
• "Debug my RAG pipeline"
• "Route queries to different engines"

---

Usage Examples

Example 1: Start a New Project

User: "Help me build a LlamaIndex agent with semantic chunking"

Claude reads: SKILL.md → Quick Start → references/ingestion.md

Output: Step-by-step guide with code snippets

Example 2: Debug Retrieval

User: "My LlamaIndex agent says 'I don't know' but the data exists"

Claude reads: SKILL.md → Troubleshooting → references/observability.md#retrieval-failures

Output: Debugging checklist with Phoenix trace analysis

Example 3: Add Graph Store

User: "How do I add Neo4j to my PropertyGraphIndex?"

Claude reads: SKILL.md → references/property-graphs.md#neo4jpropertygraphstore

Output: Configuration code and connection patterns

---

Development Session Summary

Steps Completed

Remediation Applied

1. ✅ Added "When Not to Use" section to SKILL.md 2. ✅ Enhanced TOC in orchestration.md 3. ✅ Specific exception handling in scripts 4. ✅ Added verification step in Quick Start 5. ✅ Version ceilings in requirements.txt 6. ✅ Standardized "See Also" sections 7. ✅ Deep-dive links in Troubleshooting

---

Updating the Skill

Adding New Content

1. Edit relevant reference file 2. Update SKILL.md Reference Index if adding new tasks 3. Update "See Also" links if adding cross-references

Modifying Scripts

1. Edit script in scripts/ directory 2. Test with sample documents 3. Update CONFIG section if adding new parameters

Version Bumps

1. Update version ceilings in requirements.txt 2. Test compatibility with new versions 3. Update "Last verified" date

---

Troubleshooting the Skill

Skill Not Activating

Cause: Trigger keywords not in description Fix: Verify SKILL.md frontmatter description contains relevant terms

Claude Missing Details

Cause: Information not in SKILL.md or top-level reference Fix: Add summary to SKILL.md with reference link

Scripts Failing

Cause: Dependency version mismatch Fix: Pin versions more tightly or test with newer versions

---

Source Corpus

The skill was derived from:

Document: "The Architecture of Agentic Systems: A Comprehensive Guide to LlamaIndex Implementation"

Key Sources:

• LlamaIndex official documentation (v0.10+)
• Arize Phoenix integration guides
• Neo4j property graph patterns
• ReAct agent research papers

---

Contact / Maintenance

This skill was created in a single session. For updates:

1. Re-run evaluation against latest improving-skills rubric 2. Apply any new remediation 3. Test with current LlamaIndex version 4. Update version constraints as needed

---

License

Content derived from public documentation and original synthesis. Use freely for skill development purposes.

Developing LlamaIndex Systems

A comprehensive Claude Code skill for building production-grade agentic RAG systems with LlamaIndex in Python.

 

Overview

This skill provides deep expertise in LlamaIndex's agentic capabilities, covering six core pillars:

When to Use This Skill

This skill activates when you ask Claude Code to:

• "Build a LlamaIndex agent"
• "Set up semantic chunking"
• "Index source code with CodeSplitter"
• "Implement hybrid search"
• "Create a knowledge graph with LlamaIndex"
• "Implement query routing"
• "Debug RAG pipeline"
• "Add Phoenix observability"
• "Create an event-driven workflow"

Or when discussing: PropertyGraphIndex, SemanticSplitterNodeParser, CodeSplitter, BM25Retriever, hybrid search, ReAct agent, Workflow pattern, LLMRerank, Text-to-Cypher

---

Installing with Skilz (Universal Installer)

The recommended way to install this skill across different AI coding agents is using the skilz universal installer. This skill supports the Agent Skill Standard, which means it works with 14+ coding agents including Claude Code, OpenAI Codex, Cursor, and Gemini CLI.

Install Skilz

pip install skilz

Quick Install (Claude Code)

# Install to user home (available in all projects)
skilz install -g https://github.com/SpillwaveSolutions/developing-llamaindex-systems

# Install to current project only
skilz install -g https://github.com/SpillwaveSolutions/developing-llamaindex-systems --project

Install from SkillzWave Marketplace

# Claude Code (user home)
skilz install SpillwaveSolutions_developing-llamaindex-systems/developing-llamaindex-systems

# Claude Code (project level)
skilz install SpillwaveSolutions_developing-llamaindex-systems/developing-llamaindex-systems --project

Other Agents

Add --project to any command above for project-level installation.

Skilz supports 14+ coding agents including Windsurf, Qwen Code, Aidr, and more. For the full list of supported platforms, visit SkillzWave.ai/platforms or see the skilz-cli GitHub repository.

View this skill on the marketplace: SkillzWave Listing

---

Quick Start

# 1. Semantic chunking
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding

splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,
    embed_model=OpenAIEmbedding()
)
nodes = splitter.get_nodes_from_documents(docs)

# 2. Build index
from llama_index.core import VectorStoreIndex
index = VectorStoreIndex(nodes)

# 3. Query
response = index.as_query_engine().query("What is X?")

Directory Structure

developing-llamaindex-systems/
├── SKILL.md                        # Main skill definition
├── README.md                       # This file
├── CONTINUATION-GUIDE.md           # Session continuation guide
├── references/
│   ├── ingestion.md               # Chunking strategies, IngestionPipeline
│   ├── retrieval-strategies.md    # BM25, hybrid search, fusion
│   ├── property-graphs.md         # PropertyGraphIndex, extractors, retrievers
│   ├── context-rag.md             # Query routing, decomposition, reranking
│   ├── orchestration.md           # ReAct agents, Workflows, multi-agent
│   └── observability.md           # Phoenix, debugging, evaluation
└── scripts/
    ├── requirements.txt           # Pinned dependencies
    ├── ingest_semantic.py         # Production ingestion script
    └── agent_workflow.py          # Event-driven workflow template

Reference Guide

Requirements

• Python 3.9+
• LlamaIndex 0.10+
• OpenAI API key (or configure local models)

Install dependencies:

pip install -r scripts/requirements.txt

Key Features

Semantic Chunking

Embedding-based chunking that preserves logical coherence, ideal for legal documents, technical manuals, and research papers.

Code Splitting

Language-aware splitting for source code with configurable chunk sizes and overlap.

Hybrid Retrieval

Combine BM25 keyword search with vector similarity using configurable alpha weighting.

Property Graphs

Hybrid retrieval combining vector search with graph traversal. Supports Neo4j and in-memory graph stores.

Query Routing

LLM-based routing to direct queries to specialized engines based on intent.

Event-Driven Workflows

Type-safe, async workflows with branching, cycles, and human-in-the-loop support.

Observability

One-line Arize Phoenix integration for full tracing, plus custom handlers for metrics and alerting.

When NOT to Use This Skill

• LangChain projects - Different framework
• Non-Python environments - Python 3.9+ only
• Simple Q&A bots - Overkill if you don't need graphs, routing, or workflows
• Offline/local-only setups - Scripts default to OpenAI APIs; modification required for local models

---

Links

• SkillzWave Marketplace - Largest Agentic Marketplace for Agent Skills
• SpillWave - Leaders in AI Agent Development
• Agent Skill Standard - Cross-platform skill specification

License

MIT

Context RAG

Dynamic query routing, decomposition, and retrieval refinement.

Contents

• RouterQueryEngine
• Selector Types
• QueryEngineTool Setup
• Router Implementation
• Query Routing Examples
• SubQuestionQueryEngine
• NodePostprocessors
• SimilarityPostprocessor
• LLMRerank
• CohereRerank
• SentenceTransformerRerank
• Postprocessor Comparison
• Reranking Pipeline
• Complete Examples
• See Also

---

RouterQueryEngine

Routes queries to the optimal data source from a pool of available engines.

Architecture

User Query → Selector → [Engine A | Engine B | Engine C] → Response

The selector analyzes the query and chooses which engine(s) to invoke based on tool descriptions.

Selector Types

LLMSingleSelector

Chooses exactly one engine. Best for mutually exclusive sources.

from llama_index.core.selectors import LLMSingleSelector

selector = LLMSingleSelector.from_defaults()

Use Case: "SQL Database" vs "Vector Store" — query goes to one or the other.

LLMMultiSelector

Chooses one or more engines. Aggregates results.

from llama_index.core.selectors import LLMMultiSelector

selector = LLMMultiSelector.from_defaults()

Use Case: "Compare sales data with market sentiment" — triggers both SQL and news engines.

PydanticSingleSelector

Structured JSON output for reliability. Reduces parsing errors.

from llama_index.core.selectors import PydanticSingleSelector

selector = PydanticSingleSelector.from_defaults()

Use Case: Production systems requiring deterministic selection.

QueryEngineTool Setup

Each engine wrapped as a tool with a description. The description is critical — it's the "system prompt" for routing decisions.

from llama_index.core.tools import QueryEngineTool

# Good descriptions: specific, action-oriented
summary_tool = QueryEngineTool.from_defaults(
    query_engine=summary_engine,
    description=(
        "Useful for high-level summaries, thematic overviews, "
        "and questions about what the document is about."
    )
)

detail_tool = QueryEngineTool.from_defaults(
    query_engine=detail_engine,
    description=(
        "Useful for retrieving specific facts, numbers, dates, "
        "and precise details from the document."
    )
)

sql_tool = QueryEngineTool.from_defaults(
    query_engine=sql_engine,
    description=(
        "Useful for structured data queries: sales figures, "
        "revenue, counts, and database records."
    )
)

Description Guidelines:

• Start with "Useful for..."
• List specific query types
• Mention data characteristics
• Keep under 200 characters

Router Implementation

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

router = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[summary_tool, detail_tool, sql_tool],
    verbose=True,  # Log routing decisions
)

# Queries route automatically
response = router.query("What is this document about?")  # → summary_tool
response = router.query("What was Q3 revenue?")          # → sql_tool

Query Routing Examples

---

SubQuestionQueryEngine

Decomposes complex queries into atomic sub-queries, executes them (optionally in parallel), and synthesizes results.

Mechanism

Complex Query → [Sub-Q1, Sub-Q2, ...] → [Answer1, Answer2, ...] → Synthesis

Implementation

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

# Define tools for different data sources
tools = [
    QueryEngineTool.from_defaults(
        query_engine=apple_engine,
        description="Information about Apple products and specifications"
    ),
    QueryEngineTool.from_defaults(
        query_engine=samsung_engine,
        description="Information about Samsung products and specifications"
    ),
]

engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    verbose=True,
)

# Complex query triggers decomposition
response = engine.query(
    "Compare the battery life of iPhone 15 and Galaxy S24"
)
# Sub-Q1: "What is the battery life of iPhone 15?" → apple_engine
# Sub-Q2: "What is the battery life of Galaxy S24?" → samsung_engine
# Synthesis: Combined comparison

When to Use

• Comparative questions ("Compare X and Y")
• Multi-part queries ("What is A and how does it relate to B?")
• Cross-source aggregation
• Questions requiring multiple data sources

Configuration

engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    use_async=True,   # Parallel sub-query execution
    verbose=True,
)

---

NodePostprocessors

Refine retrieved nodes after initial retrieval. Applied in sequence.

SimilarityPostprocessor

Filters nodes below a similarity threshold.

from llama_index.core.postprocessor import SimilarityPostprocessor

postprocessor = SimilarityPostprocessor(
    similarity_cutoff=0.7,  # Remove nodes below 0.7 similarity
)

When to Use:

• Prevent low-quality context from reaching LLM
• Weak vector matches causing hallucinations
• Cost reduction (fewer tokens to process)

LLMRerank

LLM reads query + nodes and re-scores relevance.

from llama_index.core.postprocessor import LLMRerank

postprocessor = LLMRerank(
    top_n=3,                    # Keep top 3 after reranking
    choice_batch_size=5,        # Process 5 nodes at a time
)

Characteristics:

• Highest precision
• High latency and cost (LLM call per batch)
• Best for critical queries

CohereRerank

Cross-encoder reranking via Cohere API.

from llama_index.postprocessor.cohere_rerank import CohereRerank

postprocessor = CohereRerank(
    api_key="your-api-key",
    top_n=3,
    model="rerank-english-v3.0",
)

Prerequisites:

pip install llama-index-postprocessor-cohere-rerank

Characteristics:

• Excellent accuracy
• Faster than LLMRerank
• API cost per request

SentenceTransformerRerank

Local cross-encoder model. No API costs.

from llama_index.core.postprocessor import SentenceTransformerRerank

postprocessor = SentenceTransformerRerank(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2",
    top_n=3,
)

Characteristics:

• No API costs
• Runs locally (GPU recommended)
• Good accuracy, fast inference

---

Postprocessor Comparison

Selection Guide

Budget constrained?
├─ Yes → SentenceTransformerRerank (local, free)
│
└─ No, accuracy priority:
    ├─ Latency sensitive → CohereRerank
    └─ Quality critical → LLMRerank

---

Reranking Pipeline

Combine postprocessors for optimal results.

Pattern: Filter → Rerank

from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    SentenceTransformerRerank,
)

query_engine = index.as_query_engine(
    similarity_top_k=10,  # Retrieve more initially
    node_postprocessors=[
        # Step 1: Remove obvious misses
        SimilarityPostprocessor(similarity_cutoff=0.5),
        # Step 2: Rerank survivors
        SentenceTransformerRerank(top_n=3),
    ]
)

Pattern: Two-Stage Reranking

from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    SentenceTransformerRerank,
)
from llama_index.postprocessor.cohere_rerank import CohereRerank

query_engine = index.as_query_engine(
    similarity_top_k=20,
    node_postprocessors=[
        # Stage 1: Fast local rerank
        SentenceTransformerRerank(top_n=10),
        # Stage 2: Precise API rerank
        CohereRerank(top_n=3),
    ]
)

Pattern: Conditional Reranking

Apply expensive reranking only when needed:

class ConditionalReranker:
    def __init__(self, threshold=0.8):
        self.threshold = threshold
        self.expensive_reranker = LLMRerank(top_n=3)
    
    def postprocess_nodes(self, nodes, query_bundle):
        # Check if top result is confident
        if nodes and nodes[0].score > self.threshold:
            return nodes[:3]  # Skip reranking
        # Otherwise, apply expensive reranking
        return self.expensive_reranker.postprocess_nodes(nodes, query_bundle)

---

Complete Examples

Multi-Source Router with Reranking

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import PydanticSingleSelector
from llama_index.core.tools import QueryEngineTool
from llama_index.core.postprocessor import SentenceTransformerRerank

# Build engines with reranking
def build_engine(index):
    return index.as_query_engine(
        similarity_top_k=10,
        node_postprocessors=[
            SentenceTransformerRerank(top_n=3),
        ]
    )

# Setup tools
tools = [
    QueryEngineTool.from_defaults(
        query_engine=build_engine(docs_index),
        description="Technical documentation and guides"
    ),
    QueryEngineTool.from_defaults(
        query_engine=build_engine(faq_index),
        description="Frequently asked questions and answers"
    ),
    QueryEngineTool.from_defaults(
        query_engine=build_engine(changelog_index),
        description="Version history and release notes"
    ),
]

# Router
router = RouterQueryEngine(
    selector=PydanticSingleSelector.from_defaults(),
    query_engine_tools=tools,
)

response = router.query("What changed in version 2.0?")  # → changelog

SubQuestion with Cross-Source Synthesis

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

tools = [
    QueryEngineTool.from_defaults(
        query_engine=financials_engine,
        description="Financial data: revenue, costs, margins"
    ),
    QueryEngineTool.from_defaults(
        query_engine=market_engine,
        description="Market analysis: competitors, trends, sentiment"
    ),
    QueryEngineTool.from_defaults(
        query_engine=product_engine,
        description="Product specifications and roadmap"
    ),
]

engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    use_async=True,
)

response = engine.query(
    "How does our product positioning compare to competitors "
    "and what's the revenue impact?"
)
# Decomposes into product, market, and financial sub-queries

---

See Also

• ../SKILL.md — Return to main skill overview
• ingestion.md — Data preparation before retrieval
• property-graphs.md — Graph-based retrieval as router target
• orchestration.md — Integrate routers into agent workflows
• observability.md — Debug routing decisions

Semantic Ingestion

Deep dive into data ingestion, chunking strategies, and metadata enrichment.

Contents

• IngestionPipeline
• Node Parsers
• CodeSplitter
• SemanticSplitterNodeParser
• SentenceSplitter
• SentenceWindowNodeParser
• Comparison Table
• Metadata Extractors
• Complete Pipeline Example
• Performance Tuning
• See Also

---

IngestionPipeline

Central processing unit for transforming documents into queryable nodes.

Architecture

Documents → [Parse] → [Transform] → [Embed] → Nodes

Three stages: 1. Parsing: Raw files (PDF, HTML, JSON) → Document objects 2. Transformation: Documents → Node objects via node parsers 3. Embedding: Nodes → Vector representations

Basic Setup

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

pipeline = IngestionPipeline(
    transformations=[
        SemanticSplitterNodeParser(
            buffer_size=1,
            breakpoint_percentile_threshold=95,
            embed_model=OpenAIEmbedding()
        ),
        OpenAIEmbedding(),
    ]
)

nodes = pipeline.run(documents=documents)

Incremental Updates

Pipeline supports deduplication for incremental indexing via caching.

In-Memory Cache

from llama_index.core.ingestion import IngestionPipeline, IngestionCache

cache = IngestionCache()
pipeline = IngestionPipeline(
    transformations=[...],
    cache=cache,
)

# First run: processes all
nodes = pipeline.run(documents=docs)

# Second run: only processes new/changed documents
nodes = pipeline.run(documents=updated_docs)

Redis Cache (Production)

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

# Connect to Redis
redis_kvstore = RedisKVStore(
    host="localhost",
    port=6379,
    # password="your-password",  # If auth required
)

cache = IngestionCache(cache=redis_kvstore)

pipeline = IngestionPipeline(
    transformations=[splitter, embed_model],
    cache=cache,
)

# Documents are fingerprinted; unchanged docs skip processing
nodes = pipeline.run(documents=docs)

Prerequisites:

pip install redis llama-index-storage-kvstore-redis

Cache Behavior

Clearing Cache

# Clear entire cache
cache.clear()

# Or with Redis, delete specific keys
redis_kvstore.delete("ingestion_cache:doc_hash_abc123")

---

Node Parsers

CodeSplitter

AST-aware code chunking that respects function and class boundaries. Uses tree-sitter for parsing.

Why AST-Aware Chunking?

Standard text splitters break code at arbitrary character boundaries, losing context:

# Bad: Function split mid-implementation
def calculate_tax(amount):
    rate = 0.15
    if amount > 10000:
        rate = 0.25
# --- chunk boundary ---
    return amount * rate  # Lost context!

CodeSplitter chunks at logical boundaries (functions, classes, methods).

Basic Usage

from llama_index.core.node_parser import CodeSplitter

splitter = CodeSplitter(
    language="python",      # Required: target language
    chunk_lines=40,         # Lines per chunk
    chunk_lines_overlap=15, # Overlap between chunks
    max_chars=1500,         # Maximum characters per chunk
)

# From documents
nodes = splitter.get_nodes_from_documents(documents)

Prerequisites:

pip install llama-index tree-sitter tree-sitter-languages

Supported Languages

Configuration

splitter = CodeSplitter(
    language="python",
    chunk_lines=40,         # Target lines per chunk
    chunk_lines_overlap=15, # Context preservation
    max_chars=1500,         # Hard limit for LLM context
)

Parameters:

Extracting Code Metadata

Combine CodeSplitter with metadata extraction for richer search:

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import CodeSplitter
from llama_index.core.extractors import SummaryExtractor

pipeline = IngestionPipeline(
    transformations=[
        CodeSplitter(language="python", chunk_lines=40),
        SummaryExtractor(),  # Generate natural language descriptions
        embed_model,
    ]
)

nodes = pipeline.run(documents=code_documents)
# Each node now has:
# - text: The code chunk
# - metadata.section_summary: "This function calculates tax..."

Multi-Language Corpus

Handle mixed-language codebases:

from pathlib import Path

LANGUAGE_MAP = {
    ".py": "python",
    ".ts": "typescript",
    ".tsx": "tsx",
    ".js": "javascript",
    ".jsx": "jsx",
    ".java": "java",
    ".go": "go",
}


def get_code_splitter(file_path: str) -> CodeSplitter:
    """Return appropriate splitter for file type."""
    ext = Path(file_path).suffix.lower()
    language = LANGUAGE_MAP.get(ext, "python")
    return CodeSplitter(
        language=language,
        chunk_lines=40,
        chunk_lines_overlap=15,
    )


# Process each file with correct splitter
all_nodes = []
for doc in documents:
    splitter = get_code_splitter(doc.metadata.get("file_path", ""))
    nodes = splitter.get_nodes_from_documents([doc])

    # Add language metadata
    for node in nodes:
        node.metadata["language"] = splitter.language
        node.metadata["source_type"] = "code"

    all_nodes.extend(nodes)

When to Use

• Source code indexing: SDK code, library implementations
• Code search: Find functions by name or behavior
• Documentation generation: Code + docs in unified corpus
• Tutorial writing: Reference actual implementation patterns

---

SemanticSplitterNodeParser

Embedding-based chunking that preserves logical coherence.

Mechanism

1. Text divided into sentences 2. Buffer of sentences encoded to vectors 3. Cosine similarity calculated between adjacent buffers 4. Split occurs when similarity drops below threshold

Configuration

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

splitter = SemanticSplitterNodeParser(
    buffer_size=1,                        # Sentences per buffer (1-5)
    breakpoint_percentile_threshold=95,   # Split sensitivity (0-100)
    embed_model=OpenAIEmbedding(),        # Required
)

Parameters

When to Use

• Legal documents (preserve clause boundaries)
• Technical manuals (preserve procedure steps)
• Research papers (preserve argument flow)
• Any document where logical coherence matters

Trade-offs

• Pro: Semantically coherent chunks
• Con: Requires embedding calls during ingestion (cost + latency)
• Con: Variable chunk sizes (harder to predict token usage)

---

SentenceSplitter

Fixed token-window splitting. Fast but semantically blind.

Configuration

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=1024,      # Tokens per chunk
    chunk_overlap=20,     # Overlap between chunks
)

When to Use

• Large corpus bulk processing
• Speed priority over precision
• Homogeneous content (news articles, blog posts)
• Budget constraints (no embedding cost during ingestion)

---

SentenceWindowNodeParser

Single-sentence nodes with surrounding context stored in metadata.

Configuration

from llama_index.core.node_parser import SentenceWindowNodeParser

splitter = SentenceWindowNodeParser(
    window_size=3,              # Sentences before/after
    window_metadata_key="window",
    original_text_metadata_key="original_text",
)

Retrieval Pattern

Requires MetadataReplacementPostProcessor to expand context at query time:

from llama_index.core.postprocessor import MetadataReplacementPostProcessor

query_engine = index.as_query_engine(
    similarity_top_k=5,
    node_postprocessors=[
        MetadataReplacementPostProcessor(target_metadata_key="window")
    ]
)

When to Use

• Fine-grained retrieval needs (exact sentence matching)
• QA over dense technical content
• When both precision and context matter

---

Node Parser Comparison

---

Metadata Extractors

Enrich nodes with self-describing context.

TitleExtractor

Infers document and section titles via LLM.

from llama_index.core.extractors import TitleExtractor

extractor = TitleExtractor(
    nodes=5,  # Number of nodes to derive title from
)

Output metadata: {"document_title": "...", "section_title": "..."}

SummaryExtractor

Generates concise summary of node content.

from llama_index.core.extractors import SummaryExtractor

extractor = SummaryExtractor(
    summaries=["self", "prev", "next"],  # What to summarize
)

Output metadata: {"section_summary": "...", "prev_section_summary": "..."}

KeywordExtractor

Extracts entities and keywords for hybrid search.

from llama_index.core.extractors import KeywordExtractor

extractor = KeywordExtractor(
    keywords=5,  # Number of keywords
)

Output metadata: {"keywords": ["revenue", "Q3", "growth"]}

Combining Extractors

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.extractors import (
    TitleExtractor, SummaryExtractor, KeywordExtractor
)

pipeline = IngestionPipeline(
    transformations=[
        splitter,
        TitleExtractor(),
        SummaryExtractor(),
        KeywordExtractor(keywords=5),
        embed_model,
    ]
)

Result: Each node contains:

Node(
    text="...",
    metadata={
        "document_title": "Q3 Earnings Report",
        "section_title": "Risk Factors",
        "section_summary": "Discusses market volatility...",
        "keywords": ["risk", "volatility", "market"],
    }
)

---

Complete Pipeline Example

Production-ready ingestion with semantic chunking and full metadata:

import os
from llama_index.core import SimpleDirectoryReader
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.core.extractors import (
    TitleExtractor, SummaryExtractor, KeywordExtractor
)
from llama_index.embeddings.openai import OpenAIEmbedding

# Setup
embed_model = OpenAIEmbedding(model_name="text-embedding-3-small")

# Load documents
docs = SimpleDirectoryReader("./data").load_data()

# Build pipeline
pipeline = IngestionPipeline(
    transformations=[
        SemanticSplitterNodeParser(
            buffer_size=1,
            breakpoint_percentile_threshold=95,
            embed_model=embed_model,
        ),
        TitleExtractor(),
        KeywordExtractor(keywords=5),
        embed_model,
    ]
)

# Execute
nodes = pipeline.run(documents=docs, show_progress=True)

print(f"Created {len(nodes)} nodes")

---

Performance Tuning

Reduce Semantic Chunking Latency

1. Lower buffer_size: buffer_size=1 is fastest 2. Local embeddings: Replace OpenAI with HuggingFace (see below) 3. Batch processing: Process documents in batches to amortize overhead

Local Embedding Models

Eliminate API costs and latency with local models.

HuggingFace Embeddings

from llama_index.embeddings.huggingface import HuggingFaceEmbedding

# Small and fast (recommended for development)
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# Better quality (production)
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en-v1.5")

# Best quality (if GPU available)
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-large-en-v1.5")

Prerequisites:

pip install llama-index-embeddings-huggingface sentence-transformers

Model Comparison

GPU Acceleration

# Automatic GPU detection
embed_model = HuggingFaceEmbedding(
    model_name="BAAI/bge-base-en-v1.5",
    device="cuda",  # or "mps" for Apple Silicon
)

# Verify GPU usage
import torch
print(f"Using device: {embed_model._device}")
print(f"CUDA available: {torch.cuda.is_available()}")

Using with Semantic Splitter

# Local embeddings for both splitting and indexing
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,
    embed_model=embed_model,  # Uses local model
)

# No API calls during ingestion
nodes = splitter.get_nodes_from_documents(docs, show_progress=True)

Hybrid Strategy for Large Corpora

Use SentenceSplitter for bulk content, SemanticSplitter for critical documents:

# Bulk content
bulk_splitter = SentenceSplitter(chunk_size=1024)
bulk_nodes = bulk_splitter.get_nodes_from_documents(bulk_docs)

# Critical documents
semantic_splitter = SemanticSplitterNodeParser(...)
critical_nodes = semantic_splitter.get_nodes_from_documents(critical_docs)

# Combine
all_nodes = bulk_nodes + critical_nodes

---

See Also

• ../SKILL.md — Return to main skill overview
• property-graphs.md — Store nodes in PropertyGraphIndex
• context-rag.md — Query routing and postprocessing
• observability.md — Debug ingestion issues

Observability

Tracing, debugging, and evaluation for agentic systems.

Contents

• Instrumentation Module
• Dispatchers
• Spans and Events
• Custom Handlers
• Arize Phoenix Integration
• Setup
• Trace Visualization
• Debugging Scenarios
• Retrieval Failures
• Agent Loops
• Latency Issues
• Routing Problems
• Evaluators
• FaithfulnessEvaluator
• RelevancyEvaluator
• CorrectnessEvaluator
• Evaluation Pipeline
• Production Monitoring
• See Also

---

Instrumentation Module

Replaces legacy callback system in LlamaIndex v0.10+. Provides deep visibility into LLM operations.

Architecture

LlamaIndex Module → Dispatcher → [SpanHandler | EventHandler] → Output

Dispatchers

Central hub that broadcasts events. Every module has its own dispatcher.

from llama_index.core.instrumentation import get_dispatcher

# Get dispatcher for a specific module
dispatcher = get_dispatcher(__name__)

Spans and Events

Spans track:

• Retrieval operations
• LLM calls
• Embedding generation
• Tool execution

Custom Handlers

Create custom handlers for logging, metrics, or external platforms.

Basic Logging Handler

from llama_index.core.instrumentation.event_handlers import BaseEventHandler
from llama_index.core.instrumentation.span_handlers import BaseSpanHandler

class LoggingSpanHandler(BaseSpanHandler):
    def new_span(self, id, parent_span_id, **kwargs):
        print(f"Span started: {id}")

    def end_span(self, id, **kwargs):
        print(f"Span ended: {id}")

class LoggingEventHandler(BaseEventHandler):
    def handle(self, event):
        print(f"Event: {event.class_name()} - {event.dict()}")

# Register handlers
from llama_index.core.instrumentation import get_dispatcher
dispatcher = get_dispatcher()
dispatcher.add_span_handler(LoggingSpanHandler())
dispatcher.add_event_handler(LoggingEventHandler())

Metrics Collection Handler

import time
from collections import defaultdict
from llama_index.core.instrumentation.span_handlers import BaseSpanHandler

class MetricsSpanHandler(BaseSpanHandler):
    def __init__(self):
        super().__init__()
        self.span_starts = {}
        self.metrics = defaultdict(list)

    def new_span(self, id, parent_span_id, **kwargs):
        self.span_starts[id] = time.time()

    def end_span(self, id, **kwargs):
        if id in self.span_starts:
            duration = time.time() - self.span_starts[id]
            span_type = kwargs.get("span_type", "unknown")
            self.metrics[span_type].append(duration)
            del self.span_starts[id]

    def get_stats(self):
        return {
            span_type: {
                "count": len(durations),
                "avg_ms": sum(durations) / len(durations) * 1000,
                "max_ms": max(durations) * 1000,
            }
            for span_type, durations in self.metrics.items()
        }

# Usage
metrics_handler = MetricsSpanHandler()
dispatcher.add_span_handler(metrics_handler)

# After some queries...
print(metrics_handler.get_stats())
# {"retrieval": {"count": 10, "avg_ms": 45.2, "max_ms": 120.1}, ...}

Token Usage Tracker

from llama_index.core.instrumentation.event_handlers import BaseEventHandler
from llama_index.core.instrumentation.events.llm import LLMCompletionEndEvent

class TokenTracker(BaseEventHandler):
    def __init__(self):
        self.total_prompt_tokens = 0
        self.total_completion_tokens = 0

    @classmethod
    def class_name(cls) -> str:
        return "TokenTracker"

    def handle(self, event):
        if isinstance(event, LLMCompletionEndEvent):
            if hasattr(event, "token_counts"):
                self.total_prompt_tokens += event.token_counts.get("prompt", 0)
                self.total_completion_tokens += event.token_counts.get("completion", 0)

    def get_usage(self):
        return {
            "prompt_tokens": self.total_prompt_tokens,
            "completion_tokens": self.total_completion_tokens,
            "total_tokens": self.total_prompt_tokens + self.total_completion_tokens,
        }

# Usage
token_tracker = TokenTracker()
dispatcher.add_event_handler(token_tracker)

# After queries...
print(token_tracker.get_usage())

Alerting Handler

class AlertingEventHandler(BaseEventHandler):
    def __init__(self, latency_threshold_ms=5000):
        self.latency_threshold = latency_threshold_ms / 1000

    def handle(self, event):
        # Alert on slow retrievals
        if hasattr(event, "duration") and event.duration > self.latency_threshold:
            self.send_alert(f"Slow operation: {event.class_name()} took {event.duration:.2f}s")

        # Alert on errors
        if hasattr(event, "exception") and event.exception:
            self.send_alert(f"Error in {event.class_name()}: {event.exception}")

    def send_alert(self, message):
        # Integration: Slack, PagerDuty, email, etc.
        print(f"ALERT: {message}")

---

Arize Phoenix Integration

Native observability platform adhering to OpenInference standard.

Setup

One-line integration instruments entire LlamaIndex stack:

import phoenix as px
import llama_index.core

# 1. Launch Phoenix server (local)
px.launch_app()

# 2. Set global handler
llama_index.core.set_global_handler("arize_phoenix")

# All subsequent operations are traced
response = query_engine.query("What is X?")

Prerequisites:

pip install arize-phoenix

Phoenix UI available at http://localhost:6006 after launch.

Cloud Phoenix

For production, use hosted Phoenix:

import os
os.environ["PHOENIX_API_KEY"] = "your-api-key"
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com"

import llama_index.core
llama_index.core.set_global_handler("arize_phoenix")

Trace Visualization

Phoenix provides:

---

Debugging Scenarios

Retrieval Failures

Symptom: Agent says "I don't know" despite relevant data existing.

Debug Steps:

1. Open Phoenix trace for the query 2. Find "Retrieval" span 3. Inspect retrieved chunks

Common Causes:

Fix Pattern:

# Before: fixed chunking
splitter = SentenceSplitter(chunk_size=256)

# After: semantic chunking
splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,
    embed_model=embed_model
)

Agent Loops

Symptom: Agent keeps calling tools without reaching final answer.

Debug Steps:

1. Open Phoenix trace 2. Look for repeated "Action" spans 3. Check "Observation" payload for each

Common Causes:

Fix Pattern:

# Before: vague tool output
def search(query: str) -> str:
    results = db.search(query)
    return str(results)  # Raw object dump

# After: structured output
def search(query: str) -> str:
    """Search database and return formatted results."""
    results = db.search(query)
    if not results:
        return "No results found for this query."
    return json.dumps({
        "count": len(results),
        "results": results[:5],
        "summary": f"Found {len(results)} matches"
    })

Latency Issues

Symptom: Queries taking too long (>10s).

Debug Steps:

1. Open Phoenix trace waterfall 2. Identify longest spans 3. Check component breakdown

Common Causes:

Latency Optimization:

# Identify: Phoenix shows LLMRerank taking 3s

# Before
node_postprocessors=[
    LLMRerank(top_n=5)  # Slow
]

# After: two-stage
node_postprocessors=[
    SentenceTransformerRerank(top_n=10),  # Fast filter
    LLMRerank(top_n=3)  # Precise on smaller set
]

Routing Problems

Symptom: Queries going to wrong engine in RouterQueryEngine.

Debug Steps:

1. Open Phoenix trace 2. Find "Router" span 3. Check selector decision

Common Causes:

Fix Pattern:

# Before: vague
QueryEngineTool.from_defaults(
    query_engine=engine,
    description="Handles questions"  # Too generic
)

# After: specific
QueryEngineTool.from_defaults(
    query_engine=engine,
    description=(
        "Answers questions about product specifications, "
        "dimensions, materials, and technical details. "
        "Use for 'how big', 'what material', 'specs' queries."
    )
)

---

Evaluators

LLM-as-judge evaluation of agent responses.

FaithfulnessEvaluator

Checks if answer is derived solely from retrieved context (anti-hallucination).

from llama_index.core.evaluation import FaithfulnessEvaluator

evaluator = FaithfulnessEvaluator()

result = evaluator.evaluate_response(
    query="What is the return policy?",
    response=response,  # Agent response object
)

print(f"Faithful: {result.passing}")  # True/False
print(f"Score: {result.score}")       # 0.0-1.0
print(f"Feedback: {result.feedback}") # Explanation

Use Case: Detect when agent invents information not in documents.

RelevancyEvaluator

Checks if answer actually addresses the user's query.

from llama_index.core.evaluation import RelevancyEvaluator

evaluator = RelevancyEvaluator()

result = evaluator.evaluate_response(
    query="How do I reset my password?",
    response=response,
)

print(f"Relevant: {result.passing}")

Use Case: Detect tangential or off-topic responses.

CorrectnessEvaluator

Grades response against a reference "gold standard" answer.

from llama_index.core.evaluation import CorrectnessEvaluator

evaluator = CorrectnessEvaluator()

result = evaluator.evaluate(
    query="What is 2+2?",
    response="The answer is 4.",
    reference="4",  # Gold standard
)

print(f"Correct: {result.passing}")
print(f"Score: {result.score}")  # 1-5 scale

Use Case: Regression testing with known Q&A pairs.

---

Evaluation Pipeline

Systematic evaluation across test dataset:

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    BatchEvalRunner,
)

# Setup evaluators
faithfulness = FaithfulnessEvaluator()
relevancy = RelevancyEvaluator()

# Test queries
test_queries = [
    "What is the return policy?",
    "How do I contact support?",
    "What are the shipping options?",
]

# Run batch evaluation
runner = BatchEvalRunner(
    evaluators={
        "faithfulness": faithfulness,
        "relevancy": relevancy,
    },
    workers=4,
)

# Generate responses and evaluate
responses = [query_engine.query(q) for q in test_queries]
eval_results = await runner.aevaluate_responses(
    queries=test_queries,
    responses=responses,
)

# Aggregate results
for metric, results in eval_results.items():
    scores = [r.score for r in results]
    print(f"{metric}: {sum(scores)/len(scores):.2f} avg")

---

Production Monitoring

Key Metrics

Continuous Evaluation

import asyncio
from datetime import datetime

async def monitor_loop(query_engine, evaluator, sample_queries):
    while True:
        for query in sample_queries:
            response = query_engine.query(query)
            result = evaluator.evaluate_response(query=query, response=response)
            
            # Log to monitoring system
            log_metric("faithfulness", result.score, timestamp=datetime.now())
            
            if not result.passing:
                alert(f"Faithfulness failure: {query}")
        
        await asyncio.sleep(300)  # Every 5 minutes

Phoenix + Grafana

Export Phoenix metrics to Grafana for dashboards:

# Phoenix exports OpenTelemetry format
# Configure OTLP exporter to send to Grafana Cloud
import os
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://otlp.grafana.com"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = "Authorization=Basic ..."

---

See Also

• ../SKILL.md — Return to main skill overview
• ingestion.md — Debug chunking issues identified in traces
• context-rag.md — Debug routing decisions
• orchestration.md — Trace workflow steps and tool calls

Agentic Orchestration

ReAct agents, function calling, event-driven Workflows, and multi-agent patterns.

Contents

• ReAct Agent Pattern — Thought→Action→Observation loop
• Function Calling vs Text Parsing — Modern structured approach
• Parallel Tool Calls — Concurrent execution
• Tools — Wrapping functions for agent use
• FunctionTool — Custom function wrapper + docstring best practices
• ToolSpecs — Pre-packaged tool bundles from LlamaHub
• Workflows — Event-driven finite state machine (v0.10+)
• Core Components — Workflow, Event, @step, Context
• Events — Custom event definitions
• Steps — Async methods with type-hint routing
• Context — Shared state across steps
• Basic Workflow — Minimal working example
• Advanced Patterns — Beyond simple loops
• Branching — Conditional routing
• Loops — Iterative refinement
• Human-in-the-Loop — Approval flows
• Concierge Multi-Agent — Specialized sub-agents
• Pattern Comparison — Decision guide
• Complete Examples — Production-ready code
• See Also — Related references

---

ReAct Agent Pattern

Default architecture for tool-using agents. Reasoning and Acting in a loop.

Mechanism

User Query → [Thought → Action → Observation] → ... → Final Answer

1. Thought: Agent analyzes input, decides if tool needed 2. Action: Agent generates tool call 3. Observation: Tool executes, returns result 4. Repeat: Agent analyzes observation, continues or answers

Function Calling vs Text Parsing

Text Parsing (Legacy)

Agent outputs text like Action: search("query"), parsed via regex.

from llama_index.core.agent import ReActAgent

agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    verbose=True,
)

Drawbacks: Parsing errors, prompt engineering required.

Function Calling (Modern)

LLM outputs structured JSON matching tool schema. Native support in GPT-4, Mistral.

from llama_index.core.agent import FunctionCallingAgent

agent = FunctionCallingAgent.from_tools(
    tools,
    llm=llm,  # Must support function calling
    verbose=True,
)

Benefits: No parsing errors, reliable tool invocation.

Parallel Tool Calls

LLM can request multiple tools in single turn. System executes in parallel threads.

# Query: "Get weather for Tokyo and New York"
# LLM emits two tool calls simultaneously:
#   - get_weather(city="Tokyo")
#   - get_weather(city="New York")
# Both execute in parallel, results aggregated

Enabling Parallel Execution

from llama_index.core.agent import FunctionCallingAgent

# Parallel calls enabled by default
agent = FunctionCallingAgent.from_tools(
    tools,
    llm=llm,
    allow_parallel_tool_calls=True,  # Default: True
    verbose=True,
)

# Query triggers parallel execution
response = agent.chat("Compare weather in Tokyo, London, and Sydney")
# Executes 3 get_weather calls in parallel

Controlling Parallelism

# Disable if tools have dependencies or side effects
agent = FunctionCallingAgent.from_tools(
    tools,
    llm=llm,
    allow_parallel_tool_calls=False,  # Sequential execution
)

# Or control at tool level with dependencies
def get_user_orders(user_id: str) -> str:
    """Get orders. Requires user_id from get_current_user first."""
    ...

Parallel in Workflows

from llama_index.core.workflow import Workflow, step, Event

class ParallelSearchEvent(Event):
    queries: list[str]

class SearchResultEvent(Event):
    query: str
    results: list[str]

class ParallelWorkflow(Workflow):

    @step
    async def parallel_search(self, ev: ParallelSearchEvent) -> list[SearchResultEvent]:
        # Emit multiple events - all processed in parallel
        events = []
        for query in ev.queries:
            events.append(SearchResultEvent(query=query, results=self.search(query)))
        return events  # Return list = parallel emission

---

Tools

FunctionTool

Wraps any Python function as an agent tool.

from llama_index.core.tools import FunctionTool

def search_database(query: str, limit: int = 10) -> str:
    """
    Search the product database.
    
    Args:
        query: Search terms
        limit: Maximum results to return
    
    Returns:
        JSON string of matching products
    """
    # Implementation
    results = db.search(query, limit=limit)
    return json.dumps(results)

tool = FunctionTool.from_defaults(fn=search_database)

Critical: The docstring becomes the tool description. Write verbose, clear docstrings.

Docstring Best Practices

def calculate_shipping(
    weight_kg: float,
    destination: str,
    express: bool = False
) -> str:
    """
    Calculate shipping cost for a package.
    
    Use this tool when the user asks about shipping costs,
    delivery prices, or postage for sending items.
    
    Args:
        weight_kg: Package weight in kilograms
        destination: Destination country code (e.g., "US", "UK")
        express: Whether to use express shipping (2-day vs 7-day)
    
    Returns:
        JSON with cost breakdown and estimated delivery date
    """
    ...

ToolSpecs

Pre-packaged tool bundles from LlamaHub.

from llama_index.tools.google import GmailToolSpec

# Load entire Gmail capability
gmail_spec = GmailToolSpec()
tools = gmail_spec.to_tool_list()

# Tools include: search_messages, create_draft, send_email, etc.
agent = FunctionCallingAgent.from_tools(tools, llm=llm)

Available ToolSpecs:

• GmailToolSpec — Email operations
• GoogleCalendarToolSpec — Calendar management
• SlackToolSpec — Slack messaging
• NotionToolSpec — Notion pages/databases
• WikipediaToolSpec — Wikipedia search

Install: pip install llama-index-tools-google

---

Workflows

Event-driven finite state machine. Introduced in LlamaIndex v0.10.

Why Workflows?

ReAct limitations:

• Linear loop only
• No branching or cycles
• Hard to suspend/resume
• Complex state management

Workflows enable:

• Branching logic
• Cyclic flows (loops)
• Human-in-the-loop
• Multi-agent coordination
• Clean state management

Core Components

Events

Define custom events for data passing:

from llama_index.core.workflow import Event

class QueryEvent(Event):
    """Carries the user query for processing."""
    query: str

class RetrievalEvent(Event):
    """Carries query with retrieved context."""
    query: str
    context: list[str]

class ClassificationEvent(Event):
    """Carries query classification result."""
    query: str
    category: str  # "factual", "analytical", "conversational"

Steps

Steps listen for specific event types (via input type hint):

from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent

class MyWorkflow(Workflow):
    
    @step
    async def classify(self, ev: StartEvent) -> QueryEvent:
        """First step: receives StartEvent."""
        query = ev.get("query")
        return QueryEvent(query=query)
    
    @step
    async def retrieve(self, ev: QueryEvent) -> RetrievalEvent:
        """Triggered by QueryEvent."""
        context = self.retriever.retrieve(ev.query)
        return RetrievalEvent(query=ev.query, context=context)
    
    @step
    async def respond(self, ev: RetrievalEvent) -> StopEvent:
        """Final step: emits StopEvent to end workflow."""
        response = self.llm.complete(f"Context: {ev.context}\nQuery: {ev.query}")
        return StopEvent(result=str(response))

Context

Shared state accessible from all steps:

from llama_index.core.workflow import Context

class StatefulWorkflow(Workflow):
    
    @step
    async def step_one(self, ctx: Context, ev: StartEvent) -> NextEvent:
        # Store in context
        await ctx.set("user_id", ev.get("user_id"))
        await ctx.set("history", [])
        return NextEvent()
    
    @step
    async def step_two(self, ctx: Context, ev: NextEvent) -> StopEvent:
        # Retrieve from context
        user_id = await ctx.get("user_id")
        history = await ctx.get("history")
        return StopEvent(result=f"User {user_id}")

Basic Workflow

from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Event

class QueryEvent(Event):
    query: str

class SimpleAgent(Workflow):
    def __init__(self, query_engine, **kwargs):
        super().__init__(**kwargs)
        self.query_engine = query_engine
    
    @step
    async def route(self, ev: StartEvent) -> QueryEvent:
        return QueryEvent(query=ev.get("query"))
    
    @step
    async def answer(self, ev: QueryEvent) -> StopEvent:
        response = self.query_engine.query(ev.query)
        return StopEvent(result=str(response))

# Run
async def main():
    agent = SimpleAgent(query_engine=engine, timeout=60, verbose=True)
    result = await agent.run(query="What is the capital of France?")
    print(result)

---

Advanced Patterns

Branching

One step emits different events based on conditions:

class ClassifyEvent(Event):
    query: str
    category: str

class TechnicalEvent(Event):
    query: str

class GeneralEvent(Event):
    query: str

class BranchingWorkflow(Workflow):
    
    @step
    async def classify(self, ev: StartEvent) -> TechnicalEvent | GeneralEvent:
        query = ev.get("query")
        
        # Classification logic
        if "code" in query.lower() or "error" in query.lower():
            return TechnicalEvent(query=query)
        else:
            return GeneralEvent(query=query)
    
    @step
    async def handle_technical(self, ev: TechnicalEvent) -> StopEvent:
        response = self.technical_engine.query(ev.query)
        return StopEvent(result=str(response))
    
    @step
    async def handle_general(self, ev: GeneralEvent) -> StopEvent:
        response = self.general_engine.query(ev.query)
        return StopEvent(result=str(response))

Loops

Step emits event that triggers itself or earlier step:

class RefineEvent(Event):
    query: str
    response: str
    iterations: int

class LoopingWorkflow(Workflow):
    
    @step
    async def generate(self, ev: StartEvent | RefineEvent) -> RefineEvent | StopEvent:
        if isinstance(ev, StartEvent):
            query = ev.get("query")
            iterations = 0
        else:
            query = ev.query
            iterations = ev.iterations
        
        response = self.llm.complete(query)
        
        # Check quality
        if self.is_good_enough(response) or iterations >= 3:
            return StopEvent(result=str(response))
        else:
            return RefineEvent(
                query=f"Improve this: {response}",
                response=str(response),
                iterations=iterations + 1
            )

Human-in-the-Loop

Suspend workflow until external input using built-in events.

Built-in Events

Basic Pattern

from llama_index.core.workflow import InputRequiredEvent, HumanResponseEvent

class ApprovalWorkflow(Workflow):

    @step
    async def propose(self, ev: StartEvent) -> InputRequiredEvent:
        proposal = self.generate_proposal(ev.get("task"))
        # Suspend and wait for human
        return InputRequiredEvent(
            prefix="Please approve this proposal:",
            payload=proposal
        )

    @step
    async def execute(self, ev: HumanResponseEvent) -> StopEvent:
        if ev.response.lower() == "approved":
            result = self.execute_proposal()
            return StopEvent(result=result)
        else:
            return StopEvent(result="Proposal rejected")

Running with Human Input

async def run_with_approval():
    workflow = ApprovalWorkflow(timeout=300)  # Longer timeout for human

    # Start workflow - will suspend at InputRequiredEvent
    handler = workflow.run(task="Deploy to production")

    async for event in handler.stream_events():
        if isinstance(event, InputRequiredEvent):
            print(f"{event.prefix}")
            print(f"Context: {event.payload}")

            # Get human input (from UI, CLI, etc.)
            user_input = input("Your response: ")

            # Resume workflow with human response
            handler.ctx.send_event(HumanResponseEvent(response=user_input))

    result = await handler
    return result

Multi-Step Approval

class MultiApprovalWorkflow(Workflow):

    @step
    async def review_step(self, ev: StartEvent | HumanResponseEvent) -> InputRequiredEvent | StopEvent:
        ctx = self.ctx

        if isinstance(ev, StartEvent):
            # First review
            await ctx.set("reviews", [])
            await ctx.set("current_reviewer", 0)
            reviewers = ["Manager", "Security", "Legal"]
            await ctx.set("reviewers", reviewers)
        else:
            # Process previous response
            reviews = await ctx.get("reviews")
            reviews.append(ev.response)
            await ctx.set("reviews", reviews)

        reviewers = await ctx.get("reviewers")
        current = await ctx.get("current_reviewer")

        if current < len(reviewers):
            await ctx.set("current_reviewer", current + 1)
            return InputRequiredEvent(
                prefix=f"Awaiting {reviewers[current]} approval:",
                payload=await ctx.get("reviews")
            )
        else:
            return StopEvent(result={"approvals": await ctx.get("reviews")})

Concierge Multi-Agent

Entry agent routes to specialized sub-agents:

class TravelEvent(Event):
    query: str

class SupportEvent(Event):
    query: str

class ConciergeWorkflow(Workflow):
    
    @step
    async def concierge(self, ev: StartEvent) -> TravelEvent | SupportEvent | StopEvent:
        query = ev.get("query")
        intent = self.classify_intent(query)
        
        if intent == "travel":
            return TravelEvent(query=query)
        elif intent == "support":
            return SupportEvent(query=query)
        else:
            # Handle directly
            return StopEvent(result=self.general_response(query))
    
    @step
    async def travel_agent(self, ev: TravelEvent) -> StopEvent:
        # Specialized travel handling
        result = self.travel_engine.query(ev.query)
        return StopEvent(result=str(result))
    
    @step
    async def support_agent(self, ev: SupportEvent) -> StopEvent:
        # Specialized support handling
        result = self.support_engine.query(ev.query)
        return StopEvent(result=str(result))

---

Pattern Comparison

Decision Guide

Simple tool usage?
├─ Yes → FunctionCallingAgent (ReAct)
│
└─ No, need:
    ├─ Conditional logic → Branching Workflow
    ├─ Iteration/refinement → Looping Workflow
    ├─ Human approval → Human-in-the-Loop Workflow
    ├─ Multiple specialists → Concierge Workflow
    └─ All of the above → Compose patterns in single Workflow

---

Complete Examples

Production Agent with Tools and Workflow

import asyncio
from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Event
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

# Define tools
def search_docs(query: str) -> str:
    """Search internal documentation."""
    return engine.query(query)

def search_web(query: str) -> str:
    """Search the web for current information."""
    return web_search(query)

tools = [
    FunctionTool.from_defaults(fn=search_docs),
    FunctionTool.from_defaults(fn=search_web),
]

# Workflow with tool selection
class ToolEvent(Event):
    query: str
    tool_name: str

class SmartAgent(Workflow):
    def __init__(self, tools, llm, **kwargs):
        super().__init__(**kwargs)
        self.tools = {t.metadata.name: t for t in tools}
        self.llm = llm
    
    @step
    async def select_tool(self, ev: StartEvent) -> ToolEvent | StopEvent:
        query = ev.get("query")
        # LLM selects tool
        selection = self.llm.complete(
            f"Select tool for: {query}\nTools: {list(self.tools.keys())}"
        )
        tool_name = str(selection).strip()
        
        if tool_name in self.tools:
            return ToolEvent(query=query, tool_name=tool_name)
        else:
            return StopEvent(result="No suitable tool found")
    
    @step
    async def execute_tool(self, ev: ToolEvent) -> StopEvent:
        tool = self.tools[ev.tool_name]
        result = tool.call(ev.query)
        return StopEvent(result=result)

# Run
async def main():
    agent = SmartAgent(tools=tools, llm=OpenAI(), timeout=60)
    result = await agent.run(query="Find our refund policy")
    print(result)

asyncio.run(main())

---

See Also

• ../SKILL.md — Return to main skill overview
• context-rag.md — Query engines to use as tools
• property-graphs.md — Graph retrieval in agent tools
• observability.md — Trace agent steps and tool calls

Property Graphs

Knowledge graph construction, storage backends, extraction strategies, and retrieval modes.

Contents

• PropertyGraphIndex Overview
• Graph Storage Backends
• SimplePropertyGraphStore
• Neo4jPropertyGraphStore
• Other Backends
• Knowledge Extraction
• ImplicitPathExtractor
• SimpleLLMPathExtractor
• SchemaLLMPathExtractor
• Extractor Comparison
• Retrieval Strategies
• VectorContextRetriever
• TextToCypherRetriever
• CypherTemplateRetriever
• Retriever Comparison
• Complete Example
• Schema Design Guidelines
• See Also

---

PropertyGraphIndex Overview

Hybrid index combining vector embeddings with labeled property graph structure.

Graph Model

• Nodes: Entities (Person, Company) or text chunks
• Edges: Relationships (FOUNDED, WORKS_AT, MENTIONS)
• Properties: Metadata on nodes and edges (dates, scores)

Key Capability

Vector embeddings attach to graph nodes, enabling "Vector-to-Graph" retrieval: 1. Semantic search finds relevant nodes 2. Graph traversal discovers connected facts 3. Combined context returned to LLM

Basic Construction

from llama_index.core import PropertyGraphIndex, SimpleDirectoryReader
from llama_index.embeddings.openai import OpenAIEmbedding

docs = SimpleDirectoryReader("./data").load_data()

index = PropertyGraphIndex.from_documents(
    docs,
    embed_model=OpenAIEmbedding(),
    show_progress=True,
)

---

Graph Storage Backends

SimplePropertyGraphStore

In-memory or file-based storage. No external dependencies.

from llama_index.core.graph_stores import SimplePropertyGraphStore

graph_store = SimplePropertyGraphStore()

index = PropertyGraphIndex.from_documents(
    docs,
    property_graph_store=graph_store,
    embed_model=embed_model,
)

# Persist to disk
index.storage_context.persist(persist_dir="./storage")

Characteristics:

• Serializes to JSON/Dict
• No native Cypher support
• Supports networkx visualization
• Best for: Prototyping, small datasets (<10K nodes)

Neo4jPropertyGraphStore

Production-grade graph database integration.

from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore

graph_store = Neo4jPropertyGraphStore(
    username="neo4j",
    password="password",
    url="bolt://localhost:7687",
    database="neo4j",
)

index = PropertyGraphIndex.from_documents(
    docs,
    property_graph_store=graph_store,
    embed_model=embed_model,
)

Characteristics:

• Native Cypher query execution
• ACID transactions
• Built-in vector index support
• Best for: Production, large scale, complex queries

Prerequisites:

pip install llama-index-graph-stores-neo4j
# Neo4j server running (Docker or cloud)

Other Backends

Install pattern: pip install llama-index-graph-stores-{backend}

---

Knowledge Extraction

Extractors analyze text and output graph triples (Subject, Predicate, Object).

ImplicitPathExtractor

Extracts document structure without LLM. Zero cost.

from llama_index.core.indices.property_graph import ImplicitPathExtractor

extractor = ImplicitPathExtractor()

Generated Relationships:

• NEXT / PREVIOUS — Sequential chunks
• SOURCE — Chunk to source document
• PARENT — Chunk to parent section

When to Use:

• Document navigation ("read next section")
• Structural queries ("show parent document")
• Baseline graph with no LLM cost

SimpleLLMPathExtractor

LLM-powered extraction with dynamic ontology.

from llama_index.core.indices.property_graph import SimpleLLMPathExtractor
from llama_index.llms.openai import OpenAI

extractor = SimpleLLMPathExtractor(
    llm=OpenAI(model="gpt-4-turbo"),
    max_paths_per_chunk=10,  # Limit triples per node
)

Example Output:

Text: "Apple released the Vision Pro headset in 2024"
→ (Apple)--[RELEASED]-->(Vision Pro)
→ (Vision Pro)--[TYPE]-->(Headset)
→ (Vision Pro)--[RELEASED_IN]-->(2024)

Custom Prompt:

extractor = SimpleLLMPathExtractor(
    llm=llm,
    extract_prompt=(
        "Extract entities and relationships from the text. "
        "Focus on people, organizations, and their interactions. "
        "Output as (entity1)--[RELATIONSHIP]-->(entity2)"
    ),
)

When to Use:

• Discovery and exploration
• Broad knowledge mapping
• Unknown or variable schemas

SchemaLLMPathExtractor

LLM extraction constrained to predefined ontology.

from llama_index.core.indices.property_graph import SchemaLLMPathExtractor

# Define allowed entities and relationships
schema = {
    "PERSON": ["WORKS_AT", "FOUNDED", "INVESTED_IN"],
    "COMPANY": ["LOCATED_IN", "ACQUIRED", "PRODUCES"],
    "PRODUCT": ["RELEASED_BY", "COMPETES_WITH"],
}

extractor = SchemaLLMPathExtractor(
    llm=llm,
    possible_entities=list(schema.keys()),
    possible_relations=list(set(r for rels in schema.values() for r in rels)),
    strict=True,  # Reject non-conforming extractions
)

When to Use:

• Regulated domains (finance, healthcare)
• Consistent querying requirements
• Preventing ontology drift (WORKS_AT vs EMPLOYED_BY)

---

Extractor Comparison

Combining Extractors

Use multiple extractors for comprehensive graphs:

index = PropertyGraphIndex.from_documents(
    docs,
    kg_extractors=[
        ImplicitPathExtractor(),  # Structure (free)
        SimpleLLMPathExtractor(max_paths_per_chunk=5),  # Concepts
    ],
    embed_model=embed_model,
)

---

Retrieval Strategies

VectorContextRetriever

Most robust. Vector search + graph traversal.

from llama_index.core.indices.property_graph import VectorContextRetriever

retriever = VectorContextRetriever(
    index.property_graph_store,
    embed_model=embed_model,
    include_text=True,   # Include node text in context
    path_depth=2,        # Traversal hops from matched nodes
    similarity_top_k=5,  # Initial vector matches
)

# Or via index
retriever = index.as_retriever(
    include_text=True,
    similarity_top_k=5,
)

Mechanism: 1. Vector search finds semantically similar nodes 2. Graph traversal collects connected nodes (up to path_depth) 3. Combined context returned

When to Use:

• General-purpose retrieval
• Robustness priority (no LLM code generation)
• Unknown query patterns

TextToCypherRetriever

LLM generates Cypher queries from natural language.

from llama_index.core.indices.property_graph import TextToCypherRetriever

retriever = TextToCypherRetriever(
    index.property_graph_store,
    llm=llm,
)

# Query
nodes = retriever.retrieve("How many employees does Apple have?")
# LLM generates: MATCH (c:Company {name: 'Apple'})-[:EMPLOYS]->(e) RETURN count(e)

When to Use:

• Complex aggregations (COUNT, SUM, AVG)
• Filtering with conditions
• Trusted environments only

Risks:

• Cypher syntax errors
• Injection vulnerabilities (sandbox required)
• Non-deterministic results

CypherTemplateRetriever

Parameterized Cypher for safety + flexibility.

from llama_index.core.indices.property_graph import CypherTemplateRetriever

retriever = CypherTemplateRetriever(
    index.property_graph_store,
    llm=llm,
    cypher_template=(
        "MATCH (p:Person {name: $name})-[:WROTE]->(b:Book) "
        "RETURN b.title AS title, b.year AS year"
    ),
    template_params=["name"],
)

# Query: "What books did George Orwell write?"
# LLM extracts: name="George Orwell"
# Executes template with parameter

When to Use:

• Known query patterns
• Security-sensitive environments
• 100% syntactic correctness required

---

Retriever Comparison

---

Complete Example

Build PropertyGraphIndex with Neo4j, schema extraction, and hybrid retrieval:

from llama_index.core import PropertyGraphIndex, SimpleDirectoryReader
from llama_index.core.indices.property_graph import (
    ImplicitPathExtractor,
    SchemaLLMPathExtractor,
)
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI

# 1. Setup stores
graph_store = Neo4jPropertyGraphStore(
    username="neo4j",
    password="password",
    url="bolt://localhost:7687",
)

embed_model = OpenAIEmbedding(model_name="text-embedding-3-small")
llm = OpenAI(model="gpt-4-turbo")

# 2. Define schema
schema = {
    "PERSON": ["WORKS_AT", "FOUNDED"],
    "COMPANY": ["LOCATED_IN", "PRODUCES"],
    "PRODUCT": ["RELEASED_BY"],
}

# 3. Build index
docs = SimpleDirectoryReader("./data").load_data()

index = PropertyGraphIndex.from_documents(
    docs,
    property_graph_store=graph_store,
    embed_model=embed_model,
    kg_extractors=[
        ImplicitPathExtractor(),
        SchemaLLMPathExtractor(
            llm=llm,
            possible_entities=list(schema.keys()),
            possible_relations=["WORKS_AT", "FOUNDED", "LOCATED_IN", "PRODUCES", "RELEASED_BY"],
            strict=True,
        ),
    ],
    show_progress=True,
)

# 4. Query with VectorContext
retriever = index.as_retriever(include_text=True, similarity_top_k=5)
nodes = retriever.retrieve("Who founded the company?")

---

Schema Design Guidelines

1. Use consistent relationship names: Pick WORKS_AT or EMPLOYED_BY, not both 2. Limit entity types: 5-10 types for manageable graphs 3. Include inverse relationships: WORKS_AT ↔ EMPLOYS for bidirectional traversal 4. Add temporal properties: {since: "2020"} on edges for time-based queries 5. Test with sample data: Validate schema captures important relationships

---

See Also

• ../SKILL.md — Return to main skill overview
• ingestion.md — Semantic chunking before graph construction
• context-rag.md — Routing queries to graph vs. vector stores
• observability.md — Debug graph retrieval issues

Retrieval Strategies

Advanced retrieval approaches: BM25 keyword search, hybrid fusion, and search mode selection.

Contents

• BM25Retriever
• Core Concept
• Basic Usage
• Persistence
• Hybrid Search
• Why Hybrid?
• Fusion Strategies
• Alpha Weighting
• Implementation Pattern
• Search Mode Selection
• Performance Considerations
• Complete Examples
• See Also

---

BM25Retriever

Sparse keyword-based retrieval using the BM25 algorithm. Complements vector search by finding exact term matches that semantic embeddings may miss.

Core Concept

BM25 (Best Matching 25) ranks documents by term frequency with diminishing returns:

• Term Frequency (TF): Documents mentioning query terms more often rank higher
• Inverse Document Frequency (IDF): Rare terms get higher weight than common terms
• Length Normalization: Long documents don't automatically rank higher

When BM25 Excels:

• Exact string matching (function names, error codes, IDs)
• Technical documentation with specific terminology
• Queries containing unique identifiers
• When semantic similarity misses literal matches

Basic Usage

from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.core.schema import TextNode

# Create nodes (typically from ingestion pipeline)
nodes = [
    TextNode(text="The RecursiveCharacterTextSplitter handles chunking...", id_="node1"),
    TextNode(text="Error code ERROR_CODE_404 indicates missing resource...", id_="node2"),
    TextNode(text="Vector embeddings capture semantic meaning...", id_="node3"),
]

# Build BM25 index
retriever = BM25Retriever.from_defaults(
    nodes=nodes,
    similarity_top_k=5,
)

# Query
results = retriever.retrieve("ERROR_CODE_404")
# Returns node2 as top result (exact term match)

Prerequisites:

pip install llama-index-retrievers-bm25

Persistence

Save and load BM25 index for production use:

from pathlib import Path

# Save index
persist_path = Path("./bm25_index")
persist_path.mkdir(exist_ok=True)
retriever.persist(str(persist_path))

# Load index (later)
retriever = BM25Retriever.from_persist_dir(str(persist_path))

Async Support

# Async retrieval
results = await retriever.aretrieve("search query")

---

Hybrid Search

Combines vector semantic search with BM25 keyword search for robust retrieval.

Why Hybrid?

Key Insight: Vector search understands meaning; BM25 finds exact terms. Together, they cover more ground.

Fusion Strategies

Reciprocal Rank Fusion (RRF)

Combines rankings without score normalization. Position-based, robust to score scale differences.

def rrf_score(rank: int, k: int = 60) -> float:
    """RRF formula: 1 / (k + rank)"""
    return 1.0 / (k + rank)

# Document in position 1 from both retrievers:
# RRF = 1/(60+1) + 1/(60+1) = 0.0328

Characteristics:

• Simple and robust
• No score normalization needed
• Works with any number of retrievers

Relative Score Fusion (RSF)

Normalizes scores to 0-1 range then combines with weighting.

# Normalize scores
max_vector_score = max(r.score for r in vector_results) or 1.0
max_bm25_score = max(r.score for r in bm25_results) or 1.0

for result in combined_results:
    vector_normalized = result.vector_score / max_vector_score
    bm25_normalized = result.bm25_score / max_bm25_score
    result.score = alpha * vector_normalized + (1 - alpha) * bm25_normalized

Characteristics:

• Tunable via alpha parameter
• Requires score normalization
• More control over strategy balance

Alpha Weighting

The alpha parameter controls vector vs keyword balance:

Tuning Guidelines:

• Conceptual queries (how, why, explain): Higher alpha (0.7-0.9)
• Technical queries (function names, error codes): Lower alpha (0.3-0.5)
• Mixed queries: Default alpha (0.5)

Implementation Pattern

Production-ready hybrid search combining vector and BM25:

from typing import Optional
from llama_index.core.schema import NodeWithScore, QueryBundle
from llama_index.retrievers.bm25 import BM25Retriever


class HybridRetriever:
    """
    Combines vector and BM25 retrieval with configurable fusion.
    """

    def __init__(
        self,
        vector_retriever,
        bm25_retriever: BM25Retriever,
        alpha: float = 0.5,
    ):
        self.vector_retriever = vector_retriever
        self.bm25_retriever = bm25_retriever
        self.alpha = alpha  # 1.0 = pure vector, 0.0 = pure BM25

    async def aretrieve(
        self,
        query: str,
        top_k: int = 5,
    ) -> list[NodeWithScore]:
        """Execute hybrid search with score fusion."""

        # 1. Run both retrievers
        vector_results = await self.vector_retriever.aretrieve(query)
        bm25_results = await self.bm25_retriever.aretrieve(query)

        # 2. Normalize scores
        max_vector = max((r.score for r in vector_results), default=1.0) or 1.0
        max_bm25 = max((r.score for r in bm25_results), default=1.0) or 1.0

        # 3. Combine results
        combined: dict[str, dict] = {}

        for result in vector_results:
            node_id = result.node.node_id
            combined[node_id] = {
                "node": result.node,
                "vector_score": result.score / max_vector,
                "bm25_score": 0.0,
            }

        for result in bm25_results:
            node_id = result.node.node_id
            bm25_normalized = result.score / max_bm25

            if node_id in combined:
                combined[node_id]["bm25_score"] = bm25_normalized
            else:
                combined[node_id] = {
                    "node": result.node,
                    "vector_score": 0.0,
                    "bm25_score": bm25_normalized,
                }

        # 4. Calculate final scores
        fused_results = []
        for data in combined.values():
            final_score = (
                self.alpha * data["vector_score"] +
                (1 - self.alpha) * data["bm25_score"]
            )
            fused_results.append(
                NodeWithScore(node=data["node"], score=final_score)
            )

        # 5. Sort and return top_k
        fused_results.sort(key=lambda x: x.score, reverse=True)
        return fused_results[:top_k]

---

Search Mode Selection

Pattern for supporting multiple search modes via API or CLI:

from enum import Enum


class QueryMode(str, Enum):
    VECTOR = "vector"   # Pure semantic search
    BM25 = "bm25"       # Pure keyword search
    HYBRID = "hybrid"   # Combined (default)


class QueryService:
    """Executes queries based on selected mode."""

    def __init__(
        self,
        vector_retriever,
        bm25_retriever: BM25Retriever,
    ):
        self.vector_retriever = vector_retriever
        self.bm25_retriever = bm25_retriever
        self.hybrid_retriever = HybridRetriever(
            vector_retriever, bm25_retriever
        )

    async def query(
        self,
        text: str,
        mode: QueryMode = QueryMode.HYBRID,
        alpha: float = 0.5,
        top_k: int = 5,
    ) -> list[NodeWithScore]:
        """Execute query with specified mode."""

        if mode == QueryMode.VECTOR:
            return await self.vector_retriever.aretrieve(text)
        elif mode == QueryMode.BM25:
            return await self.bm25_retriever.aretrieve(text)
        else:  # HYBRID
            self.hybrid_retriever.alpha = alpha
            return await self.hybrid_retriever.aretrieve(text, top_k=top_k)

---

Performance Considerations

BM25 Index Size

BM25 adds sparse index storage:

• Typically 20-50% of vector index size
• Stores term frequencies, document lengths
• Fast to build (no embeddings required)

Query Latency

Parallel Execution

Run vector and BM25 searches concurrently:

import asyncio

async def hybrid_search_parallel(self, query: str):
    """Execute both searches in parallel."""
    vector_task = asyncio.create_task(
        self.vector_retriever.aretrieve(query)
    )
    bm25_task = asyncio.create_task(
        self.bm25_retriever.aretrieve(query)
    )

    vector_results, bm25_results = await asyncio.gather(
        vector_task, bm25_task
    )
    return self._fuse_results(vector_results, bm25_results)

---

Complete Examples

Production Hybrid Search Service

import asyncio
from pathlib import Path

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.vector_stores.chroma import ChromaVectorStore

import chromadb


class HybridSearchService:
    """Production-ready hybrid search with persistence."""

    def __init__(self, persist_dir: str = "./search_index"):
        self.persist_dir = Path(persist_dir)
        self.persist_dir.mkdir(exist_ok=True)

        self.embed_model = OpenAIEmbedding()
        self._vector_index = None
        self._bm25_retriever = None

    async def index_documents(self, documents: list):
        """Build both vector and BM25 indexes."""

        # 1. Create nodes
        pipeline = IngestionPipeline(
            transformations=[
                SentenceSplitter(chunk_size=1024),
                self.embed_model,
            ]
        )
        nodes = pipeline.run(documents=documents)

        # 2. Build vector index
        chroma_client = chromadb.PersistentClient(
            path=str(self.persist_dir / "chroma")
        )
        collection = chroma_client.get_or_create_collection("documents")
        vector_store = ChromaVectorStore(chroma_collection=collection)

        storage_context = StorageContext.from_defaults(
            vector_store=vector_store
        )
        self._vector_index = VectorStoreIndex(
            nodes=nodes,
            storage_context=storage_context,
            embed_model=self.embed_model,
        )

        # 3. Build BM25 index
        self._bm25_retriever = BM25Retriever.from_defaults(nodes=nodes)
        self._bm25_retriever.persist(str(self.persist_dir / "bm25"))

        return len(nodes)

    def load(self):
        """Load existing indexes."""
        # Load vector index
        chroma_client = chromadb.PersistentClient(
            path=str(self.persist_dir / "chroma")
        )
        collection = chroma_client.get_collection("documents")
        vector_store = ChromaVectorStore(chroma_collection=collection)
        self._vector_index = VectorStoreIndex.from_vector_store(
            vector_store, embed_model=self.embed_model
        )

        # Load BM25 index
        self._bm25_retriever = BM25Retriever.from_persist_dir(
            str(self.persist_dir / "bm25")
        )

    async def search(
        self,
        query: str,
        mode: str = "hybrid",
        alpha: float = 0.5,
        top_k: int = 5,
    ):
        """Execute search with specified mode."""
        vector_retriever = self._vector_index.as_retriever(
            similarity_top_k=top_k
        )

        if mode == "vector":
            return await vector_retriever.aretrieve(query)
        elif mode == "bm25":
            return await self._bm25_retriever.aretrieve(query)
        else:
            # Hybrid: parallel execution
            vector_task = asyncio.create_task(
                vector_retriever.aretrieve(query)
            )
            bm25_task = asyncio.create_task(
                self._bm25_retriever.aretrieve(query)
            )

            vector_results, bm25_results = await asyncio.gather(
                vector_task, bm25_task
            )

            return self._fuse_results(
                vector_results, bm25_results, alpha, top_k
            )

    def _fuse_results(
        self,
        vector_results,
        bm25_results,
        alpha: float,
        top_k: int,
    ):
        """Fuse results with alpha weighting."""
        # Normalize
        max_v = max((r.score for r in vector_results), default=1.0) or 1.0
        max_b = max((r.score for r in bm25_results), default=1.0) or 1.0

        combined = {}

        for r in vector_results:
            combined[r.node.node_id] = {
                "node": r.node,
                "v": r.score / max_v,
                "b": 0.0,
            }

        for r in bm25_results:
            nid = r.node.node_id
            if nid in combined:
                combined[nid]["b"] = r.score / max_b
            else:
                combined[nid] = {"node": r.node, "v": 0.0, "b": r.score / max_b}

        # Score and sort
        from llama_index.core.schema import NodeWithScore

        results = [
            NodeWithScore(
                node=d["node"],
                score=alpha * d["v"] + (1 - alpha) * d["b"]
            )
            for d in combined.values()
        ]
        results.sort(key=lambda x: x.score, reverse=True)
        return results[:top_k]


# Usage
async def main():
    service = HybridSearchService()

    # Index documents
    from llama_index.core import SimpleDirectoryReader
    docs = SimpleDirectoryReader("./data").load_data()
    count = await service.index_documents(docs)
    print(f"Indexed {count} nodes")

    # Search
    results = await service.search(
        "RecursiveCharacterTextSplitter",
        mode="hybrid",
        alpha=0.3,  # Favor BM25 for exact term
    )

    for r in results:
        print(f"Score: {r.score:.4f}")
        print(f"Text: {r.node.text[:100]}...")


if __name__ == "__main__":
    asyncio.run(main())

---

See Also

• ../SKILL.md - Return to main skill overview
• ingestion.md - Create nodes for indexing
• context-rag.md - Query routing and reranking
• property-graphs.md - Graph-based hybrid retrieval