使用 SQL 自动生成对象描述¶

Cortex 支持的对象描述功能允许您使用 Snowflake Cortex 的 COMPLETE 函数，自动为表、视图和列生成描述。此功能利用 Snowflake 托管的大型语言模型 (LLMs) 来评估对象元数据，并在需要时利用示例数据来生成描述。

本主题介绍如何使用存储过程以编程方式生成描述。有关使用 Snowsight 生成描述的信息，请参阅使用 Snowflake Cortex 生成描述。

生成描述¶

AI_GENERATE_TABLE_DESC 存储过程自动为表和视图生成描述。它还可以为该表或视图的列生成描述。

AI_GENERATE_TABLE_DESC 存储过程接受两种实参：

要为其生成描述的表或视图的名称。
一个可选的配置对象，允许您执行以下操作：
- 为指定表或视图的列生成描述。
- 使用表或视图中的样本数据可能会提高列描述的准确性。

示例：生成表描述

CALL AI_GENERATE_TABLE_DESC( 'my_table');

Copy

示例：在不使用样本数据的情况下生成表和列描述

CALL AI_GENERATE_TABLE_DESC(
  'mydb.sch1.hr_data',
  {
    'describe_columns': true,
    'use_table_data': false
  });

Copy

示例：使用样本数据生成视图和列描述以提高准确性

CALL AI_GENERATE_TABLE_DESC(
  'mydb.sch1.v1',
  {
    'describe_columns': true,
    'use_table_data': true
  });

Copy

有关存储过程的完整语法，请参阅 AI_GENERATE_TABLE_DESC。

处理响应¶

AI_GENERATE_TABLE_DESC 存储过程返回一个 JSON 对象，该对象包含生成的描述以及有关表和列的一般信息。在此对象中，description 字段包含生成的描述。

假设您创建了下表：

CREATE OR REPLACE TABLE mydb.sch1.hr_data (fname VARCHAR, age INTEGER);

INSERT INTO hr_data (fname, age)
    VALUES
        ('Thomas',    44),
        ('Katherine', 29),
        ('Lisa',      29);

Copy

对于此表，以下是 AI_GENERATE_TABLE_DESC 返回的 JSON 对象的示例：

{
  "COLUMNS": [
    {
      "database_name": "mydb",
      "description": "The first name of the employee.",
      "name": "FNAME",
      "schema_name": "sch1",
      "table_name": "hr_data"
    }
    {
      "database_name": "mydb",
      "description": "A column holding data of type DecimalType representing age values.",
      "name": "AGE",
      "schema_name": "sch1",
      "table_name": "hr_data"
    },
  ],
  "TABLE": [
    {
      "database_name": "mydb",
      "description": " The table contains records of employee data, specifically demographic information. Each record includes an employee's age and name.",
      "name": "hr_data",
      "schema_name": "sch1"
    }
  ]
}

有关每个 JSON 字段的更多信息，请参阅返回。

将生成的描述设置为注释¶

要将生成的描述设置为表、视图或列的注释，必须手动执行包含 SET COMMENT 参数的 SQL 语句。例如，要为表 t1 保存生成的描述，请执行 ALTER TABLE t1 SET COMMENT = 'ai generated description';。

您可以编写自定义代码来自动生成和保存描述。有关执行此操作的存储过程的示例，请参阅示例。

访问控制要求¶

调用 AI_GENERATE_TABLE_DESC 存储过程的用户必须具有以下权限和角色：

表或视图的 SELECT 权限。
SNOWFLAKE.CORTEX_USER 数据库角色。

该功能的可用性¶

您所在的区域必须支持 Snowflake Cortex 在生成描述时使用的 LLM（例如 Mistral-7b 和 Llama 3.1-8b）。请检查 :ref:` COMPLETE 函数的可用性 <label-cortex_llm_availability>`。如果您所在区域不支持 COMPLETE 函数，则必须启用跨区域推理才能使用此功能。

使用示例数据¶

为列生成描述时，您只能依赖元数据，或者您可以选择使用示例数据来改进 Snowflake Cortex 支持的描述。示例数据是指在使用 Snowflake Cortex 生成描述时评估的特定列中的数据。如果您选择使用示例数据，Snowflake 会使用部分示例数据来生成描述，从而获得更准确的描述。Snowflake 不将示例数据存储为使用情况数据。

成本注意事项¶

生成描述会产生以下费用：

正在使用的仓库消耗的 credit。
因使用 Snowflake Cortex 与较小的 LLMs（例如 Mistral-7b 和 Llama 3.1-8b）而收取的 credit。这些费用作为 AI 服务出现在账单上，其中包括 Snowflake Cortex 的所有使用情况。

限制¶

您无法为超过 1,000 列的对象生成列描述。

法律声明¶

此功能依赖于 COMPLETE 函数来生成建议的对象描述。当用户启动描述生成时，可能会通过 COMPLETE 函数收集使用情况数据。

在用户保存生成的描述之前，Snowflake 不会保留该描述。

有关使用 AI 的更多信息，请参阅 Snowflake AI 和 ML。

示例¶

示例：生成描述并将其设置为注释¶

第 1 步：创建存储过程

以下存储过程执行以下操作：

自动为架构中的所有表（及其列）生成描述。
将这些描述设置为表和列的注释。

CREATE OR REPLACE PROCEDURE DESCRIBE_TABLES_SET_COMMENT (database_name STRING, schema_name STRING,
  set_table_comment BOOLEAN,
  set_column_comment BOOLEAN)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.10'
  PACKAGES=('snowflake-snowpark-python','joblib')
  HANDLER = 'main'
AS
$$
import json
from joblib import Parallel, delayed
import multiprocessing

