Snowflake 作为智能体运行时：从静态管道迈向自主数据系统 ｜ 技术实践

人工定义步骤 → 系统执行步骤

人工定义目标与工具 → 智能体自主规划步骤 → 系统执行

/********************************************************************************* * SNOWFLAKE AGENT RUNTIME DEMO: Product Optimization Logic * This script creates the Database, Security, Demo Data, and AI Views. *********************************************************************************/--- 1. INFRASTRUCTURE & SECURITY SETUP ----- Execute as ACCOUNTADMIN to handle role and warehouse creationUSE ROLE ACCOUNTADMIN;-- Create a dedicated Warehouse for AI workloadsCREATE WAREHOUSE IF NOT EXISTS product_agent_wh     WITH WAREHOUSE_SIZE = 'SMALL'     AUTO_SUSPEND = 60     AUTO_RESUME = TRUE;    -- Create the Agent RoleCREATE ROLE IF NOT EXISTS product_agent_role;GRANT USAGE ON WAREHOUSE product_agent_wh TO ROLE product_agent_role;-- Grant access to Snowflake Cortex (AI Functions)GRANT ROLE SNOWFLAKE.CORTEX_USER TO ROLE product_agent_role;-- Assign the role to current user (replace 'CURRENT_USER' if needed)SET my_user_var = CURRENT_USER();GRANT ROLE product_agent_role TO USER IDENTIFIER($my_user_var);--- 2. DATABASE & SCHEMA SETUP ---CREATE OR REPLACE DATABASE product_intelligence;CREATE OR REPLACE SCHEMA analytics;-- Give the Agent role full access to the new schemaGRANT ALL PRIVILEGES ON DATABASE product_intelligence TO ROLE product_agent_role;GRANT ALL PRIVILEGES ON SCHEMA product_intelligence.analytics TO ROLE product_agent_role;-- Switch to the Agent Role for the remainder of the setupUSE ROLE product_agent_role;USE WAREHOUSE product_agent_wh;USE DATABASE product_intelligence;USE SCHEMA analytics;--- 3. TABLE CREATION ----- Product CatalogCREATE OR REPLACE TABLE products (    product_id INT,    product_name STRING,    category STRING,    price FLOAT);-- Sales Transactions (Weekly Aggregates)CREATE OR REPLACE TABLE sales (    product_id INT,    week DATE,    sales_amount FLOAT);-- Customer Reviews (Unstructured Data)CREATE OR REPLACE TABLE reviews (    product_id INT,    rating INT,    review_text STRING);--- 4. UPDATED DATA INGESTION (Demo Set) ---INSERT INTO products VALUES     (101, 'Eco-Friendly Yoga Mat', 'Fitness', 50.00),    (102, 'Old-Gen Wireless Earbuds', 'Electronics', 85.00),    (103, 'Basic Office Desk', 'Furniture', 120.00),    -- New additions    (104, 'Titanium Coffee Press', 'Kitchen', 195.00),   -- High Price Failure    (105, 'Premium Leather Keychain', 'Fashion', 45.00), -- High Price Failure    (106, 'Device Stand X-99', 'Electronics', 15.00);    -- Name/Boring Failure-- Updated Sales: Adding drops for the new itemsINSERT INTO sales VALUES     (101, '2026-02-01', 5000.00), (101, '2026-02-08', 5200.00),    (102, '2026-02-01', 8000.00), (102, '2026-02-08', 1500.00),    (103, '2026-02-01', 3000.00), (103, '2026-02-08', 2950.00),    (104, '2026-02-01', 4000.00), (104, '2026-02-08', 800.00),       (105, '2026-02-01', 1200.00), (105, '2026-02-08', 200.00),       (106, '2026-02-01', 900.00),  (106, '2026-02-08', 850.00);-- Updated FeedbackINSERT INTO reviews VALUES     (101, 5, 'Best mat I have ever owned! Grip is perfect.'),    (102, 1, 'Battery dies after 10 minutes. Absolute trash.'),    (104, 2, 'It looks beautiful, but $195 for a coffee press is insane. Way too high.'),    (104, 1, 'Does not work.'),    (105, 2, 'It is just a small piece of leather. $45 is a high price for a keychain.'),    (105, 3, 'Bad quality.'),    (106, 3, 'Functional, but the name X-99 is so boring. Hard to find in search.');USE ROLE ACCOUNTADMIN;-- Create the integration to allow the Agent to send emailsCREATE OR REPLACE NOTIFICATION INTEGRATION product_alert_email  TYPE = EMAIL  ENABLED = TRUE  ALLOWED_RECIPIENTS = ('product-team@yourcompany.com');   -- TODO: Replace with your actual email  -- Grant the agent role permission to use this email toolGRANT USAGE ON INTEGRATION product_alert_email TO ROLE product_agent_role;

