Building RAG Pipelines with Azure Cognitive Search

How to Wire Up Retrieval-Augmented Generation That Actually Works in Production

If you've spent any time building LLM-powered applications, you've almost certainly hit the wall where your model confidently returns outdated, hallucinated, or simply wrong information. That's where Retrieval-Augmented Generation (RAG) enters the picture - and Azure Cognitive Search (now branded as Azure AI Search) is one of the most capable platforms to build it on.

This isn't a surface-level walkthrough. We're going deep on architecture decisions, indexing strategies, chunking logic, and the quirks you'll only discover after you've already pushed something to production.

---

What RAG Actually Solves (And What It Doesn't)

Before touching any Azure resources, let's be honest about what RAG is doing for you.

RAG augments an LLM's response by retrieving relevant context from an external knowledge base at inference time, then injecting that context into the prompt. The model answers based on what you give it, not just what it memorized during training.

RAG solves:

• Hallucinations caused by knowledge cutoffs
• Domain-specific knowledge gaps in general-purpose models
• The impracticality of fine-tuning for every document update

RAG does NOT solve:

• Poor document quality upstream
• Bad chunking strategies that fragment semantic meaning
• Retrievals that return the right document but the wrong passage

Azure AI Search handles the retrieval layer exceptionally well - but only if you set it up correctly.

---

The Architecture at a Glance

Here's the end-to-end RAG pipeline we're building:

Raw Documents (PDF, DOCX, HTML)
        ↓
[Document Processing / Chunking]
        ↓
[Azure AI Search Indexer + Skillset]
        ↓
[Vector Embeddings via Azure OpenAI]
        ↓
[Azure AI Search Index]
        ↓
User Query → [Embedding] → [Hybrid Search] → [Top-K Results]
        ↓
[Prompt Construction]
        ↓
[Azure OpenAI GPT-4] → Final Response

Each of these stages has real decisions baked into it. Let's work through them one by one.

---

Step 1: Setting Up Your Azure AI Search Resource

Start by provisioning your search service. For production RAG workloads, the tier you choose matters more than most people realize.

# Create resource group
az group create \
  --name rg-rag-pipeline \
  --location eastus2

# Create Azure AI Search service (Standard tier minimum for production)
az search service create \
  --name my-rag-search \
  --resource-group rg-rag-pipeline \
  --sku Standard \
  --partition-count 1 \
  --replica-count 2

Tier Reality Check: The Free and Basic tiers do not support vector search. You need Standard or above. If you're running semantic ranking alongside vector search (and you should be), budget accordingly - semantic ranking is billed per 1,000 queries.

---

Step 2: Document Ingestion and Chunking Strategy

This is where most RAG pipelines quietly fail. Poor chunking is the silent killer of retrieval accuracy.

The Chunking Problem

Imagine a 50-page technical manual. If you chunk it naively at 512 tokens, you'll split sentences, separate a table from its header, and divorce a code example from its explanation. Your embeddings will be meaningless noise.

Recommended Chunking Approaches

Fixed-size chunking with overlap - Simple, works reasonably well for dense prose:

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", ". ", " "]
)

chunks = splitter.split_text(document_text)

Semantic chunking - Better for mixed-content documents where topic shifts matter:

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import AzureOpenAIEmbeddings

embeddings = AzureOpenAIEmbeddings(
    azure_deployment="text-embedding-ada-002",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_KEY"]
)

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90
)

chunks = semantic_splitter.split_text(document_text)

Hierarchical chunking - Best for structured documents like legal contracts or technical specs. You store both a parent chunk (broader context) and child chunks (specific retrieval units), then return the parent when a child matches:

# Store parent-child relationships in metadata
def create_hierarchical_chunks(document, parent_size=2048, child_size=256):
    parent_chunks = []
    child_chunks = []
    
    parents = split_text(document, chunk_size=parent_size, overlap=0)
    
    for p_idx, parent in enumerate(parents):
        parent_id = f"parent_{p_idx}"
        parent_chunks.append({
            "id": parent_id,
            "content": parent,
            "chunk_type": "parent"
        })
        
        children = split_text(parent, chunk_size=child_size, overlap=32)
        for c_idx, child in enumerate(children):
            child_chunks.append({
                "id": f"child_{p_idx}_{c_idx}",
                "content": child,
                "parent_id": parent_id,
                "chunk_type": "child"
            })
    
    return parent_chunks, child_chunks