def generate_descr(session, database_name, schema_name, table, set_table_comment, set_column_comment):
  table_name =  table['TABLE_NAME']
  async_job = session.sql(f"CALL AI_GENERATE_TABLE_DESC( '{database_name}.{schema_name}.{table_name}',{{'describe_columns': true, 'use_table_data': true}})").collect_nowait()
  result = async_job.result()
  output = json.loads(result[0][0])
  columns_ret = output["COLUMNS"]
  table_ret = output["TABLE"][0]

  table_description = table_ret["description"]
  table_name = table_ret["name"]
  database_name = table_ret["database_name"]
  schema_name = table_ret["schema_name"]

  if (set_table_comment):
      table_description = table_description.replace("'", "\\'")
      session.sql(f"""ALTER TABLE {database_name}.{schema_name}.{table_name} SET COMMENT = '{table_description}'""").collect()

  for column in columns_ret:
      column_description = column["description"];
      column_name = column["name"];
      if not column_name.isupper():
        column_name = '"' + column_name + '"'

      if (set_column_comment):
          column_description = column_description.replace("'", "\\'")
          session.sql(f"""ALTER TABLE  {database_name}.{schema_name}.{table_name} MODIFY COLUMN {column_name}  COMMENT '{column_description}'""").collect()

  return 'Success';

def main(session, database_name, schema_name, set_table_comment, set_column_comment):

    schema_name = schema_name.upper()
    database_name = database_name.upper()
    tablenames = session.sql(f"""SELECT table_name
                      FROM {database_name}.information_schema.tables
                      WHERE table_schema = '{schema_name}'
                      AND table_type = 'BASE TABLE'""").collect()
    try:
        Parallel(n_jobs=multiprocessing.cpu_count(), backend="threading")(
                delayed(generate_descr)(
                    session,
                    database_name,
                    schema_name,
                    table,
                    set_table_comment,
                    set_column_comment,
                ) for table in tablenames
            )
        return 'Success'
    except Exception as e:
        # Catch and return the error message
        return f"An error occurred: {str(e)}"
$$;

Copy

第 2 步：调用存储过程

假设您的架构已命名为 my_db.sch1，请按如下方式调用存储过程，以为表和列生成描述：

CALL describe_tables_set_comment('my_db', 'sch1', true, true);

Copy

您可以运行 DESC TABLE 命令来验证生成的描述是否已设置为表的注释。

示例：生成描述并将其保存到目录表中¶

第 1 步：创建存储过程

以下存储过程执行以下操作：

自动为架构中的所有表（及其列）生成描述。
填充目录表，其中每行代表一个包含生成的描述的表或列。

CREATE OR REPLACE PROCEDURE DESCRIBE_TABLES_SET_CATALOG (database_name string, schema_name string, catalog_table string)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.10'
  PACKAGES=('snowflake-snowpark-python','joblib')
  HANDLER = 'main'
AS
$$
import json
from joblib import Parallel, delayed
import multiprocessing

def generate_descr(session, database_name, schema_name, table, catalog_table):
    table_name =  table['TABLE_NAME']
    async_job = session.sql(f"CALL AI_GENERATE_TABLE_DESC( '{database_name}.{schema_name}.{table_name}',{{'describe_columns': true, 'use_table_data': true}})").collect_nowait()
    result = async_job.result()
    output = json.loads(result[0][0])
    columns_ret = output["COLUMNS"]
    table_ret = output["TABLE"][0]

    table_description = table_ret["description"]
    table_description = table_description.replace("'", "\\'")
    table_name = table_ret["name"]
    database_name = table_ret["database_name"]
    schema_name = table_ret["schema_name"]

    session.sql(f"""INSERT INTO {catalog_table} (domain, description, name, database_name, schema_name, table_name)
                          VALUES ('TABLE', '{table_description}', '{table_name}', '{database_name}', '{schema_name}', null)""").collect()

    for column in columns_ret:
        column_description = column["description"];
        column_description = column_description.replace("'", "\\'")
        column_name = column["name"];
        if not column_name.isupper():
            column_name = '"' + column_name + '"'
        session.sql(f"""INSERT INTO {catalog_table} (domain, description, name, database_name, schema_name, table_name)
                          VALUES ('COLUMN', '{column_description}', '{column_name}', '{database_name}', '{schema_name}', '{table_name}')""").collect()

    return 'Success';

def main(session, database_name, schema_name, catalog_table):

    schema_name = schema_name.upper()
    database_name = database_name.upper()
    catalog_table_upper = catalog_table.upper()
    tablenames = session.sql(f"""SELECT table_name
                      FROM {database_name}.information_schema.tables
                      WHERE table_schema = '{schema_name}'
                      AND table_type = 'BASE TABLE'
                      AND table_name !='{catalog_table_upper}'""").collect()
    try:
        Parallel(n_jobs=multiprocessing.cpu_count(), backend="threading")(
                delayed(generate_descr)(
                    session,
                    database_name,
                    schema_name,
                    table,
                    catalog_table,
                ) for table in tablenames
            )
        return 'Success'
    except Exception as e:
        # Catch and return the error message
        return f"An error occurred: {str(e)}"
$$;

Copy

第 2 步：创建要填充的目录表

使用以下代码创建存储表和列描述的目录表。

CREATE OR REPLACE TABLE catalog_table (
  domain VARCHAR,
  description VARCHAR,
  name VARCHAR,
  database_name VARCHAR,
  schema_name VARCHAR,
  table_name VARCHAR
  );

Copy

第 3 步：调用存储过程

假设您的架构已命名为 my_db.sch1，请按如下方式调用存储过程：

CALL describe_tables_set_catalog('my_db', 'sch1', 'catalog_table');

Copy