Data Engineering for AI on Azure: Building a RAG-Ready Data Pipeline

Scenario: A Copilot for Internal Policies That Keeps Hallucinating

Imagine a data team asked to build an internal Copilot for:

• HR and IT policies
• Project documentation
• Architecture decision records

The stack is roughly:

• Source: SharePoint Online, a couple of Azure SQL databases, and a legacy file share
• AI: Azure OpenAI (GPT-4 family) with RAG
• Storage & search: Azure Data Lake Storage Gen2 + Azure AI Search
• Orchestration: Azure Data Factory or Synapse pipelines

Symptoms from the first pilot:

• Answers reference obsolete policies that still live somewhere in the file share
• Copilot ignores important context buried deep in long PDFs
• Query latency is inconsistent; some questions take much longer than others

We’ll tighten this pipeline in three dimensions:

1. Clean, authoritative source data
2. Chunking and metadata for RAG
3. Vector search on Azure AI Search, wired correctly to Azure OpenAI

1. Start With Authoritative, Queryable Source Data

RAG quality depends more on what you index than on the model you call.

Define the “truth” layer

For our policy Copilot, the team first defines a single authoritative source per domain:

• HR policies: SharePoint site HR-Policies (modern pages + PDFs)
• IT policies: SharePoint site IT-Governance
• Architecture decisions: Azure DevOps wiki and ADR markdown in a repo

Rules:

• Only these locations are indexed into the RAG pipeline
• Legacy file shares are either migrated or explicitly excluded
• Each item is versioned, and only the latest published version is exposed to AI

Ingest into Azure Data Lake with structure

Use Azure Data Factory or Synapse pipelines to land content into Azure Data Lake Storage Gen2 in a structured layout, e.g.:

/raw
  /sharepoint
    /hr_policies
    /it_policies
  /devops
    /adr
/curated
  /policies
    /hr
    /it
  /adr

Key practices:

• Normalize formats where possible: convert Office docs and PDFs to text/HTML
• Preserve source identifiers (SharePoint URL, document ID, version) as metadata
• Maintain a curated zone that contains only:
  • Latest versions
  • Documents that pass basic quality checks (not empty, not duplicates)

For text extraction from Office and PDF into the lake, many teams use Azure Functions or Logic Apps calling Microsoft Graph and an extraction library. The exact implementation differs, but the important point: the curated zone holds clean, text-ready content.

2. Chunking: How You Slice Text Determines What the Model Sees

Long documents are a poor fit for RAG if you index them as single blobs. You want semantic chunks that:

• Fit within the token budget of your prompt
• Preserve local context (section headings, table labels)
• Are individually retrievable

For Azure AI Search and Azure OpenAI, chunking is generally done before indexing vectors.

Chunking strategy for policies

For our scenario, a practical chunking strategy:

• Chunk size: 500–1500 characters or ~200–500 tokens per chunk
• Boundaries:
  • Split on headings (H1, H2), bullet lists, and paragraph breaks
  • Avoid splitting in the middle of sentences or tables
• Overlap: 10–20% overlap between consecutive chunks to preserve context

This is not a hard rule; you’ll tune it based on your documents and model. But it’s a good starting point for GPT-4 class models.

Implement chunking in an Azure Function or Python job

A common pattern is a Python Azure Function or containerized job that reads text from the curated zone, chunks it, and writes a chunked representation back to the lake or directly into Azure AI Search.

A simplified example of a chunking function in Python:

import re
from typing import List

def chunk_text(text: str, max_chars: int = 1500, overlap_chars: int = 200) -> List[str]:
    # Normalize whitespace
    normalized = re.sub(r"\s+", " ", text).strip()

    chunks = []
    start = 0
    length = len(normalized)

    while start < length:
        end = min(start + max_chars, length)

        # Try to break at a sentence boundary
        boundary = normalized.rfind(". ", start, end)
        if boundary == -1 or boundary <= start + 100:  # fallback if no good boundary
            boundary = end
        else:
            boundary += 1  # include the period

        chunk = normalized[start:boundary].strip()
        if chunk:
            chunks.append(chunk)

        if boundary >= length:
            break

        # Move start forward with overlap
        start = max(boundary - overlap_chars, 0)

    return chunks

