Enterprise AI Data Engineering With Snowflake Cortex and RAG

Where the Data Actually Lives

Every enterprise I have worked with hits the same wall. Mountains of data. Warehouses, ticketing systems, PDFs, old email archives. Most of that data is not ready for AI to use. Leadership wants a chatbot that can answer questions about policy and product specs. But no one knows where the data lives or how to get it ready. A bigger model will not fix that.  It gets solved by data engineering.

The Pattern That Shows Up Again and Again

• A chatbot wired straight to a foundation model, no retrieval layer at all. It answers from memory and gets the specifics wrong.
• Documents sitting in five different systems, none of them governed the same way.
• Embeddings computed once, at launch, and never refreshed again.
• A RAG pipeline built without anyone checking who has write access to the source documents.

The gap is never the model. It is what happens before the model ever sees the question.

This is not a problem reserved for large enterprises with dedicated ML teams. A three-person startup wiring a chatbot to their support docs runs into the exact same wall. The model does not care how big your organization is. If your data is messy and ungoverned, your answers will be too.

What Snowflake Cortex Actually Does

Snowflake Cortex is a set of AI functions built directly into the Snowflake platform. Embeddings, summarization, completion. Call them right inside SQL or through the Python connector. The data never has to leave.

That matters more than it sounds. Every time data crosses a boundary to reach an external AI vendor, that is one more place it can leak, one more thing to govern, one more contract to review. Cortex removes the boundary.

Teams use it two ways in practice. Straight from SQL, for lightweight transformations. Through Snowpark Python, when the logic needs to sit inside a broader application. Either way, the embedding or the completion happens inside Snowflake’s governed environment.

cortex_embed.py
import snowflake.connector

conn = snowflake.connector.connect(
    user='YOUR_USER', password='YOUR_PASSWORD', account='YOUR_ACCOUNT',
    warehouse='COMPUTE_WH', database='ENTERPRISE_DB', schema='AI_SCHEMA'
)
cur = conn.cursor()
cur.execute(
    "SELECT SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2', 'Quarterly compliance policy summary') AS embedding"
)
result = cur.fetchone()
print(len(result[0]))  # vector length

Connect. Call a Cortex function inside a SQL statement. Pull the result back. That pattern repeats everywhere in this pipeline.

What RAG Actually Is

Retrieval-augmented generation is not complicated once you strip the marketing off it. Retrieve relevant text from your own data. Hand it to the model. Ask it to answer using only that context, not memory. That single design decision is what cuts hallucination and makes citations possible.

The workflow, broken into its actual steps: a question comes in, gets embedded into a vector, the system searches for the closest matching chunks, and those chunks get passed to the model with the original question. The model answers from what it was given. Nothing more.

rag_query.py
def answer_question(question, cursor, top_k=3):
    cursor.execute(
        "SELECT SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2', %s)", (question,)
    )
    query_vector = cursor.fetchone()[0]
    chunks = search_similar_chunks(cursor, query_vector, top_k)
    context = '\n\n'.join(chunks)
    return generate_answer(cursor, question, context)

Everything else in this article is just filling in the details behind three steps. Embed. Retrieve. Generate.

The Full Architecture

Data flows in from enterprise sources. It gets cleaned and staged in Snowflake. Cortex generates embeddings. A vector store supports fast retrieval. A retrieval layer sits in between. The enterprise user sees none of this. They just ask a question. Most teams underestimate the plumbing. Embeddings and generation are easy. Reliable ingestion is hard. So is a fresh index. So is retrieval quality as the document set grows. That is where the real work is.

Building the Pipeline

This is standard data engineering. Read raw files. Validate. Stage. Merge into production using a key that blocks duplicate loads. Nothing exotic. Get it wrong here. Every answer downstream gets worse. It happens quietly. No error tells you why.

load_documents.py
import pandas as pd

def load_and_clean(csv_path):
    df = pd.read_csv(csv_path)
    df = df.dropna(subset=['document_text']).drop_duplicates()
    df['document_text'] = df['document_text'].str.strip().str.replace(r'\s+', ' ', regex=True)
    return df

def load_documents_to_snowflake(df, conn, table='DOCS_STAGING'):
    cur = conn.cursor()
    for _, row in df.iterrows():
        cur.execute(
            f"INSERT INTO {table} (doc_id, source, document_text) VALUES (%s, %s, %s)",
            (row['doc_id'], row['source'], row['document_text'])
        )
    conn.commit()
    cur.close()

