使用 Account Usage 管理 Cortex AI 函数成本¶

Snowflake Cortex AI 函数（AI_COMPLETE、AI_SUMMARIZE、AI_TRANSLATE、AI_SENTIMENT 等）根据词元或页面使用情况计算消耗的 Credit。如果没有监控和控制，使用这些函数的成本可能会迅速上升，原因如下：

未优化的提示会生成过多词元
长时间运行或失控的查询
缺乏针对每个用户的支出限额
使用模式的可见性不足

本主题提供策略建议，帮助您监控、管理和控制与 Snowflake Cortex AI 函数相关的成本。使用 CORTEX_AI_FUNCTIONS_USAGE_HISTORY 视图，您可以跟踪使用模式并实施自动成本控制。这些方式可以帮助您监控使用情况，在超出支出限额时接收警报，根据每月限额控制对函数的访问，并阻止失控的查询。

使用历史记录视图¶

SNOWFLAKE.ACCOUNT_USAGE CORTEX_AI_FUNCTIONS_USAGE_HISTORY 视图通过 SQL 提供所有 Cortex AI 函数的详细遥测数据。虽然最短可能会在函数开始执行后的 10 分钟内提供数据，但该视图的最大延迟为 60 分钟。有关此视图的详细信息，请参阅 CORTEX_AI_FUNCTIONS_USAGE_HISTORY 视图。

基本使用情况监控¶

以下查询可帮助您了解 AI 函数使用模式。自行定期运行这些函数，或将其集成到仪表板中以持续监控。

按功能和模型划分的每日 Credit 使用量¶

跟踪每日支出趋势，以确定使用量峰值，并了解哪些函数和模型消耗的 Credit 最多。

SELECT
    DATE_TRUNC('day', START_TIME) AS usage_date,
    FUNCTION_NAME,
    MODEL_NAME,
    SUM(CREDITS) AS total_credits,
    COUNT(DISTINCT QUERY_ID) AS query_count
FROM SNOWFLAKE.ACCOUNT_USAGE.CORTEX_AI_FUNCTIONS_USAGE_HISTORY
WHERE START_TIME >= DATEADD('day', -30, CURRENT_TIMESTAMP())
GROUP BY 1, 2, 3
ORDER BY usage_date DESC, total_credits DESC;

按用户统计的每月 Credit 使用量¶

确定主要使用者并跟踪每个用户在一段时间内的支出。此查询会与 USERS 视图联接，以提供用户详细信息，包括电子邮件和默认角色，以便于识别和跟进。

SELECT
    DATE_TRUNC('month', h.START_TIME) AS usage_month,
    u.NAME AS user_name,
    u.EMAIL,
    u.DEFAULT_ROLE,
    SUM(h.CREDITS) AS total_credits,
    COUNT(DISTINCT h.QUERY_ID) AS query_count
FROM SNOWFLAKE.ACCOUNT_USAGE.CORTEX_AI_FUNCTIONS_USAGE_HISTORY h
JOIN SNOWFLAKE.ACCOUNT_USAGE.USERS u
    ON h.USER_ID = u.USER_ID
WHERE h.START_TIME >= DATEADD('month', -3, CURRENT_TIMESTAMP())
GROUP BY 1, 2, 3, 4
ORDER BY usage_month DESC, total_credits DESC;

成本控制¶

定义自动机制来检测过度支出并采取纠正措施。这些查询可以彼此独立使用，也可以组合使用以进行全面的成本治理。

账户级每月支出警报¶

设置自动警报，监控整个账户每月的 AI 函数 Credit 使用总量。当支出超过定义的阈值时，警报会向指定管理员发送电子邮件通知。设置警报需要满足以下先决条件：

ACCOUNTADMIN 角色或相应权限，以创建通知集成和警报
用于执行警报条件检查的仓库
警报收件人经过验证的电子邮件地址

首先，创建一项通知集成（如果还没有）。此示例会替换任何名为 ai_cost_alerts 的现有集成。

