AI_MASK通过Flink算子实现基于大模型的数据脱敏，支持指定实体屏蔽，但受限于模型平台吞吐量和版本要求。 - 实时计算Flink版

使用限制

仅实时计算引擎VVR 11.4及以上版本支持。
AI_MASK语句相关的Flink算子的吞吐量受到模型平台限流的限制。当触及平台允许的访问流量上限时，Flink作业会表现出以该算子为瓶颈的反压现象。在限流情况严重时，可能会触发相关算子的超时报错及作业重启。

语法

AI_MASK(
  MODEL => MODEL <MODEL NAME>, 
  INPUT => <INPUT COLUMN NAME>,
  MASK_ENTITIES => <MASK ENTITIES>
)

入参

参数	数据类型	说明
MODEL <MODEL NAME>	MODEL	注册的模型服务名字。详情请参见模型设置注册模型服务。注意：目前需要该模型的输出类型为 VARIANT 类型。
<INPUT COLUMN NAME>	STRING	模型待分析的原始文本。
<MASK ENTITIES>	ARRAY<STRING>	需要脱敏的实体。注意：目前该入参需要是一个常量。

输出

参数	数据类型	说明
masked_text	STRING	脱敏后的文字。
detected_entities	ARRAY<STRING>	检测到的实体。

示例

测试数据

id	content
1	Timmo真的太爱学习了吧，一有空闲就看到他在看学习资料。

测试语句

示例SQL创建千问Qwen-Plus模型，使用AI_MASK进行数据脱敏。

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'task' = 'chat/completions',
    'model' = 'qwen-plus'
);

CREATE TEMPORARY VIEW infos(id, content)
AS VALUES (1, 'Timmo真的太爱学习了吧，一有空闲就看到他在看学习资料。');

-- Use positional argument to call AI_MASK
SELECT id, masked_text, detected_entities
FROM infos,
LATERAL TABLE(
  AI_MASK(
    MODEL general_model, 
    content, 
    ARRAY['name']
    ));

-- Use named argument to call AI_MASK
SELECT id, masked_text, detected_entities
FROM infos,
LATERAL TABLE(
  AI_MASK(
    MODEL => MODEL general_model, 
    INPUT => content, 
    MASK_ENTITIES => ARRAY['name']
    ));

输出结果

id	masked_text	detected_entities
1	[NAME]真的太爱学习了吧，一有空闲就看到他在看学习资料。	[{"entity":"Timmo","type":"name"}]