用户↓智能体（大语言模型）↓工具层↓Snowflake↓数据 + 计算资源

用户请求↓智能体解析目标↓智能体检视可用工具↓智能体生成 SQL 代码↓Snowflake 执行查询↓智能体评估返回结果↓智能体决策下一步动作↓最终输出或执行操作

CREATE OR REPLACE VIEW product_improvement_insights AS -- might be better to have a tableSELECT    p.product_id,    p.product_name,    p.price,    s.week,    SUM(s.sales_amount) AS total_sales,    -- Calculate Sales Delta (the 'drop')    SUM(s.sales_amount) - LAG(SUM(s.sales_amount)) OVER (        PARTITION BY p.product_id ORDER BY s.week    ) AS sales_delta,    -- AI-driven Sentiment and Summarization    AVG(rating) AS avg_rating,    SNOWFLAKE.CORTEX.SENTIMENT(LISTAGG(r.review_text, ' ')) AS sentiment_score,    SNOWFLAKE.CORTEX.SUMMARIZE(LISTAGG(r.review_text, ' ')) AS feedback_summaryFROM products pJOIN sales s ON p.product_id = s.product_idJOIN reviews r ON p.product_id = r.product_idGROUP BY ALL;

-- THE AGENT EXECUTION (Reasoning over the new data) ---SELECTproduct_name,    price,    sales_delta,    sentiment_score,    feedback_summary,    SNOWFLAKE.CORTEX.COMPLETE('llama3-70b',         CONCAT('Context: The product ', product_name, ' (Price: $', price, ') has a sales drop of $', sales_delta,                '. Customer feedback: ', feedback_summary,                '. Task: If the price is the main complaint, suggest a specific new price. If the name is the issue, suggest a catchy new name. If the probelm is with the product quality send an email to the product team with short explanation of the issue' )    ) AS agent_recommendationFROM product_improvement_insightsWHERE sales_delta < 0 ORDER BY sales_delta ASC;

