Overview

Why Gen2 Warehouses?

System Architecture

User Query → Coordinator Agent → 6 Specialized Components:
  ├── Document Retrieval Tool (Cortex Search)
  ├── SQL Generator (Cortex Analyst)
  ├── Metrics Calculator (Aggregations)
  ├── Chart Generator (Visualization)
  ├── Semantic Model Agent (Business Logic)
  └── Response Synthesizer (Final Output)

What You'll Build

Prerequisites

Setup Environment

Create Database and Schema

-- Create dedicated database for multi-agent RAG
CREATE DATABASE IF NOT EXISTS MULTIAGENT_RAG;
USE DATABASE MULTIAGENT_RAG;

CREATE SCHEMA IF NOT EXISTS AGENTS;
USE SCHEMA AGENTS;

-- Create Gen2 warehouse optimized for concurrent agent execution
CREATE WAREHOUSE IF NOT EXISTS MULTIAGENT_WH
  WAREHOUSE_SIZE = 'MEDIUM'
  WAREHOUSE_TYPE = 'STANDARD'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 5
  SCALING_POLICY = 'STANDARD';

USE WAREHOUSE MULTIAGENT_WH;

Create Sample Data Tables

-- Healthcare patient data for demonstration
CREATE OR REPLACE TABLE PATIENT_VISITS (
  visit_id INT,
  patient_id INT,
  diagnosis VARCHAR,
  treatment_cost DECIMAL(10,2),
  visit_date DATE,
  zip_code VARCHAR(10),
  readmission BOOLEAN
);

-- Insert sample data
INSERT INTO PATIENT_VISITS VALUES
  (1, 1001, 'Hypertension', 1500.00, '2024-01-15', '10453', FALSE),
  (2, 1002, 'Diabetes', 2200.50, '2024-01-18', '10453', TRUE),
  (3, 1003, 'Hypertension', 1800.00, '2024-02-10', '10454', FALSE),
  (4, 1004, 'Diabetes', 2500.00, '2024-02-22', '10455', TRUE),
  (5, 1005, 'Heart Disease', 5000.00, '2024-03-05', '10453', FALSE);

Document Ingestion & Embedding

Using Native PARSE_DOCUMENT Feature

-- Create table for document chunks
CREATE OR REPLACE TABLE DOCUMENT_CHUNKS (
  chunk_id INT AUTOINCREMENT,
  document_id STRING,
  chunk_text STRING,
  metadata VARIANT
);

-- Use PARSE_DOCUMENT to extract and chunk text
CREATE OR REPLACE PROCEDURE ingest_documents()
RETURNS STRING
LANGUAGE SQL
AS
$$
BEGIN
  -- Parse document using native PARSE_DOCUMENT
  INSERT INTO DOCUMENT_CHUNKS (document_id, chunk_text, metadata)
  SELECT 
    'MED_GUIDE_001' as document_id,
    value:chunk::STRING as chunk_text,
    OBJECT_CONSTRUCT(
      'source', 'medical_guideline',
      'date', CURRENT_DATE(),
      'chunk_index', index
    ) as metadata
  FROM TABLE(
    FLATTEN(
      input => PARSE_DOCUMENT(
        @document_stage/medical_guidelines.pdf,
        {'mode': 'LAYOUT', 'chunk_size': 500}
      )
    )
  );
  
  RETURN 'Documents ingested successfully';
END;
$$;

Setup Cortex Search Service

-- Create Cortex Search service for vector similarity
CREATE OR REPLACE CORTEX SEARCH SERVICE medical_docs_search
  ON chunk_text
  WAREHOUSE = MULTIAGENT_WH
  TARGET_LAG = '1 minute'
  AS (
    SELECT 
      chunk_id,
      chunk_text,
      document_id,
      metadata
    FROM DOCUMENT_CHUNKS
  );

Component 1: Document Retrieval Tool

Create Document Retrieval Tool with Cortex Search