This is deliberately small. In production, batch inserts. Wrap them in a transaction. Add a merge step. This stops duplicate rows on reruns. What matters most: by this point, data is clean. It is already validated.

Chunking Is Where Most Pipelines Quietly Fail

Chunking matters more than expected. Too large dilutes relevance. Too small loses context. Most teams pick 300 to 800 tokens. They add some overlap too. Even that gets argued over.

chunk_and_embed.py
def chunk_text(text, chunk_size=500, overlap=50):
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(' '.join(words[start:end]))
        start = end - overlap
    return chunks

def embed_and_store(cursor, doc_id, text):
    for i, chunk in enumerate(chunk_text(text)):
        cursor.execute(
            "INSERT INTO DOC_EMBEDDINGS (doc_id, chunk_id, chunk_text, embedding) "
            "SELECT %s, %s, %s, SNOWFLAKE.CORTEX.EMBED_TEXT_768('e5-base-v2', %s)",
            (doc_id, i, chunk, chunk)
        )

A sliding window chunks the document. Cortex computes the vector inside Snowflake, in the same insert. No separate embedding service to babysit.

Change the chunk size later, and you are not tweaking a config. You are re-embedding everything.

Querying With Vector Search

With embeddings stored, retrieval becomes a nearest-neighbor search. Snowflake runs this natively. Plain SQL. No separate vector database to stand up. No extra system to operate.

What Actually Pays Off

• Cache embeddings for repeated queries. Do not recompute what you already have.
• Tag every chunk with its access permissions. Retrieval that ignores row-level security is a data leak waiting to happen.
• Log every retrieval and every generation call. You cannot debug a bad answer you did not record.
• Version your chunking strategy. Changing it later means re-embedding everything, whether you planned for it or not. 

log_wrapper.py
import time

def answer_with_logging(question, cursor, logger):
    start = time.time()
    answer = answer_question(question, cursor)
    duration = time.time() - start
    logger.info(f"query='{question}' duration={duration:.2f}s")
    return answer

This kind of logging gets skipped early. It becomes essential the moment a real user reports a confusing answer, and you need a trail to investigate.

Where This Breaks

Every team hits the same failure modes the first time they run RAG against messy enterprise data.

The Usual Suspects

• Stale index. The vector store stopped reflecting reality weeks ago, and nobody noticed.
• Chunk drift. Retrieval quality degrades because the chunk size was wrong from the start.
• Prompt injection through retrieved documents. Anyone with write access to your corpus has indirect access to your model.
• Latency that traces back to an unindexed similarity search or a context window stuffed too full.

None of these is exotic. All of them are common.

Stale indexes get fixed with a scheduled incremental refresh, not a full re-embed. Chunk drift gets fixed by testing a few chunk sizes against a fixed set of real questions. Latency gets fixed with tuning, not a redesign.

Prompt injection is the one people fix last, usually after it has already happened once.

sanitize_chunk.py
def sanitize_chunk(chunk_text):
    forbidden_phrases = ['ignore previous instructions', 'disregard the above']
    lowered = chunk_text.lower()
    for phrase in forbidden_phrases:
        if phrase in lowered:
            chunk_text = chunk_text.replace(phrase, '[removed]')
    return chunk_text

This is a basic example. It is not a complete defense. Retrieved content is untrusted input. It works the same as user input in a web app. Treat it that way from the first line of code.

This Does Not End at Launch

Enterprise AI built on Snowflake Cortex and RAG is not about clever prompting. It is disciplined data engineering wearing a chatbot’s face.

Clean ingestion. Sensible chunking. Governed embeddings. A retrieval layer that respects access control. That does more for answer quality than swapping in a bigger model ever will. Cortex removes the operational overhead of moving data somewhere else to get an embedding. RAG keeps the model grounded in what your organization actually knows, instead of what a foundation model memorized during training.

If you are starting this kind of project, resist the urge to jump straight to the LLM integration. Get the pipeline right first. Measure retrieval quality against real questions from real users. Treat the whole thing as production software from day one, monitoring, logging, and security built in rather than bolted on later.

Start with the data. The model is the easy part.