CREATE OR REPLACE NOTIFICATION INTEGRATION ai_cost_alerts
    TYPE = EMAIL
    ENABLED = TRUE
    ALLOWED_RECIPIENTS = ('admin@company.com', 'finops@company.com')

接下来，创建一个表来跟踪每个月的警报发送时间。这有助于防止一个月内重复发送警报。

CREATE TABLE IF NOT EXISTS AI_FUNCTIONS_ALERT_STATE (
    ALERT_NAME VARCHAR NOT NULL,
    ALERT_MONTH DATE NOT NULL,
    SENT_AT TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP(),
    CREDITS_AT_ALERT NUMBER(38,6),
    PRIMARY KEY (ALERT_NAME, ALERT_MONTH)
);

现在，创建一个存储过程来检查本月是否已发送警报、记录警报状态，并发送电子邮件通知。

CREATE OR REPLACE PROCEDURE SEND_MONTHLY_SPEND_ALERT(P_THRESHOLD FLOAT)
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
EXECUTE AS CALLER
AS
$$
    // Check if alert already sent this month
    var check_sent = snowflake.execute({
        sqlText: `SELECT COUNT(*) AS cnt FROM AI_FUNCTIONS_ALERT_STATE
                WHERE ALERT_NAME = 'monthly_spend'
                AND ALERT_MONTH = DATE_TRUNC('month', CURRENT_DATE())`
    });
    check_sent.next();
    var already_sent = check_sent.getColumnValue(1);

    if (already_sent > 0) {
        return 'Alert already sent for this month';
    }

    // Get current spend
    var spend_result = snowflake.execute({
        sqlText: `SELECT COALESCE(SUM(CREDITS), 0) AS total
                FROM SNOWFLAKE.ACCOUNT_USAGE.CORTEX_AI_FUNCTIONS_USAGE_HISTORY
                WHERE START_TIME >= DATE_TRUNC('month', CURRENT_TIMESTAMP())`
    });
    spend_result.next();
    var v_credits = spend_result.getColumnValue(1);

    // Check threshold
    if (v_credits <= P_THRESHOLD) {
        return 'Threshold not exceeded. Current: ' + v_credits + ' / ' + P_THRESHOLD;
    }

    // Record alert
    snowflake.execute({
        sqlText: `INSERT INTO AI_FUNCTIONS_ALERT_STATE (ALERT_NAME, ALERT_MONTH, CREDITS_AT_ALERT)
                VALUES ('monthly_spend', DATE_TRUNC('month', CURRENT_DATE()), ?)`,
        binds: [v_credits]
    });

    // Send email - update the recipient email address
    snowflake.execute({
        sqlText: `CALL SYSTEM$SEND_EMAIL(
            'ai_cost_alerts',
            'admin@company.com',
            'AI Functions Monthly Spend Alert',
            'Monthly AI Function credit consumption has exceeded the threshold.\\n\\n' ||
            'Current spend: ' || ${v_credits}::VARCHAR || ' credits\\n' ||
            'Threshold: ' || ${P_THRESHOLD}::VARCHAR || ' credits\\n\\n' ||
            'Please review usage accordingly.'
        )`
    });

    return 'Alert sent. Credits: ' + v_credits;
$$;

最后，创建警报，根据每小时支出阈值检查使用情况，并在需要时调用过程来发送通知。您应该将 1000 个 Credit 的限额（在以下示例中出现在两个位置）调整为所需的阈值。

CREATE OR REPLACE ALERT ai_functions_monthly_spend_alert
    WAREHOUSE = <your_warehouse>
    SCHEDULE = 'USING CRON 0 * * * * UTC'  -- Runs every hour
    IF (EXISTS (
        SELECT 1
        FROM SNOWFLAKE.ACCOUNT_USAGE.CORTEX_AI_FUNCTIONS_USAGE_HISTORY
        WHERE START_TIME >= DATE_TRUNC('month', CURRENT_TIMESTAMP())
        HAVING SUM(CREDITS) > 1000  -- adjust the limit accordingly
    ))
    THEN
        CALL SEND_MONTHLY_SPEND_ALERT(1000);  -- please adjust the limit accordingly