CREATE OR REPLACE FUNCTION document_retrieval_tool(user_query STRING)
RETURNS TABLE (chunk_text STRING, relevance_score FLOAT, source STRING)
AS
$$
  -- Use Cortex Search for semantic similarity
  SELECT 
    chunk_text,
    score as relevance_score,
    metadata:source::STRING as source
  FROM TABLE(
    MEDICAL_DOCS_SEARCH!SEARCH(
      query => user_query,
      limit => 5
    )
  )
  ORDER BY score DESC
$$;

Component 2: SQL Generator with Cortex Analyst

Create Semantic Model for Cortex Analyst

# semantic_model.yaml
name: patient_visits_model
tables:
  - name: PATIENT_VISITS
    description: "Patient visit records with diagnosis and cost information"
    base_table:
      database: MULTIAGENT_RAG
      schema: AGENTS
      table: PATIENT_VISITS
    dimensions:
      - name: diagnosis
        synonyms: ["condition", "illness", "disease"]
        description: "Patient diagnosis or medical condition"
      - name: zip_code
        synonyms: ["ZIP", "postal code", "area"]
        description: "Geographic location ZIP code"
    measures:
      - name: treatment_cost
        synonyms: ["cost", "price", "expense"]
        description: "Cost of treatment in dollars"
        aggregation: AVG
      - name: readmission
        description: "Whether patient was readmitted"
        aggregation: SUM

Setup Cortex Analyst

-- Create stage for semantic model
CREATE STAGE IF NOT EXISTS semantic_models;

-- Upload semantic_model.yaml to stage
-- PUT file://semantic_model.yaml @semantic_models;

-- Create Cortex Analyst function
CREATE OR REPLACE FUNCTION sql_generator_agent(user_question STRING)
RETURNS TABLE
LANGUAGE SQL
AS
$$
  -- Use Cortex Analyst for high-quality SQL generation
  SELECT * FROM TABLE(
    SNOWFLAKE.CORTEX.ANALYST!QUERY(
      question => user_question,
      semantic_model => '@semantic_models/semantic_model.yaml'
    )
  )
$$;

Component 3: Metrics Calculator

Advanced Metrics Agent

CREATE OR REPLACE FUNCTION metrics_calculator_agent(metric_type STRING, filters VARIANT)
RETURNS VARIANT
LANGUAGE SQL
AS
$$
  WITH metrics AS (
    SELECT
      COUNT(DISTINCT patient_id) as total_patients,
      AVG(treatment_cost) as avg_cost,
      SUM(CASE WHEN readmission THEN 1 ELSE 0 END) * 100.0 / COUNT(*) as readmission_rate,
      APPROX_PERCENTILE(treatment_cost, 0.5) as median_cost,
      STDDEV(treatment_cost) as cost_stddev,
      MAX(visit_date) as latest_visit,
      COUNT(*) as total_visits
    FROM PATIENT_VISITS
    WHERE
      ($filters:diagnosis IS NULL OR diagnosis = $filters:diagnosis)
      AND ($filters:zip_code IS NULL OR zip_code = $filters:zip_code)
  )
  SELECT OBJECT_CONSTRUCT(
    'total_patients', total_patients,
    'avg_cost', ROUND(avg_cost, 2),
    'readmission_rate', ROUND(readmission_rate, 2),
    'median_cost', ROUND(median_cost, 2),
    'cost_variance', ROUND(cost_stddev, 2),
    'latest_activity', latest_visit,
    'total_visits', total_visits
  ) as metrics
  FROM metrics
$$;

Component 4: Chart Generator

Visualization Data Agent

