Overview

Prerequisites

What You'll Learn

What You'll Need

What You'll Build

Set Up Required Objects and Privileges

Create Database and Schema

CREATE DATABASE doc_ai_db;
CREATE SCHEMA doc_ai_db.doc_ai_schema;

Create Custom Role

USE ROLE ACCOUNTADMIN;
CREATE ROLE doc_ai_role;

Grant Required Privileges

-- 1. Grant the SNOWFLAKE.CORTEX_USER database role to the doc_ai_role role:
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE doc_ai_role;

-- 2. Grant warehouse usage and operating privileges to the doc_ai_role role:
GRANT USAGE, OPERATE ON WAREHOUSE <your_warehouse> TO ROLE doc_ai_role;

-- 3. Grant the privileges to use the database and schema you created to the doc_ai_role:
GRANT USAGE ON DATABASE doc_ai_db TO ROLE doc_ai_role;
GRANT USAGE ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;

-- 4. Grant the create stage privilege on the schema to the doc_ai_role role to store the documents for extraction:
GRANT CREATE STAGE ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;

-- 5. Grant the privileges required to create a processing pipeline using streams and tasks to the doc_ai_role role:
GRANT CREATE STREAM, CREATE TABLE, CREATE TASK, CREATE FUNCTION, CREATE VIEW ON SCHEMA doc_ai_db.doc_ai_schema TO ROLE doc_ai_role;
GRANT EXECUTE TASK ON ACCOUNT TO ROLE doc_ai_role;

-- 6. Grant the doc_ai_role to tutorial user for use in the next steps of the tutorial:
GRANT ROLE doc_ai_role TO USER <your_user_name>;

Enable Cross-Region Inference (If Needed)

ALTER ACCOUNT SET CORTEX_ENABLED_CROSS_REGION = 'ANY_REGION';

Assume the New Role

USE ROLE doc_ai_role;

Set Up Prompt Table

Create the Prompt Table

CREATE TABLE IF NOT EXISTS prompt_templates (
    template_id VARCHAR PRIMARY KEY,
    response_format VARIANT
);

Define Extraction Schema

INSERT INTO prompt_templates
SELECT
    'INSPECTION_REVIEWS',
    PARSE_JSON('{
        "schema": {
            "type": "object",
            "properties": {
                "list_of_units": {
                    "description": "Extract the table showing all units and their reported conditions",
                    "type": "object",
                    "column_ordering": ["unit_name", "condition"],
                    "properties": {
                        "unit_name": {
                            "description": "Name of the unit",
                            "type": "array"
                        },
                        "condition": {
                            "description": "Condition reported for the unit",
                            "type": "array"
                        }
                    }
                },
                "inspection_date": {
                    "description": "What is the inspection date?",
                    "type": "string"
                },
                "inspection_grade": {
                    "description": "What is the grade?",
                    "type": "string"
                },
                "inspector": {
                    "description": "Who performed the inspection?",
                    "type": "string"
                }
            }
        }
    }');

Create AI_EXTRACT Wrapper Function

Create Internal Stage

CREATE STAGE IF NOT EXISTS my_pdf_stage
  DIRECTORY = (ENABLE = TRUE)
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

Create the Wrapper Function

CREATE OR REPLACE FUNCTION extract_document_data(
    stage_name STRING,
    file_path STRING
)
RETURNS VARIANT
LANGUAGE SQL
AS $$
    SELECT AI_EXTRACT(
        file => TO_FILE(stage_name, file_path),
        responseFormat => (
            SELECT response_format 
            FROM prompt_templates 
        )
    ):response
$$;

Test AI_EXTRACT

Upload Sample Documents

View Extracted Data

SELECT
    RELATIVE_PATH,
    extract_document_data('@my_pdf_stage', RELATIVE_PATH) as EXTRACTED_DATA
FROM DIRECTORY('@my_pdf_stage');

Create Automated Processing Pipeline

Create Stream on Stage

CREATE STREAM my_pdf_stream ON STAGE my_pdf_stage;
ALTER STAGE my_pdf_stage REFRESH;

Create Results Table

CREATE OR REPLACE TABLE pdf_reviews (
  file_name VARCHAR,
  file_size VARIANT,
  last_modified VARCHAR,
  snowflake_file_url VARCHAR,
  json_content VARIANT
);

Create Processing Task

CREATE OR REPLACE TASK load_new_file_data
  WAREHOUSE = <your_warehouse>
  SCHEDULE = '1 minutes'
  COMMENT = 'Process new files in the stage and insert data into the pdf_reviews table.'
WHEN SYSTEM$STREAM_HAS_DATA('my_pdf_stream')
AS
INSERT INTO pdf_reviews (
  SELECT
    RELATIVE_PATH AS file_name,
    size AS file_size,
    last_modified,
    file_url AS snowflake_file_url,
    extract_document_data('@my_pdf_stage',RELATIVE_PATH) AS json_content
  FROM my_pdf_stream
  WHERE METADATA$ACTION = 'INSERT'
);

Resume the Task

ALTER TASK load_new_file_data RESUME;

Process New Documents

Upload Documents to Stage

Note that this will not process the existing files in the stage. To include these files, delete the existing files from the stage prior to uploading the folder contents. You can quickly remove all existing files from the stage using:

REMOVE @my_pdf_stage;

View Extracted Information

SELECT * FROM pdf_reviews;

Note that as the task runs every minute, the table may take approximately 1-2 minutes to be populated.

Create Analysis View

CREATE VIEW pdf_reviews_view AS
SELECT 
    file_name,
    file_size,
    last_modified,
    snowflake_file_url,
    json_content:inspection_date::STRING AS inspection_date,
    json_content:inspection_grade::STRING AS inspection_grade,
    json_content:inspector::STRING AS inspector,
    json_content:list_of_units:unit_name::ARRAY AS list_of_units_name,
    json_content:list_of_units:condition::ARRAY AS list_of_units_condition,
FROM pdf_reviews;

SELECT * FROM pdf_reviews_view;

Conclusion

What You Learned

Related Resources