---

Step 3: Creating the Azure AI Search Index

The index schema is your retrieval contract. Get this wrong and you'll be rebuilding it six weeks later.

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SemanticConfiguration,
    SemanticSearch,
    SemanticPrioritizedFields,
    SemanticField
)
from azure.core.credentials import AzureKeyCredential

index_client = SearchIndexClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["SEARCH_API_KEY"])
)

fields = [
    SearchField(name="id", type=SearchFieldDataType.String, key=True),
    SearchField(
        name="content",
        type=SearchFieldDataType.String,
        searchable=True,
        analyzer_name="en.microsoft"
    ),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=1536,
        vector_search_profile_name="hnsw-profile"
    ),
    SearchField(name="source_file", type=SearchFieldDataType.String, filterable=True),
    SearchField(name="page_number", type=SearchFieldDataType.Int32, filterable=True),
    SearchField(name="parent_id", type=SearchFieldDataType.String, filterable=True),
    SearchField(
        name="metadata",
        type=SearchFieldDataType.String,
        searchable=False
    )
]

vector_search = VectorSearch(
    algorithms=[
        HnswAlgorithmConfiguration(
            name="hnsw-algo",
            parameters={
                "m": 4,
                "efConstruction": 400,
                "efSearch": 500,
                "metric": "cosine"
            }
        )
    ],
    profiles=[
        VectorSearchProfile(
            name="hnsw-profile",
            algorithm_configuration_name="hnsw-algo"
        )
    ]
)

semantic_config = SemanticConfiguration(
    name="my-semantic-config",
    prioritized_fields=SemanticPrioritizedFields(
        content_fields=[SemanticField(field_name="content")]
    )
)

index = SearchIndex(
    name="rag-documents",
    fields=fields,
    vector_search=vector_search,
    semantic_search=SemanticSearch(configurations=[semantic_config])
)

index_client.create_or_update_index(index)
print("Index created successfully")

HNSW Parameters Explained

The Hierarchical Navigable Small World (HNSW) algorithm powers your vector search. These parameters have real throughput and accuracy implications:

Parameter	What It Does	Production Guidance
m	Number of bidirectional links per node	Higher = better recall, more memory. Start at 4, test up to 16
efConstruction	Nodes examined during index build	Higher = slower build, better index quality. 400 is solid
efSearch	Nodes examined during query	Higher = slower queries, better recall. Tune against your latency SLA
metric	Distance function	Use cosine for text embeddings

---

Step 4: Generating and Uploading Embeddings

from openai import AzureOpenAI
from azure.search.documents import SearchClient

openai_client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_KEY"],
    api_version="2024-02-01"
)

search_client = SearchClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    index_name="rag-documents",
    credential=AzureKeyCredential(os.environ["SEARCH_API_KEY"])
)

def generate_embedding(text: str) -> list[float]:
    response = openai_client.embeddings.create(
        input=text,
        model="text-embedding-ada-002"
    )
    return response.data[0].embedding

def upload_chunks_with_embeddings(chunks: list[dict], batch_size: int = 100):
    """Upload in batches to avoid hitting API limits"""
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        documents = []
        
        for chunk in batch:
            embedding = generate_embedding(chunk["content"])
            documents.append({
                "id": chunk["id"],
                "content": chunk["content"],
                "content_vector": embedding,
                "source_file": chunk.get("source_file", ""),
                "page_number": chunk.get("page_number", 0),
                "parent_id": chunk.get("parent_id", ""),
                "metadata": json.dumps(chunk.get("metadata", {}))
            })
        
        result = search_client.upload_documents(documents=documents)
        print(f"Batch {i//batch_size + 1}: {len(result)} documents uploaded")
        
        # Respect rate limits
        time.sleep(0.5)