CREATE OR REPLACE FUNCTION chart_generator_agent(chart_type STRING, data_query STRING)
RETURNS VARIANT
LANGUAGE SQL
AS
$$
  -- Generate chart data based on type
  SELECT CASE chart_type
    WHEN 'bar_chart' THEN
      (SELECT OBJECT_CONSTRUCT(
        'chart_type', 'bar',
        'labels', ARRAY_AGG(diagnosis),
        'values', ARRAY_AGG(patient_count),
        'title', 'Patient Distribution by Diagnosis'
      )
      FROM (
        SELECT diagnosis, COUNT(*) as patient_count
        FROM PATIENT_VISITS
        GROUP BY diagnosis
        ORDER BY patient_count DESC
      ))
    
    WHEN 'time_series' THEN
      (SELECT OBJECT_CONSTRUCT(
        'chart_type', 'line',
        'labels', ARRAY_AGG(visit_month),
        'values', ARRAY_AGG(avg_cost),
        'title', 'Average Treatment Cost Over Time'
      )
      FROM (
        SELECT 
          TO_VARCHAR(visit_date, 'YYYY-MM') as visit_month,
          AVG(treatment_cost) as avg_cost
        FROM PATIENT_VISITS
        GROUP BY visit_month
        ORDER BY visit_month
      ))
    
    ELSE
      OBJECT_CONSTRUCT('error', 'Invalid chart type')
  END as chart_data
$$;

Component 5: Semantic Model Agent

Business Logic Layer

CREATE OR REPLACE FUNCTION semantic_model_agent(entity STRING, operation STRING)
RETURNS VARIANT
LANGUAGE SQL
AS
$$
  SELECT CASE entity
    WHEN 'patient_risk' THEN
      -- Risk scoring based on clinical and cost factors
      (SELECT OBJECT_CONSTRUCT(
        'high_risk_patients', ARRAY_AGG(OBJECT_CONSTRUCT(
          'patient_id', patient_id,
          'risk_score', risk_score,
          'reason', reason
        )),
        'total_count', COUNT(*)
      )
      FROM (
        SELECT 
          patient_id,
          -- Composite risk scoring model
          (CASE WHEN readmission THEN 50 ELSE 0 END +
           CASE WHEN treatment_cost > 3000 THEN 30 ELSE 0 END +
           CASE WHEN DATEDIFF('day', visit_date, CURRENT_DATE()) < 30 THEN 20 ELSE 0 END
          ) as risk_score,
          CASE 
            WHEN readmission AND treatment_cost > 3000 THEN 'Multiple risk factors: readmission + high cost'
            WHEN readmission THEN 'Readmission history'
            WHEN treatment_cost > 3000 THEN 'High treatment cost'
            ELSE 'Recent visit'
          END as reason
        FROM PATIENT_VISITS
        HAVING risk_score > 50
      ))
    
    WHEN 'cost_analysis' THEN
      -- Cost analysis with diagnosis breakdown
      (SELECT OBJECT_CONSTRUCT(
        'by_diagnosis', OBJECT_CONSTRUCT(
          'diagnosis', ARRAY_AGG(diagnosis),
          'avg_cost', ARRAY_AGG(avg_cost),
          'patient_count', ARRAY_AGG(patient_count)
        ),
        'total_spend', SUM(total_cost)
      )
      FROM (
        SELECT 
          diagnosis,
          AVG(treatment_cost) as avg_cost,
          COUNT(*) as patient_count,
          SUM(treatment_cost) as total_cost
        FROM PATIENT_VISITS
        GROUP BY diagnosis
      ))
    
    ELSE
      OBJECT_CONSTRUCT('error', 'Unknown entity type')
  END as result
$$;

Component 6: Response Synthesizer

Coordinator Agent with Advanced Model