-- enable the alert
ALTER ALERT ai_functions_monthly_spend_alert RESUME;

小技巧

为了进行测试，请先将限额设置为 0，以立即触发警报。确认警报按预期工作后，使用所需阈值重新创建警报。

使用阈值 0 进行测试后，运行以下 SQL 以允许在当月再次触发警报。

DELETE FROM AI_FUNCTIONS_ALERT_STATE
WHERE ALERT_NAME = 'monthly_spend'
AND ALERT_MONTH = DATE_TRUNC('month', CURRENT_DATE());

您可以通过查询警报历史记录和警报状态表，来确保警报正在运行，如下所示：

-- Make sure alert exists
SHOW ALERTS LIKE 'ai_functions_monthly_spend_alert';

-- Check alert history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP()),
    ALERT_NAME => 'ai_functions_monthly_spend_alert'
))
ORDER BY SCHEDULED_TIME DESC;

-- Check which months have had alerts sent
SELECT * FROM AI_FUNCTIONS_ALERT_STATE ORDER BY ALERT_MONTH DESC;

每用户每月支出限额¶

此示例会实施每用户每月支出限额。用户被授予专用的自定义 AI_FUNCTIONS_USER_ROLE，该角色提供对 Cortex AI 函数的访问权限。表存储单个用户的每月词元预算。当用户超出当月预算时，一个每小时任务会通过移除 AI_FUNCTIONS_USER_ROLE，撤销其对 AI 函数的访问权限。每月任务会在下月初恢复角色。

重要

默认情况下，所有用户都有权访问 AI 函数（以及其他 Snowflake Cortex 功能），因为 SNOWFLAKE.CORTEX_USER 数据库角色会获授 PUBLIC 角色。要强制执行每用户限额，您必须撤销来自PUBLIC 的 SNOWFLAKE.CORTEX_USER，并仅通过AI_FUNCTIONS_USER_ROLE 授予。使用以下 SQL 撤销来自 PUBLIC 的角色：

REVOKE DATABASE ROLE SNOWFLAKE.CORTEX_USER FROM ROLE PUBLIC;

确保仅授予所有需要访问 Cortex 功能的用户 AI_FUNCTIONS_USER_ROLE。使用包含 SNOWFLAKE.CORTEX_USER 的任何其他角色，都允许用户绕过本示例中实施的支出限额控制。在某些情况下，您可以使用更具体的角色；例如，可向只需要访问 Cortex Analyst 的用户授予 SNOWFLAKE.CORTEX_ANALYST_USER 角色，而不是 SNOWFLAKE.CORTEX_USER。

要设置每用户支出限额，请先创建一个角色来控制对 AI 函数的访问，以便允许此访问权限与其他权限分开管理。

-- Create a role specifically for AI Function access
CREATE ROLE IF NOT EXISTS AI_FUNCTIONS_USER_ROLE;

-- Grant necessary privileges to the role
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE AI_FUNCTIONS_USER_ROLE;

-- Grant usage on warehouse
GRANT USAGE ON WAREHOUSE AI_FUNCTIONS_WAREHOUSE TO ROLE AI_FUNCTIONS_USER_ROLE;

现在，设置访问控制表，该表跟踪哪些用户具有 AI 函数访问权限、其个人支出限额及其撤销历史记录。它充当自动监控和访问恢复过程的事实来源。

CREATE TABLE IF NOT EXISTS AI_FUNCTIONS_ACCESS_CONTROL (
    USER_NAME VARCHAR NOT NULL,
    USER_ID NUMBER,
    GRANTED_AT TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP(),
    MONTHLY_CREDIT_LIMIT NUMBER(38,6) DEFAULT 100,  -- adjust the limit accordingly
    IS_ACTIVE BOOLEAN DEFAULT TRUE,
    REVOKED_AT TIMESTAMP_LTZ,
    REVOCATION_REASON VARCHAR,
    PRIMARY KEY (USER_NAME)
);

接下来，创建一个存储过程以向用户授予 AI 函数访问权限，并在访问控制表中注册其支出限额。该代码会从 Account Usage 视图中查找用户的 ID，以启用监控查询中的高效联接。