This function:

• Keeps chunks below max_chars
• Prefers sentence boundaries
• Adds overlap between chunks

You would call this per document, then attach metadata like document_id, section_heading, version, and source_url to each chunk.

3. Vector Search on Azure AI Search: Index Design That Serves RAG

Azure AI Search (previously Azure Cognitive Search) supports vector search with:

• Vector fields storing embeddings
• Hybrid search combining vectors with text and filters

Index schema for RAG data

For our policy Copilot, a typical index schema might include:

• id (key)
• content (full chunk text, searchable)
• contentVector (vector field for embeddings)
• title (section or document title)
• policyType (HR, IT, ADR, etc.)
• effectiveDate (for filtering out outdated policies)
• isCurrentVersion (boolean)
• sourceUrl

You define this index via REST, SDK, or the Azure portal. The critical part: mark the vector field correctly and ensure dimensions match the embedding model you use.

Creating embeddings with Azure OpenAI

Embeddings are created by calling Azure OpenAI with an embeddings model (for example, text-embedding-3-large or another current model). The response returns a vector you store in the index.

A minimal Python example using the Azure OpenAI client libraries (pattern only):

from azure.core.credentials import AzureKeyCredential
from azure.search.documents import SearchClient
from openai import AzureOpenAI

# Azure AI Search setup
search_service_endpoint = "https://<your-search-service>.search.windows.net"
index_name = "policy-index"
search_api_key = "<search-api-key>"  # store securely in Key Vault in real use

search_client = SearchClient(
    endpoint=search_service_endpoint,
    index_name=index_name,
    credential=AzureKeyCredential(search_api_key)
)

# Azure OpenAI setup
client = AzureOpenAI(
    api_key="<openai-api-key>",  # also should be in Key Vault
    api_version="2024-02-15-preview",
    azure_endpoint="https://<your-openai-resource>.openai.azure.com"
)

embedding_model = "text-embedding-3-large"  # example; use a current model


def embed_text(text: str) -> list:
    response = client.embeddings.create(
        model=embedding_model,
        input=text
    )
    return response.data[0].embedding


def index_chunk(chunk_id: str, content: str, metadata: dict):
    vector = embed_text(content)

    doc = {
        "id": chunk_id,
        "content": content,
        "contentVector": vector,
        **metadata,
    }

    search_client.upload_documents(documents=[doc])

This snippet illustrates:

• Using Azure OpenAI embeddings
• Storing the embedding in the contentVector field
• Uploading chunks with metadata to Azure AI Search

In production, you’d handle batching, retries, and secrets via Azure Key Vault and managed identities.

Querying with hybrid search for RAG

For RAG, you typically:

1. Embed the user query
2. Run a vector or hybrid search
3. Pass the top N chunks to Azure OpenAI as context

Azure AI Search supports vector search via the vector parameter and filter expressions.

Example pattern for querying (Python):

def search_chunks(query: str, top_k: int = 5):
    query_vector = embed_text(query)

    results = search_client.search(
        search_text="",  # empty when using pure vector search
        vector={
            "value": query_vector,
            "fields": "contentVector",
            "k": top_k,
        },
        filter="isCurrentVersion eq true"
    )

    chunks = []
    for result in results:
        chunks.append({
            "content": result["content"],
            "sourceUrl": result["sourceUrl"],
            "title": result.get("title"),
            "effectiveDate": result.get("effectiveDate"),
        })

    return chunks

This pattern:

• Uses vector search against the contentVector field
• Filters out outdated policy versions
• Returns chunk text plus source metadata ready for the prompt

You can switch to hybrid search by providing search_text as well, allowing keyword scoring to complement vector similarity.

4. Wiring the RAG Loop: Prompt Construction and Guardrails