CREATE OR REPLACE FUNCTION coordinator_agent(user_query STRING)
RETURNS TABLE (final_response STRING, sources VARIANT, execution_time FLOAT)
LANGUAGE SQL
AS
$$
  WITH agent_results AS (
    -- Execute all agents in parallel on Gen2 warehouse
    SELECT
      -- Document retrieval using Cortex Search
      (SELECT ARRAY_AGG(OBJECT_CONSTRUCT(
        'text', chunk_text,
        'score', relevance_score,
        'source', source
      )) 
      FROM TABLE(document_retrieval_tool(user_query))
      ) as doc_results,
      
      -- SQL execution via Cortex Analyst
      (SELECT ARRAY_AGG(result)
       FROM TABLE(sql_generator_agent(user_query))
      ) as sql_results,
      
      -- Metrics calculation
      metrics_calculator_agent('summary', OBJECT_CONSTRUCT()) as metrics,
      
      -- Chart data generation
      chart_generator_agent('bar_chart', '') as chart_data,
      
      -- Semantic model analysis
      semantic_model_agent('patient_risk', 'analyze') as semantic_results,
      
      SYSDATE() as start_time
  ),
  synthesized_response AS (
    SELECT
      -- Use stronger model for final synthesis
      SNOWFLAKE.CORTEX.AI_COMPLETE(
        'llama3.1-405b',
        CONCAT(
          'You are an AI assistant synthesizing multi-agent RAG results.\n\n',
          'User Question: ', user_query, '\n\n',
          'Document Context: ', TO_JSON(doc_results), '\n\n',
          'Structured Data: ', TO_JSON(sql_results), '\n\n',
          'Metrics: ', TO_JSON(metrics), '\n\n',
          'Semantic Analysis: ', TO_JSON(semantic_results), '\n\n',
          'Provide a comprehensive, context-rich response that:\n',
          '1. Directly answers the user question\n',
          '2. Cites specific data points from structured data\n',
          '3. References relevant document context\n',
          '4. Includes key metrics and insights\n',
          '5. Is production-ready and actionable'
        )
      ) as final_response,
      OBJECT_CONSTRUCT(
        'documents', doc_results,
        'structured_data', sql_results,
        'metrics', metrics,
        'charts', chart_data,
        'semantic_analysis', semantic_results
      ) as all_sources,
      DATEDIFF('millisecond', start_time, SYSDATE()) / 1000.0 as exec_time
    FROM agent_results
  )
  SELECT 
    final_response,
    all_sources as sources,
    exec_time as execution_time
  FROM synthesized_response
$$;

Testing the Multi-Agent System

Sample Query

-- Test the complete multi-agent RAG system
SELECT * FROM TABLE(
  coordinator_agent(
    'Show me ER visits for hypertensive patients in ZIP 10453 with treatment guidelines'
  )
);

Performance Optimization

Gen2 Warehouse Benefits

-- Monitor warehouse performance
SELECT 
  query_id,
  query_text,
  execution_time,
  warehouse_size,
  query_type
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE warehouse_name = 'MULTIAGENT_WH'
  AND start_time >= DATEADD('hour', -1, CURRENT_TIMESTAMP())
ORDER BY execution_time DESC;

Concurrent Agent Execution

Query Performance Metrics

Production Deployment

Monitoring Dashboard

-- Create monitoring view
CREATE OR REPLACE VIEW AGENT_PERFORMANCE AS
SELECT
  DATE_TRUNC('hour', start_time) as hour,
  COUNT(*) as query_count,
  AVG(total_elapsed_time/1000) as avg_time_sec,
  MAX(total_elapsed_time/1000) as max_time_sec,
  SUM(credits_used_cloud_services) as total_credits
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE warehouse_name = 'MULTIAGENT_WH'
  AND start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
GROUP BY hour
ORDER BY hour DESC;

Troubleshooting

Common Issues

-- Optimize search service
ALTER CORTEX SEARCH SERVICE medical_docs_search 
SET TARGET_LAG = '30 seconds';

-- Increase warehouse size for complex queries
ALTER WAREHOUSE MULTIAGENT_WH 
SET WAREHOUSE_SIZE = 'LARGE';

-- Adjust search parameters
-- Modify limit and filters in document_retrieval_tool

Conclusion

What You Built

Resources

Next Steps

Key Takeaways