CREATE OR REPLACE PROCEDURE GRANT_AI_FUNCTIONS_ACCESS(
    P_USER_NAME VARCHAR,
    P_MONTHLY_LIMIT NUMBER(38,6) DEFAULT 100  -- adjust the limit accordingly
)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_user_id NUMBER;
BEGIN
    -- Look up USER_ID from account usage
    SELECT USER_ID INTO :v_user_id
    FROM SNOWFLAKE.ACCOUNT_USAGE.USERS
    WHERE NAME = :P_USER_NAME
    LIMIT 1;

    -- Grant the AI Functions role to the user
    EXECUTE IMMEDIATE 'GRANT ROLE AI_FUNCTIONS_USER_ROLE TO USER ' || P_USER_NAME;

    -- Register or update the user in the access control table
    MERGE INTO AI_FUNCTIONS_ACCESS_CONTROL tgt
    USING (SELECT :P_USER_NAME AS USER_NAME) src
    ON tgt.USER_NAME = src.USER_NAME
    WHEN MATCHED THEN
        UPDATE SET
            USER_ID = :v_user_id,
            IS_ACTIVE = TRUE,
            MONTHLY_CREDIT_LIMIT = :P_MONTHLY_LIMIT,
            GRANTED_AT = CURRENT_TIMESTAMP(),
            REVOKED_AT = NULL,
            REVOCATION_REASON = NULL
    WHEN NOT MATCHED THEN
        INSERT (USER_NAME, USER_ID, MONTHLY_CREDIT_LIMIT, IS_ACTIVE)
        VALUES (:P_USER_NAME, :v_user_id, :P_MONTHLY_LIMIT, TRUE);

    RETURN 'Access granted to ' || P_USER_NAME || ' with monthly limit of ' || P_MONTHLY_LIMIT || ' credits';
END;
$$;

使用此存储过程将用户及其 Credit 配额添加到访问控制表中。

CALL GRANT_AI_FUNCTIONS_ACCESS('ALICE', 1000);  -- grants access to user ALICE with a monthly limit of 100 credits
CALL GRANT_AI_FUNCTIONS_ACCESS('BOB', 2000);    -- grants access to user BOB with a monthly limit of 200 credits

接下来创建每月访问刷新任务。此任务在每月的第一天运行，以恢复所有授权用户的 AI 函数访问权限。当用户的访问权限因超出其上个月的限额而被撤销时，此任务将为他们授予新月份的新预算。

-- Create a procedure to re-grant access to all entitled users
CREATE OR REPLACE PROCEDURE GRANT_ALL_ENTITLED_USERS()
RETURNS TABLE (USER_NAME VARCHAR, CREDIT_LIMIT NUMBER, ACTION VARCHAR)
LANGUAGE SQL
AS
$$
DECLARE
    result RESULTSET;
BEGIN
    result := (
        SELECT
            USER_NAME,
            MONTHLY_CREDIT_LIMIT AS CREDIT_LIMIT,
            'GRANTED' AS ACTION
        FROM AI_FUNCTIONS_ACCESS_CONTROL
    );

    -- Re-grant access for each entitled user
    FOR rec IN result DO
        CALL GRANT_AI_FUNCTIONS_ACCESS(rec.USER_NAME, rec.CREDIT_LIMIT);
    END FOR;

    RETURN TABLE(result);
END;
$$;

-- Create a task to run on the 1st of each month at midnight UTC
CREATE OR REPLACE TASK MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH
    WAREHOUSE = <your_warehouse>
    SCHEDULE = 'USING CRON 0 0 1 * * UTC'  -- 1st day of each month at 00:00 UTC
AS
    CALL GRANT_ALL_ENTITLED_USERS();

-- Enable the task
ALTER TASK MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH RESUME;

-- Run once initially to populate grantees
CALL GRANT_ALL_ENTITLED_USERS();

-- Verify task status
SHOW TASKS LIKE 'MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH';

最后，创建一个每小时任务来监控用户支出，并撤销任何超出每月限额的用户的访问权限。

