SQLを使用してオブジェクトの説明を自動的に生成する

Cortex Powered Object Descriptions機能では、Snowflake Cortex COMPLETE関数</sql-reference/functions/complete-snowflake-cortex>`を使用して、テーブル、ビュー、列の説明を自動的に生成できます。Snowflake Cortexは、Snowflakeがホストする大規模言語モデル(:doc:)を活用して、オブジェクトのメタデータを評価し、必要であればサンプルデータを使用して説明を生成します。

このトピックでは、ストアドプロシージャを使用してプログラムで説明を生成する方法について説明します。|sf-web-interface|を使用して説明を生成する方法について詳しくは、:doc:`/user-guide/ui-snowsight-cortex-descriptions`をご覧ください。

説明の生成

:doc:`/sql-reference/stored-procedures/ai_generate_table_desc`ストアドプロシージャは、テーブルとビューの説明を自動的に生成します。また、そのテーブルまたはビューの列の説明を生成することもできます。

AI_GENERATE_TABLE_DESCストアドプロシージャは2つの引数を受け入れます。

  • 説明を生成するテーブルまたはビューの名前。

  • 以下のことを行える、オプションの構成オブジェクト。

    • 指定されたテーブルまたはビューの列の説明を生成する。

    • テーブルまたはビューのサンプルデータを使用することで、列の説明の精度を向上させる可能性がある。

例:テーブルの説明を生成する
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

ストアドプロシージャの完全な構文については、:doc:`/sql-reference/stored-procedures/ai_generate_table_desc`をご覧ください。

応答の操作

AI_GENERATE_TABLE_DESCストアドプロシージャは、テーブルと列に関する一般的な情報とともに、生成された説明を含むJSONオブジェクトを返します。このオブジェクト内の:code:`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フィールドの詳細については、:ref:`label-ai_generate_table_desc_returns`をご覧ください。

生成された説明をコメントとして設定する

生成された説明をテーブル、ビュー、または列のコメントとして設定するには、SET COMMENTパラメーターを含むSQLステートメントを手動で実行する必要があります。たとえば、テーブル``t1``の生成された説明を保存するには、``ALTER TABLE t1 SET COMMENT = 'ai generated description';``を実行します。

説明を自動的に生成して保存するカスタムコードを記述することができます。これを行うストアドプロシージャの例については、:ref:`label-sql_object_descriptions_example`をご覧ください。

アクセス制御の要件

ユーザーが AI_GENERATE_TABLE_DESC ストアドプロシージャを呼び出すには、以下の権限とロールが付与されている必要があります。

  • テーブルまたはビューに対する SELECT 権限。

  • SNOWFLAKE.CORTEX_USER データベースロール。

機能の可用性

リージョンは、Snowflake Cortex（Mistral-7bやLlama 3.1-8bなど）で説明を生成するために使用するLLMをサポートしている必要があります。COMPLETE 関数 <label-cortex_llm_availability>` の可用性を確認します。COMPLETE 関数がリージョンでサポートされていない場合に、この機能を使用するには :doc:` クロスリージョン推論 </user-guide/snowflake-cortex/cross-region-inference>` を有効にする必要があります。

サンプルデータの使用

列の説明を生成する際、メタデータのみに頼ることもできますが、Snowflake Cortex Powered Descriptionを改善するためにサンプルデータを使用することもできます。サンプルデータとは、Snowflake Cortexを使用して記述を生成する際に評価される特定の列内のデータのことです。サンプルデータの使用を選択した場合、Snowflakeはサンプルデータの一部を使用して説明を生成するため、より正確な説明につながります。サンプルデータはSnowflakeに使用データとして保存されません。

コストの考慮事項

説明文の作成には以下のコストがかかります。

  • 使用中のウェアハウスで消費されたクレジット。

  • Mistral-7bやLlama 3.1-8bのような小型の LLMs、Snowflake Cortexを使用するためのクレジットチャージ。これらの請求は、Snowflake Cortex のすべての使用を含む AI-Services として請求書に表示されます。

制限事項

1,000列を超えるオブジェクトの列の説明を生成することはできません。

例

例:説明を生成し、コメントとして設定する

ステップ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
言語: 日本語