The Complete Production SOP: How to Deploy and Run LlamaIndex Advanced RAG Retrieval Pipeline in Autonomous AI Agents

---
name: "llamaindex-advanced-rag-pipeline"
description: "Build enterprise retrieval-augmented generation with hierarchical node parsing and reranking."
metadata:
  author: "LlamaIndex"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "build advanced rag llamaindex"
  - "use sentence window retriever"
  - "add reranker llamaindex"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

import os
from dotenv import load_dotenv
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

load_dotenv('.env.local')

# Enforce deterministic temperature and structured settings
Settings.llm = OpenAI(model='gpt-4o', temperature=0.0)
Settings.embed_model = OpenAIEmbedding(model='text-embedding-3-small')

from llama_index.core.node_parser import SentenceWindowNodeParser
from llama_index.core.indices.postprocessor import MetadataReplacementPostProcessor
from llama_index.postprocessor.cohere_rerank import CohereRerank
from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage

# 1. Initialize SentenceWindow parser
node_parser = SentenceWindowNodeParser.from_defaults(
    window_size=3,
    window_metadata_key='window',
    original_text_metadata_key='original_text'
)

# 2. Configure two-stage reranking: Fetch 10 candidates, rerank to top 3
cohere_rerank = CohereRerank(top_n=3)
metadata_replace = MetadataReplacementPostProcessor(target_metadata_key='window')

def create_query_engine(index: VectorStoreIndex):
    return index.as_query_engine(
        similarity_top_k=10,
        node_postprocessors=[metadata_replace, cohere_rerank]
    )

import pytest
from llama_index.core.schema import TextNode, NodeWithScore
from rag_engine import cohere_rerank

def test_reranker_truncation():
    nodes = [
        NodeWithScore(node=TextNode(text=f'Context snippet {i}'), score=0.1 * i)
        for i in range(10)
    ]
    processed = cohere_rerank.postprocess_nodes(nodes, query_str='test query')
    assert len(processed) <= 3, 'Postprocessor failed to constrain output to top_n=3'