-- Create a procedure to re-grant access to all entitled users
CREATE OR REPLACE PROCEDURE GRANT_ALL_ENTITLED_USERS()
RETURNS TABLE (USER_NAME VARCHAR, CREDIT_LIMIT NUMBER, ACTION VARCHAR)
LANGUAGE SQL
AS
$$
DECLARE
    result RESULTSET;
BEGIN
    result := (
        SELECT
            USER_NAME,
            MONTHLY_CREDIT_LIMIT AS CREDIT_LIMIT,
            'GRANTED' AS ACTION
        FROM AI_FUNCTIONS_ACCESS_CONTROL
    );

    -- Re-grant access for each entitled user
    FOR rec IN result DO
        CALL GRANT_AI_FUNCTIONS_ACCESS(rec.USER_NAME, rec.CREDIT_LIMIT);
    END FOR;

    RETURN TABLE(result);
END;
$$;

-- Create a task to run on the 1st of each month at midnight UTC
CREATE OR REPLACE TASK MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH
    WAREHOUSE = <your_warehouse>
    SCHEDULE = 'USING CRON 0 0 1 * * UTC'  -- 1st day of each month at 00:00 UTC
AS
    CALL GRANT_ALL_ENTITLED_USERS();

-- Enable the task
ALTER TASK MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH RESUME;

-- Run once initially to populate grantees
CALL GRANT_ALL_ENTITLED_USERS();

-- Verify task status
SHOW TASKS LIKE 'MONTHLY_AI_FUNCTIONS_ACCESS_REFRESH';

失控查询检测和取消¶

长时间运行的 AI 函数查询会累积巨额成本。此示例实施了一个自动化系统，用于检测超过 Credit 阈值的查询，并在查询消耗更多资源之前将其取消。将发送包含完整查询详细信息的电子邮件警报。

备注

取消查询后，客户端仍需为取消前使用的所有资源付费。取消失控查询可防止进一步累积成本，但不会退还已花费的 Credit。

此过程会查找过去 48 小时超出 Credit 阈值但仍在运行的 AI 函数查询，将其取消，并向管理员报告。

-- Create a procedure to detect and cancel expensive runaway queries
CREATE OR REPLACE PROCEDURE MONITOR_AND_CANCEL_RUNAWAY_QUERIES(
    P_CREDIT_THRESHOLD NUMBER DEFAULT 50  -- adjust the limit accordingly
)
RETURNS TABLE (
    QUERY_ID VARCHAR,
    USER_NAME VARCHAR,
    FUNCTION_NAME VARCHAR,
    MODEL_NAME VARCHAR,
    CREDITS NUMBER,
    START_TIME TIMESTAMP_LTZ,
    ACTION VARCHAR
)
LANGUAGE SQL
AS
$$
DECLARE
    result RESULTSET;