---

Step 5: Hybrid Search - The Real Secret Weapon

This is the part most tutorials skip, and it's where Azure AI Search genuinely shines over alternatives. Pure vector search misses exact keyword matches. Pure BM25 keyword search misses semantic similarity. Hybrid search combines both.

from azure.search.documents.models import VectorizedQuery

def hybrid_search(
    query: str,
    top_k: int = 5,
    filter_expression: str = None
) -> list[dict]:
    
    # Generate embedding for the query
    query_vector = generate_embedding(query)
    
    vector_query = VectorizedQuery(
        vector=query_vector,
        k_nearest_neighbors=top_k,
        fields="content_vector"
    )
    
    results = search_client.search(
        search_text=query,           # BM25 keyword search
        vector_queries=[vector_query],  # Vector similarity search
        select=["id", "content", "source_file", "page_number"],
        filter=filter_expression,
        query_type="semantic",
        semantic_configuration_name="my-semantic-config",
        query_caption="extractive",
        query_answer="extractive",
        top=top_k
    )
    
    retrieved_docs = []
    for result in results:
        retrieved_docs.append({
            "id": result["id"],
            "content": result["content"],
            "source": result["source_file"],
            "page": result["page_number"],
            "score": result["@search.score"],
            "reranker_score": result.get("@search.reranker_score")
        })
    
    return retrieved_docs

The Reciprocal Rank Fusion (RRF) Advantage

When you run hybrid search in Azure AI Search, results from BM25 and vector search are merged using Reciprocal Rank Fusion. RRF doesn't care about the raw scores from each system - it uses the rank position from each retrieval method to compute a fused score. This makes hybrid search reliable across document types.

---

Step 6: Building the Generation Layer

Now we have our retrieved context. Time to build a prompt that doesn't waste it.

def build_rag_prompt(query: str, retrieved_docs: list[dict]) -> list[dict]:
    
    # Format context with source attribution
    context_blocks = []
    for idx, doc in enumerate(retrieved_docs):
        context_blocks.append(
            f"[Source {idx+1}: {doc['source']}, Page {doc['page']}]\n"
            f"{doc['content']}"
        )
    
    context = "\n\n---\n\n".join(context_blocks)
    
    system_prompt = """You are a precise technical assistant. Answer questions 
    based ONLY on the provided context. If the context does not contain enough 
    information to answer the question, say so explicitly. Do not fabricate 
    information. Always cite the source number when referencing specific information."""
    
    user_prompt = f"""Context:
{context}

Question: {query}

Answer based on the context above, citing sources where relevant:"""
    
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]


def generate_rag_response(query: str, top_k: int = 5) -> dict:
    # Retrieve
    retrieved_docs = hybrid_search(query, top_k=top_k)
    
    if not retrieved_docs:
        return {
            "answer": "No relevant documents found for your query.",
            "sources": []
        }
    
    # Augment
    messages = build_rag_prompt(query, retrieved_docs)
    
    # Generate
    response = openai_client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        temperature=0.1,       # Low temperature for factual responses
        max_tokens=1000
    )
    
    return {
        "answer": response.choices[0].message.content,
        "sources": [
            {"source": doc["source"], "page": doc["page"]}
            for doc in retrieved_docs
        ],
        "tokens_used": response.usage.total_tokens
    }

---

Step 7: Using Azure AI Search Indexers for Automated Ingestion

For production pipelines, you don't want to manually upload documents. Azure AI Search Indexers can automatically pull from Azure Blob Storage and run your AI Skillsets on the way in.

from azure.search.documents.indexes.models import (
    SearchIndexer,
    SearchIndexerDataSourceConnection,
    SearchIndexerDataSourceType,
    SearchIndexerSkillset,
    OcrSkill,
    MergeSkill,
    SplitSkill,
    AzureOpenAIEmbeddingSkill,
    InputFieldMappingEntry,
    OutputFieldMappingEntry
)

# Create data source connection (Azure Blob Storage)
data_source =