USE ROLE product_agent_role;CREATE OR REPLACE TASK product_optimization_task    WAREHOUSE = product_agent_wh    SCHEDULE = '10 MINUTE'ASEXECUTE IMMEDIATE $$DECLARE    -- Cursor for underperforming productsFOR record IN c1 DO        -- 1. Assign cursor values to flat variables        v_product_name := record.product_name;        v_sales_delta := record.sales_delta;        v_feedback := record.feedback_summary;        -- 2. NEW: Check if we have already handled this product today        -- This prevents duplicate emails and logs every 10 minutes        LET already_logged INTEGER := (            SELECT COUNT(*)             FROM agent_action_log             WHERE product_name = :v_product_name            -- Optional: Only skip if logged in the last 24 hours            -- AND action_date > DATEADD('day', -1, CURRENT_TIMESTAMP())        );        IF (already_logged = 0) THEN            -- 3. Build the prompt            v_prompt_text := 'Context: ' || v_product_name ||               ' sales dropped by $' || v_sales_delta ||               '. Feedback: ' || v_feedback ||               '. Task: Determine if this is a Product Quality issue. ' ||              'If YES, write an email to the product team about the issue ' ||              'starting with "SUBJECT: PRODUCT ISSUE". ' ||              'If NO, suggest a fix starting with "FIX:"' ||              'You can suggest a new price or a new name for the product';            -- 4. Call Cortex LLM            v_agent_output := (SELECT SNOWFLAKE.CORTEX.COMPLETE('llama3-70b',                 :v_prompt_text));            -- Here is a good place where we will add our guardrails              -- 5. Take Action: Send Email            IF (v_agent_output ILIKE '%SUBJECT: PRODUCT ISSUE%') THEN                CALL SYSTEM$SEND_EMAIL(                    'product_alert_email',                    'product-team@yourcompany.com',     -- TODO: Change to your email!!                    'Urgent Action Required: ' || :v_product_name,                    :v_agent_output                );            END IF;            -- 6. Log the results            INSERT INTO agent_action_log (product_name, action_date, details)            VALUES (:v_product_name, CURRENT_TIMESTAMP(), :v_agent_output);                END IF; -- End of the "already logged" check            END FOR;END;$$;-- Activate the TaskALTER TASK product_optimization_task RESUME;-- 1. Switch to Account Admin to grant the global privilegeUSE ROLE ACCOUNTADMIN;-- 2. Grant the ability to execute tasks to your specific roleGRANT EXECUTE TASK ON ACCOUNT TO ROLE product_agent_role;-- 3. Switch back to your agent roleUSE ROLE product_agent_role;-- 4. Now you can manually trigger it for testingEXECUTE TASK product_optimization_task;USE ROLE product_agent_role;-- You canmonitor the task by-- Check the last 10 runs of your taskSELECT *  FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(    TASK_NAME => 'PRODUCT_OPTIMIZATION_TASK',    RESULT_LIMIT => 10      ))ORDER BY QUERY_START_TIME DESC;

SELECT     product_name,    -- action_date,    -- Label if it was an Email or a Price/Name fix    CASE         WHEN details ILIKE '%SUBJECT:%' THEN '📩 Email Sent'        WHEN details ILIKE '%FIX:%' THEN '🏷️ Price/Name Update'        ELSE '🤖 General Advice'    END AS category,    details AS agent_thought_processFROM agent_action_logGROUP BY product_name, details;

/********************************************************************************* * SUSPEND AGENT: Stop the 10-minute cycle *********************************************************************************/USE ROLE product_agent_role;-- Stop the task from executing furtherALTER TASK product_optimization_task SUSPEND;-- Verify the state is now 'suspended'DESCRIBE TASK product_optimization_task;

/********************************************************************************* * TEAR DOWN: Clean up the Snowflake Agent Runtime Demo *********************************************************************************/USE ROLE ACCOUNTADMIN;-- 1. Stop and Remove the Task (Prevents further compute charges)DROP TASK IF EXISTS product_intelligence.analytics.product_optimization_task;-- 2. Drop the Database (This removes all tables, views, and logs)DROP DATABASE IF EXISTS product_intelligence;-- 3. Remove the Email Integration-- (Note: Only do this if no other systems are using it)DROP INTEGRATION IF EXISTS product_alert_email;-- 4. Clean up Security and ComputeDROP WAREHOUSE IF EXISTS product_agent_wh;DROP ROLE IF EXISTS product_agent_role;-- 5. Confirm CleanupSHOW DATABASES LIKE 'PRODUCT_INTELLIGENCE';SHOW WAREHOUSES LIKE 'PRODUCT_AGENT_WH';

                ┌─────────────────────┐                │      用户 / 应用      │                └──────────┬──────────┘                           │                           ▼                ┌─────────────────────┐                │      AI 智能体       │                │  （推理层）           │                └──────────┬──────────┘                           │          ┌────────────────┼──────────────────┐          ▼                ▼                  ▼┌──────────────┐  ┌──────────────┐  ┌──────────────┐│ Cortex LLM   │  │      Cortex  │  │ 工具层        ││ COMPLETE()   │  │     Analyst  │  │（存储过程）    │└──────┬───────┘  └──────┬───────┘  └──────┬───────┘       │                 │                 │        └────────────────┴─────────────────┘                         │                         ▼                ┌─────────────────────┐                │       Snowflake     │                │     执行层           │                └──────────┬──────────┘                           │          ┌───────────── ─ ┼─────────────────┐          ▼                ▼                 ▼       结构化数据       使用元数据         非结构化数据