Once you have clean chunks coming back from Azure AI Search, the next step is to construct the prompt for Azure OpenAI.

Prompt design for policy answers

The orchestration layer (Function App, Web API, or Logic App) typically:

1. Receives user question
2. Calls search_chunks to get top N chunks
3. Builds a system + user prompt that:
   • States the domain (internal policies)
   • Instructs the model to only use provided context
   • Includes source references for traceability

Example prompt pattern (conceptual):

def build_prompt(question: str, chunks: list) -> dict:
    context_blocks = []
    for i, c in enumerate(chunks, start=1):
        block = f"Source {i} (title: {c['title']}):\n{c['content']}\n"
        context_blocks.append(block)

    context_text = "\n".join(context_blocks)

    system_message = (
        "You are an assistant answering questions about internal HR and IT policies. "
        "Use ONLY the provided sources. If the answer is not in the sources, say you "
        "cannot find an applicable policy. Always mention which sources you used."
    )

    user_message = (
        f"Question: {question}\n\n"
        f"Sources:\n{context_text}"
    )

    return {
        "messages": [
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_message},
        ]
    }

You then send this to Azure OpenAI’s chat completion endpoint with a GPT-4 class model.

The key data engineering responsibilities here:

• Ensure chunks are small enough to fit alongside the question and instructions
• Preserve source metadata so the model can cite where information comes from
• Filter out non-current versions and non-authoritative sources before they ever reach the model

5. Operational Concerns: Freshness, Governance, and Latency

The scenario team’s initial problems (obsolete answers, missing context, slow queries) usually trace back to three operational issues.

Freshness: keep the vector index in sync

If policies change weekly, your RAG pipeline must:

• Detect new or updated documents (SharePoint change logs, Graph delta queries, SQL change tracking)
• Re-run text extraction and chunking
• Regenerate embeddings and update Azure AI Search

Common patterns:

• Incremental pipelines with Data Factory/Synapse, triggered on a schedule
• Event-driven ingestion using Event Grid or Logic Apps for certain sources

Governance: control what AI can see

Data engineering for AI is also about access control:

• Use index-level and document-level filters to enforce policy types and versions
• Align Azure AI Search access with user identity when exposing per-user data
• Keep sensitive content out of the RAG corpus unless you have a clear access model

For our policy Copilot, the team limits the index to published policies only, excluding drafts and internal notes.

Latency: trade-offs in chunk size and search

Latency issues often come from:

• Too many chunks per query (large top_k) increasing prompt size
• Overly large chunks causing token limits or slow model processing
• Complex filters or scoring profiles in Azure AI Search

Practical steps:

• Start with top_k around 3–8 and adjust based on answer quality
• Keep chunk sizes moderate and avoid multi-page chunks
• Monitor Azure AI Search query latency and adjust index configuration if needed

6. Before/After: Fixing the Copilot’s Bad Answers

Back to our scenario.

Before:

• Index included legacy file shares and old PDF exports
• Documents were indexed as single large blobs
• Search was pure keyword, no vectors, no version filters

User question: “What is the current remote work policy for employees?”

Result: Copilot surfaces a 3-year-old policy PDF from the file share, missing the latest update.

After implementing the pipeline above:

• Only curated, current policies from SharePoint are chunked and indexed
• Each chunk carries effectiveDate and isCurrentVersion
• Vector + filter search returns only current policy chunks
• Prompt explicitly instructs the model to cite sources

Same question now returns:

• A summary aligned with the latest HR policy
• A link to the SharePoint page where the policy lives
• A note that the answer is based on specific sources and effective dates

The model hasn’t changed; the data engineering around it has.

Practical Takeaway

If your Copilot agents are confidently wrong, start by treating RAG data pipeline Azure work as first-class data engineering, not an afterthought: define authoritative sources, chunk documents with context, and build a vector index in Azure AI Search that only exposes clean, current data. Once that foundation is solid, prompt tuning and model selection become optimisations instead of firefighting.