检测并隐去个人身份信息 (PII)¶

基本脱敏示例¶

以下示例从输入文本中隐去姓名和地址。

SELECT AI_REDACT(
  input => 'My name is John Smith and I live at twenty third street, San Francisco.'
  );

基本脱敏输出：

My name is [NAME] and I live at [ADDRESS]

以下示例仅从输入文本中隐去姓名和电子邮件地址。请注意，文本中仅包含名字，该名字已被识别并处理为 [NAME]。输入文本不包含电子邮件地址，因此输出中不会显示电子邮件占位符。

SELECT AI_REDACT(
  input => 'My name is John and I live at twenty third street, San Francisco.',
  categories => ['NAME', 'EMAIL']
  );

选择性脱敏输出：

My name is [NAME] and I live at twenty third street, San Francisco.

端到端示例¶

以下示例对一个表中的行进行处理，并将隐去敏感信息后的结果插入到另一个表中。您可以使用类似的方法将脱敏后的数据存储在现有表的某一列中。脱敏后，文本将传递给 AI_SENTIMENT 函数，以提取整体情绪信息。

1. 创建一个包含未脱敏文本的表。

   CREATE OR REPLACE TABLE raw_table AS
     SELECT 'My previous manager, Washington, used to live in Kirkland. His first name was Mike.' AS my_column
     UNION ALL
     SELECT 'My name is William and I live in San Francisco. You can reach me at (415).450.0973';
   

   Copy

2. 查看未脱敏的数据。

   SELECT * FROM raw_table;
   

   Copy

3. 创建脱敏表。

   CREATE OR REPLACE TABLE redaction_table (value VARCHAR);
   

   Copy

4. 从 raw_table 中脱敏 PII，并将行插入到 redaction_table。

   INSERT INTO redaction_table
     SELECT AI_REDACT(my_column) AS value FROM raw_table;
   

   Copy

5. 查看脱敏结果。

   SELECT * FROM redaction_table;
   

   Copy

6. 对脱敏后的文本运行 AI_SENTIMENT 函数。

   SELECT
       value AS redacted_text,
       AI_SENTIMENT(value) AS summary_sentiment
     FROM redaction_table;
   

   Copy

分块示例¶

本示例演示如何通过将长文本拆分为较小块、分别对每块进行隐去处理，再将经过隐去处理后的块重新组合为最终输出，从而实现对 PII 的隐去处理。此方法可规避 AI_REDACT 的词元限制。

1. 创建一个包含患者数据的表。

   CREATE OR REPLACE TABLE patients (
     patient_id INT PRIMARY KEY,
     patient_notes TEXT
     );
   

   Copy

2. 将文本拆分为多个块，对每个块应用 AI_REDACT，并将脱敏后的块连接起来。

   CREATE OR REPLACE TABLE final_temp_table AS
     WITH chunked_data AS (
       SELECT
           patient_id,
           chunk.value AS chunk_text,
           chunk.index AS chunk_index
         FROM
           patients,
           LATERAL FLATTEN(
               input => SNOWFLAKE.CORTEX.SPLIT_TEXT_RECURSIVE_CHARACTER(
                   patient_notes,
                   'none',
                   1000
                   )
               ) AS chunk
         WHERE
           patient_notes IS NOT NULL
           AND LENGTH(patient_notes) > 0
       ),
     redacted_chunks AS (
         SELECT
             patient_id,
             chunk_index,
             chunk_text,
             TO_VARIANT(results:value) AS redacted_chunk,
             TO_VARIANT(results:error) AS error_string
           FROM (
             SELECT
                 patient_id,
                 chunk_index,
                 chunk_text,
                 AI_REDACT(chunk_text,TRUE) AS results
               FROM
                 chunked_data
           )
     ),
     final AS (
         SELECT
             chunk_text AS original,
             IFF(error_string IS NOT NULL, chunk_text, redacted_chunk) AS redacted_text,
             patient_id,
             chunk_index
           FROM
             redacted_chunks
     )
     SELECT * FROM final;
   

   Copy

3. 查询结果。

   SELECT
       patient_id,
       LISTAGG(redacted_text, '') WITHIN GROUP (ORDER BY chunk_index) AS full_output
     FROM final_temp_table
     GROUP BY patient_id;
   

   Copy

