Your business runs on documents. Invoices from vendors, contracts with customers, financial reports from partners, medical intake forms — thousands of PDFs flowing through your organization daily. Yet the data trapped inside them remains invisible to your analytics layer.
Documents are the context layer for almost every business activity — planning, approvals, deep analysis, execution — all produce a massive trail of text, slides, and sheets. Yet traditional data stacks struggle to make sense of them at the accuracy enterprises require to trust automated processes.
Traditional approaches demand stitching together OCR services, custom NLP models, orchestration engines, and external storage — creating brittle pipelines that break at scale and introduce governance blind spots.
What if you could parse, classify, extract, summarize, and search documents natively inside Snowflake?
That’s exactly what we built. Our platform implements the three core document intelligence patterns that Snowflake identifies as driving real business outcomes:
Our Document Intelligence Platform processes documents through a multi-stage pipeline that keeps everything within Snowflake’s governance perimeter:
Key Cortex AI functions used:
FunctionPurposeAI_PARSE_DOCUMENTOCR and layout-aware text extraction from PDFs/imagesAI_CLASSIFYAutomatic document type routingAI_EXTRACTStructured field extraction into JSONAI_COMPLETESummarization, sentiment analysis, comparisonCORTEX.EMBEDVector embeddings for semantic search and theme mappingCORTEX.SEARCH_PREVIEWFull-text semantic search over document corpus
We organized the platform into four schemas reflecting the data lifecycle:
CREATE DATABASE DOC_INTEL;-- Raw ingestion layerCREATE SCHEMA DOC_INTEL.RAW;-- Processed & classified documentsCREATE SCHEMA DOC_INTEL.PROCESSED;-- Analytics, themes, and searchCREATE SCHEMA DOC_INTEL.ANALYTICS;-- UDFs, secrets, and application objectsCREATE SCHEMA DOC_INTEL.APP;
The staging area receives documents via file upload or automated feeds:
CREATE STAGE DOC_INTEL.RAW.DOCUMENTS_STAGE DIRECTORY = (ENABLE = TRUE) ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');
A file log tracks every document through the pipeline:
CREATE TABLE DOC_INTEL.RAW.FILE_LOG ( FILE_ID VARCHAR DEFAULT UUID_STRING(), RELATIVE_PATH VARCHAR, FILE_NAME VARCHAR, FILE_EXTENSION VARCHAR, PROCESSING_STATUS VARCHAR DEFAULT 'PENDING', CLASSIFICATION VARCHAR, UPLOAD_TS TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP());
The first pipeline stage converts raw PDFs and images into machine-readable text while preserving layout structure:
CREATE OR REPLACE DYNAMIC TABLE DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT TARGET_LAG = '10 minutes' WAREHOUSE = DOC_INTEL_WHASSELECT fl.FILE_ID, fl.RELATIVE_PATH, fl.FILE_NAME, fl.CLASSIFICATION AS DOC_TYPE, AI_PARSE_DOCUMENT( BUILD_SCOPED_FILE_URL(@DOC_INTEL.RAW.DOCUMENTS_STAGE, fl.RELATIVE_PATH), {'mode': 'LAYOUT'} ):content::VARCHAR AS FULL_TEXT, CURRENT_TIMESTAMP() AS PARSED_ATFROM DOC_INTEL.RAW.FILE_LOG flWHERE fl.PROCESSING_STATUS IN ('CLASSIFIED', 'COMPLETED');
AI_PARSE_DOCUMENT handles PDFs, scanned images, TIFF, and Office documents. The LAYOUT mode preserves the original structure — reading order across multi-column layouts, table structures, visual hierarchy, and embedded chart images. This is critical because for enterprise documents like earnings reports (with tables, charts, and footnotes) or contracts spanning hundreds of pages, structure contains meaning. A flattened text stream loses the context that downstream AI systems need for accurate retrieval and reasoning.
Instead of manual tagging, every document is automatically routed to the correct processing pipeline:
-- Classification happens in the pipeline, but here's the native approach:SELECT FILE_NAME, AI_CLASSIFY( FULL_TEXT, ['invoice', 'contract', 'financial_report', 'medical_form', 'correspondence', 'purchase_order', 'resume'] ) AS CLASSIFICATIONFROM DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT;
The classification output includes confidence scores, enabling automatic routing for high-confidence results and human review queues for edge cases. This pattern — where AI_CLASSIFY acts as a traffic controller routing different documents to specialized extraction pipelines — eliminates the manual triage that becomes a bottleneck at enterprise scale:
CREATE TABLE DOC_INTEL.PROCESSED.REVIEW_QUEUE ( FILE_ID VARCHAR, DOC_TYPE VARCHAR, EXTRACTION_CONFIDENCE FLOAT, RAW_EXTRACT VARIANT, REVIEW_STATUS VARCHAR DEFAULT 'PENDING');
Documents below a confidence threshold (we use 0.7) are flagged for manual review, while everything above flows directly to extraction.
This is where the real value materializes. AI_EXTRACT pulls structured fields directly from unstructured text — no regex, no templates, no maintenance:
CREATE OR REPLACE DYNAMIC TABLE DOC_INTEL.PROCESSED.DT_EXTRACT_INVOICES TARGET_LAG = 'DOWNSTREAM' WAREHOUSE = DOC_INTEL_WHASSELECT ft.FILE_ID, ft.FILE_NAME, AI_EXTRACT( ft.FULL_TEXT, OBJECT_CONSTRUCT( 'vendor_name', 'Name of the vendor or supplier', 'invoice_number', 'Invoice or reference number', 'invoice_date', 'Invoice date in YYYY-MM-DD format', 'total_amount', 'Total amount due as a number', 'currency', 'Currency code (USD, EUR, etc.)', 'payment_terms', 'Payment terms (e.g. Net 30)', 'line_items', 'Array of line items with description and amount' ) ) AS RAW_EXTRACT, RAW_EXTRACT:response:total_amount::FLOAT / NULLIF(RAW_EXTRACT:response:total_amount::FLOAT, 0) AS EXTRACTION_CONFIDENCE, CURRENT_TIMESTAMP() AS EXTRACTED_ATFROM DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT ftWHERE ft.DOC_TYPE = 'invoice';
CREATE OR REPLACE DYNAMIC TABLE DOC_INTEL.PROCESSED.DT_EXTRACT_CONTRACTS TARGET_LAG = 'DOWNSTREAM' WAREHOUSE = DOC_INTEL_WHASSELECT ft.FILE_ID, ft.FILE_NAME, AI_EXTRACT( ft.FULL_TEXT, OBJECT_CONSTRUCT( 'contract_title', 'Title of the agreement', 'parties', 'Names of all parties (as array)', 'effective_date', 'Start date (YYYY-MM-DD)', 'expiration_date', 'End date (YYYY-MM-DD)', 'governing_law', 'Jurisdiction', 'total_value', 'Total contract value with currency', 'auto_renewal', 'Whether contract auto-renews (true/false)' ) ) AS RAW_EXTRACT, CURRENT_TIMESTAMP() AS EXTRACTED_ATFROM DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT ftWHERE ft.DOC_TYPE = 'contract';
The extracted data lands as VARIANT (JSON) in Snowflake, immediately queryable:
-- "Which vendors invoiced us more than $50K this quarter?"SELECT RAW_EXTRACT:response:vendor_name::STRING AS vendor, SUM(RAW_EXTRACT:response:total_amount::NUMBER) AS totalFROM DOC_INTEL.PROCESSED.DT_EXTRACT_INVOICESWHERE RAW_EXTRACT:response:invoice_date::DATE >= DATEADD('quarter', -1, CURRENT_DATE())GROUP BY vendorHAVING total > 50000ORDER BY total DESC;
Every document gets an executive summary and sentiment classification using dynamic tables that auto-refresh:
CREATE OR REPLACE DYNAMIC TABLE DOC_INTEL.ANALYTICS.DT_DOC_SUMMARIES TARGET_LAG = 'DOWNSTREAM' REFRESH_MODE = INCREMENTAL WAREHOUSE = DOC_INTEL_WHASSELECT ft.FILE_ID, ft.RELATIVE_PATH, ft.FILE_NAME, ft.DOC_TYPE, SNOWFLAKE.CORTEX.COMPLETE( 'mistral-large2', 'You are a document analyst. Summarize this ' || ft.DOC_TYPE || ' document in exactly 2-3 concise sentences. Focus on key facts, amounts, and parties involved.\n\nDocument:\n' || LEFT(ft.FULL_TEXT, 8000) ) AS EXECUTIVE_SUMMARY, TRIM(SNOWFLAKE.CORTEX.COMPLETE( 'mistral-7b', 'Classify the overall sentiment of the following text. Respond with exactly ONE word: positive, negative, neutral, or mixed.\n\n' || LEFT(ft.FULL_TEXT, 4000) )) AS SENTIMENT, SNOWFLAKE.CORTEX.EMBED('snowflake-arctic-embed-m-v2.0', LEFT(ft.FULL_TEXT, 512) ) AS SUMMARY_VEC, ft.PARSED_AT AS SUMMARIZED_ATFROM DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT ft;
The SUMMARY_VEC embedding powers downstream theme mapping and semantic similarity queries.
We discover document themes using embedding cosine similarity rather than keyword matching:
CREATE OR REPLACE DYNAMIC TABLE DOC_INTEL.ANALYTICS.DT_DOC_THEME_MAP TARGET_LAG = '1 hour' WAREHOUSE = DOC_INTEL_WHASSELECT ds.FILE_ID, ds.FILE_NAME, ds.DOC_TYPE, ct.THEME_ID, ct.THEME_NAME, VECTOR_COSINE_SIMILARITY( ds.SUMMARY_VEC::VECTOR(FLOAT, 768), ct.THEME_VEC::VECTOR(FLOAT, 768) ) AS SIMILARITY_SCOREFROM DOC_INTEL.ANALYTICS.DT_DOC_SUMMARIES dsCROSS JOIN DOC_INTEL.ANALYTICS.CORPUS_THEMES ctWHERE ds.SUMMARY_VEC IS NOT NULLQUALIFY ROW_NUMBER() OVER ( PARTITION BY ds.FILE_ID ORDER BY SIMILARITY_SCORE DESC) <= 3;
This enables automatic theme detection: “Which documents are about financial operations?” becomes a vector similarity query, not a fragile text search.
Users search across their entire document corpus using natural language:
CREATE OR REPLACE CORTEX SEARCH SERVICE DOC_INTEL.ANALYTICS.DOC_SEARCH_SERVICE ON CHUNK_TEXT ATTRIBUTES DOC_TYPE, FILE_ID, PAGE_NUMBER, RELATIVE_PATH WAREHOUSE = DOC_INTEL_WH TARGET_LAG = '1 hour'ASSELECT ft.FILE_ID, ft.DOC_TYPE, c.VALUE::VARCHAR AS CHUNK_TEXT, c.INDEX + 1 AS PAGE_NUMBER, ft.RELATIVE_PATHFROM DOC_INTEL.PROCESSED.DT_DOC_FULL_TEXT ft, LATERAL FLATTEN(SPLIT(ft.FULL_TEXT, '\n\n')) cWHERE LENGTH(c.VALUE::VARCHAR) > 50;
Querying from the Streamlit app:
results = session.sql(""" SELECT PARSE_JSON( SNOWFLAKE.CORTEX.SEARCH_PREVIEW( 'DOC_INTEL.ANALYTICS.DOC_SEARCH_SERVICE', ?, ['CHUNK_TEXT', 'DOC_TYPE', 'FILE_ID', 'PAGE_NUMBER', 'RELATIVE_PATH'], 10 ) ):results AS RESULTS""", params=[user_query]).collect()
This returns semantically relevant document chunks — not just keyword matches — with faceted filtering by document type.
We built a six-page Streamlit in Snowflake application as the user-facing layer:
stats = session.sql(""" SELECT (SELECT COUNT(*) FROM DOC_INTEL.PROCESSED.DT_DOC_CLASSIFIED) AS TOTAL_DOCS, (SELECT ROUND(AVG(EXTRACTION_CONFIDENCE), 1) FROM DOC_INTEL.PROCESSED.DT_EXTRACT_INVOICES) AS AVG_CONF, (SELECT COUNT(*) FROM DOC_INTEL.ANALYTICS.CORPUS_THEMES) AS THEME_COUNT, (SELECT COUNT(*) FROM DOC_INTEL.PROCESSED.REVIEW_QUEUE WHERE REVIEW_STATUS = 'PENDING') AS PENDING_REVIEWS""").collect()[0]with st.container(horizontal=True): st.metric("Total documents", f"{stats['TOTAL_DOCS']:,}", border=True) st.metric("Avg confidence", f"{stats['AVG_CONF']}%", border=True) st.metric("Themes discovered", stats['THEME_COUNT'], border=True) st.metric("Pending reviews", stats['PENDING_REVIEWS'], border=True)
Users drag-and-drop documents directly into Snowflake’s internal stage:
uploaded_files = st.file_up( "Drag and drop files here", type=["pdf", "png", "jpg", "jpeg", "tiff", "docx", "xlsx", "pptx"], accept_multiple_files=True,)if uploaded_files: progress = st.progress(0, text="Up...") for i, f in enumerate(uploaded_files): session.file.put_stream( f, f"@DOC_INTEL.RAW.DOCUMENTS_STAGE/{f.name}", auto_compress=False ) progress.progress((i + 1) / len(uploaded_files), text=f"Uploaded {f.name}") st.toast(f"{len(uploaded_files)} file(s) uploaded successfully.")
The Compare feature sends documents to Groq (or Cortex AI) for side-by-side analysis:
The Groq UDF is a Python UDF with External Access Integration:
CREATE OR REPLACE FUNCTION DOC_INTEL.APP.GROQ_SUMMARIZE(doc_text VARCHAR, doc_type VARCHAR)RETURNS VARIANTLANGUAGE PYTHONRUNTIME_VERSION = '3.11'PACKAGES = ('requests')HANDLER = 'summarize'EXTERNAL_ACCESS_INTEGRATIONS = (GROQ_ACCESS_INTEGRATION)SECRETS = ('groq_key' = DOC_INTEL.APP.GROQ_API_KEY)AS $$import json, _snowflake, requestsdef summarize(doc_text, doc_type): api_key = _snowflake.get_generic_secret_string('groq_key') resp = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": "llama-3.3-70b-versatile", "messages": [{"role": "user", "content": f"Summarize this {doc_type}:\n{doc_text[:8000]}"}], "temperature": 0.1, "response_format": {"type": "json_object"} }, timeout=30 ) return json.loads(resp.json()["choices"][0]["message"]["content"])$$;
Natural language queries over the entire document corpus using Cortex Agents:
if prompt := st.chat_input("Ask about your documents..."): request_body = json.dumps({"messages": [{"role": "user", "content": prompt}]}) result = session.sql(""" SELECT SNOWFLAKE.CORTEX.DATA_AGENT_RUN( 'DOC_INTEL.APP.DOC_INTEL_AGENT', PARSE_JSON(?) ) AS RESPONSE """, params=[request_body]).collect()
The monitoring dashboard tracks pipeline health, extraction quality, throughput, and credit consumption:
-- Health check: detect stale pipeline stagesCREATE OR REPLACE VIEW DOC_INTEL.MONITORING.V_PIPELINE_HEALTH ASSELECT TABLE_NAME, TIMESTAMPDIFF('MINUTE', DATA_TIMESTAMP, CURRENT_TIMESTAMP()) AS MINUTES_SINCE_LAST_RECORDFROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY())WHERE SCHEMA_NAME IN ('PROCESSED', 'ANALYTICS')QUALIFY ROW_NUMBER() OVER (PARTITION BY TABLE_NAME ORDER BY DATA_TIMESTAMP DESC) = 1;
For development, testing, or accounts without Cortex AI access, we built a hybrid pipeline that calls Groq’s free-tier API externally and writes results back to Snowflake:
def summarize_document(doc_type, file_name, text): prompt = f"""Analyze this {doc_type} document and summarize it.Document: {file_name}Text: {text[:8000]}Respond ONLY with JSON:{{ "executive_summary": "2-3 sentence summary", "key_findings": ["finding1", "finding2"], "sentiment": "positive|negative|neutral|mixed"}}""" return call_groq(prompt)# Write back to Snowflakedef write_summary(conn, file_id, file_name, path, doc_type, summary): cur = conn.cursor() cur.execute(""" MERGE INTO DOC_INTEL.ANALYTICS.DT_DOC_SUMMARIES tgt USING (SELECT %s AS FILE_ID) src ON tgt.FILE_ID = src.FILE_ID WHEN MATCHED THEN UPDATE SET EXECUTIVE_SUMMARY = %s, SENTIMENT = %s, SUMMARIZED_AT = CURRENT_TIMESTAMP() WHEN NOT MATCHED THEN INSERT (FILE_ID, RELATIVE_PATH, FILE_NAME, DOC_TYPE, EXECUTIVE_SUMMARY, SENTIMENT, SUMMARIZED_AT) VALUES (%s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP()) """, (file_id, summary["executive_summary"], summary["sentiment"], file_id, path, file_name, doc_type, summary["executive_summary"], summary["sentiment"])) conn.commit()
This means you can prototype immediately with Groq’s free tier (Llama 3.3 70B), then switch to Cortex AI in production — same schema, same app, zero code changes in the Streamlit layer.
Proving a document use case works on 10 files is straightforward. The real enterprise challenge is processing hundreds of thousands of documents daily without your pipeline collapsing. Dynamic tables solve this by letting you declare the transformation as SQL — Snowflake handles scheduling, refresh orchestration, dependency ordering, and automatic catch-up after failures. No task graph to debug, no DAG orchestrator to maintain. Setting REFRESH_MODE = INCREMENTAL guarantees each incoming document is processed exactly once, enabling predictable cost at scale.
Every user-facing SQL uses bind parameters (params=[...]) instead of f-string interpolation. This prevents SQL injection — critical when file names come from user uploads.
Not every extraction is perfect. Low-confidence results (below 70%) are automatically flagged for human review rather than silently propagating bad data downstream.
Keyword-based classification breaks when document language evolves. Vector similarity finds conceptual matches — a “liability clause” and a “risk mitigation section” are recognized as related even with zero keyword overlap.
After deploying this platform:
Youtube Demo:
Please find the complete source code on Github
Built on Snowflake. Powered by Groq/Cortex AI. Governed by default.
👏 Give it a clap if it added value 🔗 Share it with your team
➕ Follow for more
📘 Medium:
🔗 LinkedIn: satishkumar-snowflake
See you in the next one! 👋
Building a Document Intelligence Platform on Snowflake with Cortex AI was originally published in Towards AI on Medium, where people are continuing the conversation by highlighting and responding to this story.