BEGIN
    -- Find queries from the last 48 hours that exceed the threshold and are still running
    result := (
        SELECT
            h.QUERY_ID,
            u.NAME AS USER_NAME,
            h.FUNCTION_NAME,
            h.MODEL_NAME,
            h.CREDITS,
            h.START_TIME,
            h.ROLE_NAMES,
            h.QUERY_TAG,
            h.WAREHOUSE_ID,
            'CANCELLED' AS ACTION
        FROM SNOWFLAKE.ACCOUNT_USAGE.CORTEX_AI_FUNCTIONS_USAGE_HISTORY h
        LEFT JOIN SNOWFLAKE.ACCOUNT_USAGE.USERS u
            ON h.USER_ID = u.USER_ID
        WHERE h.START_TIME >= DATEADD('hour', -48, CURRENT_TIMESTAMP())
        AND h.CREDITS > :P_CREDIT_THRESHOLD
        AND h.IS_COMPLETED = FALSE
    );

    -- Cancel each runaway query and send alert
    FOR rec IN result DO
        -- Attempt to cancel the query
        BEGIN
            EXECUTE IMMEDIATE 'SELECT SYSTEM$CANCEL_QUERY(''' || rec.QUERY_ID || ''')';
        EXCEPTION
            WHEN OTHER THEN
                NULL;  -- Query may have already completed
        END;

        -- Send alert with query details
        CALL SYSTEM$SEND_EMAIL(
            'ai_cost_alerts',
            'admin@company.com',
            'Runaway AI Query Cancelled - ' || rec.QUERY_ID,
            'A runaway AI Function query has been cancelled due to excessive cost.\n\n' ||
            'Query Details:\n' ||
            '- Query ID: ' || rec.QUERY_ID || '\n' ||
            '- User: ' || COALESCE(rec.USER_NAME, 'Unknown') || '\n' ||
            '- Function: ' || rec.FUNCTION_NAME || '\n' ||
            '- Model: ' || rec.MODEL_NAME || '\n' ||
            '- Credits Used: ' || rec.CREDITS::VARCHAR || '\n' ||
            '- Threshold: ' || :P_CREDIT_THRESHOLD::VARCHAR || '\n' ||
            '- Start Time: ' || rec.START_TIME::VARCHAR || '\n' ||
            '- Roles: ' || COALESCE(rec.ROLE_NAMES::VARCHAR, 'N/A') || '\n' ||
            '- Query Tag: ' || COALESCE(rec.QUERY_TAG, 'N/A') || '\n' ||
            '- Warehouse ID: ' || COALESCE(rec.WAREHOUSE_ID::VARCHAR, 'N/A') || '\n\n' ||
            'Please investigate this query and take appropriate action.'
        );
    END FOR;

    RETURN TABLE(result);
END;
$$;

-- Create a task to monitor and cancel runaway queries every hour
CREATE OR REPLACE TASK MONITOR_RUNAWAY_AI_QUERIES
    WAREHOUSE = <your_warehouse>
    SCHEDULE = 'USING CRON 0 * * * * UTC'  -- Every hour
AS
    CALL MONITOR_AND_CANCEL_RUNAWAY_QUERIES(50);  -- adjust the limit accordingly

-- Enable the task
ALTER TASK MONITOR_RUNAWAY_AI_QUERIES RESUME;

-- Verify task status
SHOW TASKS LIKE 'MONITOR_RUNAWAY_AI_QUERIES';

-- Check task execution history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP()),
    TASK_NAME => 'MONITOR_RUNAWAY_AI_QUERIES'
))
ORDER BY SCHEDULED_TIME DESC;

小技巧

If you already know that some of your queries will run a long time, define a special role for these queries, and then exclude that role from the cancellation logic. For example, to create the role:

CREATE ROLE AI_FUNCTIONS_USER_LONG_RUNNING_ROLE;
GRANT ROLE AI_FUNCTIONS_USER_ROLE TO ROLE AI_FUNCTIONS_USER_LONG_RUNNING_ROLE;
GRANT ROLE AI_FUNCTIONS_USER_LONG_RUNNING_ROLE TO USER LONG_RUNNING_USER;

将以下条件添加到该过程的 WHERE 子句，以排除由具有此角色的用户运行的查询，防止其被取消。

AND NOT ARRAY_CONTAINS(h.ROLE_NAMES, 'AI_FUNCTIONS_USER_LONG_RUNNING_ROLE')

现在，用户可以承担该角色来运行长时间运行的查询，该查询不会被取消：

USE ROLE AI_FUNCTIONS_USER_LONG_RUNNING_ROLE;
-- then start the long-running query

最佳实践¶

在针对 AI 函数使用情况制定成本管理策略时，请牢记以下最佳实践：

从监控入手： 在实施自动化控制之前，请使用基本使用情况监控中的查询建立基本使用模式。
设置保守的初始限额： 从较低的阈值入手，并根据实际使用模式向上调整。
使用查询标签： 鼓励团队使用 QUERY_TAG 会话参数，以启用按项目或团队进行成本归因的功能。
定期审查： 定期审查访问控制表，并根据合法需求调整每用户限额。
测试警报： 在依赖电子邮件通知发出关键警报之前，先验证该功能是否正常运作。
**考虑延迟：**ACCOUNT_USAGE 视图有最长 60 分钟的延迟；在制定您的监控策略时请考虑这一点。