基本检测示例¶

以下示例在不脱敏输入的情况下，识别并返回每个检测到的 PII 实例的类别、位置和文本。

SELECT AI_REDACT(
    input => 'My old manager, Washington, used to live in Washington. His first name was Mike.',
    return_error_details => FALSE,
    mode => 'detect'
    );

基本检测输出：

{
  "spans": [
    {
      "category": "NAME",
      "end": 26,
      "start": 16,
      "text": "Washington"
    },
    {
      "category": "ADDRESS",
      "end": 54,
      "start": 44,
      "text": "Washington"
    },
    {
      "category": "NAME",
      "end": 79,
      "start": 75,
      "text": "Mike"
    }
  ]
}

使用允许列表的端到端示例¶

以下示例演示了一个使用 detect 模式和允许列表的选择性脱敏工作流程。该示例从暂存文件加载需要保留的名称列表，在 detect 模式下使用 AI_REDACT 识别 PII 的位置，然后将结果传递给 Python UDF，仅对不在允许列表中的 PII 进行脱敏。

1. 通过将列表从暂存区加载到临时表中来保留一个允许值列表。

   CREATE OR REPLACE TEMP TABLE string_list (value STRING);
   
   COPY INTO string_list
     FROM @mystage/allowlist.txt
     FILE_FORMAT = (
       TYPE = 'CSV'
       RECORD_DELIMITER = '\n'
       FIELD_DELIMITER = '\t'   -- any char NOT in file
       TRIM_SPACE = TRUE
       SKIP_HEADER = 0
       );
   

   Copy

2. 查看允许列表表

   SELECT * FROM string_list;
   

   Copy

   允许列表表输出：

   VALUE
   Mike
   David
   

3. 创建一个 Python UDF，根据允许列表对 PII 进行选择性脱敏。

   CREATE OR REPLACE FUNCTION redact_spans_with_allowlist(
     SPAN_DATA VARIANT,
     ALLOWLIST ARRAY,
     ORIGINAL_TEXT STRING
     )
     RETURNS STRING
     LANGUAGE PYTHON
     RUNTIME_VERSION = '3.8'
     HANDLER = 'redact_text'
     AS
     $$
     def redact_text(span_data, allowlist, original_text):
         spans = span_data.get('spans', [])
         # Sort descending to maintain index integrity
         sorted_spans = sorted(spans, key=lambda x: x['start'], reverse=True)
   
         result = original_text
   
         for span in sorted_spans:
             text_val = span.get('text')
             if text_val in allowlist:
                 continue
   
             start, end = span['start'], span['end']
             label = f"[{span['category']}]"
   
             # Splice the string
             result = result[:start] + label + result[end:]
   
         return result
     $$;
   

   Copy

4. 测试 UDF。

   SELECT redact_spans_with_allowlist(
     PARSE_JSON('{"spans": [{"category": "NAME", "end": 26, "start": 16, "text": "Washington"}, {"category": "NAME", "end": 79, "start": 75, "text": "Mike"}]}'),
     ARRAY_CONSTRUCT('Washington'), -- This will NOT be redacted
     'Hello, my name is Washington and his is Mike.'
     );
   

   Copy

5. 在 detect 模式下运行 AI_REDACT。

   CREATE OR REPLACE TABLE raw (message TEXT);
   
   INSERT INTO raw (message) VALUES
     ('My old manager, Washington, used to live in Washington. His first name was Mike.');
   
   SELECT
       t.message AS message,
       AI_REDACT(input=>t.message, return_error_details=>FALSE, mode=>'detect') AS spans,
       redact_spans_with_allowlist(spans, l.str_list, message) AS result
     FROM raw t
       CROSS JOIN (
         SELECT ARRAY_AGG(value) AS str_list
           FROM string_list
         ) l;
   

   Copy

端到端允许列表示例输出：

{
  "spans": [
    {"category": "NAME",
    "end": 26,
    "start": 16,
    "text": "Washington"
    },
    {"category": "ADDRESS",
    "end": 54,
    "start": 44,
    "text": "Washington"
    },
    {"category": "NAME",
    "end": 79,
    "start": 75,
    "text": "Mike"
    }
  ]
}