SQL UDF 包装 AI Function 最佳实践

通过

CREATE FUNCTION

CREATE FUNCTION

将 AI Function 封装为 SQL UDF，可以固化模型选择、prompt 逻辑和参数配置，让业务用户以最简方式调用 AI 能力。

基础模式

模式一：包装 AI_CLASSIFY（分类任务）

CREATE OR REPLACE FUNCTION public.classify_ticket(text STRING) RETURNS STRING AS AI_CLASSIFY( text, ARRAY('系统缺陷：API超时', '账号权限：SSO登录', '计费问题：升级扣费', '功能建议', '通用咨询'), JSON '{"output.behavior": "raw_string"}' ); -- 使用 SELECT public.classify_ticket('API一直报超时') AS category;

优点：

标签列表固化在 UDF 中，业务用户无需关心分类体系
```
output.behavior
```
output.behavior
参数封装在内，输出格式统一
更换分类体系时只需改 UDF 定义，调用方不受影响

模式二：包装 AI_COMPLETE 自定义 prompt（情感打分）

CREATE OR REPLACE FUNCTION public.sentiment_score(text STRING) RETURNS STRING AS AI_COMPLETE( CONCAT('请对以下文本进行情感打分，返回 -1.0 到 +1.0 之间的数值，只返回数字：', text), JSON '{"output.behavior": "raw_string"}' ); -- 使用 SELECT public.sentiment_score('产品非常好用') AS score;

优点：

prompt 模板封装在 UDF 中，调用方无需编写 prompt
```
raw_string
```
raw_string
模式确保输出可直接
```
TRY_CAST
```
TRY_CAST
为数值
可统一控制是否开启 thinking 模式等参数

模式三：多参数 UDF（运营建议生成）

CREATE OR REPLACE FUNCTION public.gen_recommendation( category STRING, volume INT, avg_sentiment DOUBLE ) RETURNS STRING AS AI_COMPLETE( CONCAT('你是一名客服运营主管。类别：', category, '，工单量：', volume, '，平均情感分：', avg_sentiment, '。请给出具体改进建议。'), JSON '{"output.behavior": "raw_string"}' ); -- 使用 SELECT public.gen_recommendation('API超时', 1240, -0.85) AS advice;

模式四：包装 AI_MASK（PII 脱敏）

CREATE OR REPLACE FUNCTION public.mask_pii(text STRING) RETURNS STRING AS AI_MASK( text, ARRAY('姓名', '电话', '邮箱', '地址'), JSON '{"output.behavior": "raw_string"}' ); -- 使用 SELECT public.mask_pii('张三的电话是13800138000') AS masked;

进阶用法

与动态表组合

将 UDF 直接嵌入动态表定义，实现声明式 AI 管道：

-- 先创建 UDF CREATE OR REPLACE FUNCTION public.classify_ticket(text STRING) RETURNS STRING AS ... ; CREATE OR REPLACE FUNCTION public.sentiment_score(text STRING) RETURNS STRING AS ... ; -- 动态表中直接使用 UDF CREATE DYNAMIC TABLE enriched_tickets REFRESH INTERVAL 30 MINUTE AS SELECT ticket_id, channel, raw_text, created_at, public.classify_ticket(raw_text) AS category, public.sentiment_score(raw_text) AS sentiment_score FROM raw_tickets;

参数化分类（动态传标签）

CREATE OR REPLACE FUNCTION public.flexible_classify( text STRING, labels ARRAY(STRING) ) RETURNS STRING AS AI_CLASSIFY(text, labels, JSON '{"output.behavior": "raw_string"}'); -- 对不同场景传不同标签 SELECT public.flexible_classify('API超时', ARRAY('技术', '业务', '其他')) AS tag;

批量处理优化

在 UDF 内部启用并发参数，提升大表处理吞吐量：

CREATE OR REPLACE FUNCTION public.batch_classify(text STRING) RETURNS STRING AS AI_CLASSIFY( text, ARRAY('系统缺陷：API超时', '账号权限：SSO登录', '计费问题：升级扣费', '功能建议', '通用咨询'), JSON '{"output.behavior": "raw_string", "task.concurrency": "4"}' );

task.concurrency

task.concurrency

控制单个查询内的并发度。建议值 4~8，上限 128。

最佳实践总结

1. 模型配置策略

场景	建议
分类、抽取、翻译等确定性任务	UDF 中省略 model 参数，使用工作区默认模型
需要特定模型的自定义 prompt	可在 UDF 内显式指定 `'conn_bailian:qwen3.6-flash'` 'conn_bailian:qwen3.6-flash'
测试阶段	先用 session SET 切换模型验证效果，再固化为 UDF

2. output.behavior 选择

模式	适用场景	UDF 返回值
`raw_string` raw_string	下游需要纯文本	直接返回模型输出，配合 `TRY_CAST` TRY_CAST 转数值
`formatted_json` formatted_json	下游需解析结构化结果	返回 JSON 字符串
`fail_on_error` fail_on_error	不容忍静默失败的场景	异常时抛错而非返回 NULL

3. 权限管理

UDF 是原生 Schema 对象，通过标准 RBAC 控制访问权限
业务用户只需
```
SELECT
```
SELECT
权限即可使用 UDF，无需直接访问 AI Function
底层模型切换对调用方透明

4. 注意事项

注意点	说明
Schema 前缀	调用 UDF 时必须使用 Schema 前缀，如 `public.classify_ticket()` public.classify_ticket()
参数顺序	带默认值的参数必须排在无默认值的参数之后
模型锁定	UDF 中指定的 model、prompt、options 一旦固化为 UDF，调用方无法修改
版本管理	通过 `CREATE OR REPLACE FUNCTION` CREATE OR REPLACE FUNCTION 更新 UDF 定义，对调用方透明
性能测试	建议在 UDF 中设置 `task.concurrency` task.concurrency ，对比不同并发度下的吞吐量

完整示例：智能客服管道

-- 1. 定义 AI UDF CREATE OR REPLACE FUNCTION public.classify_ticket(text STRING) RETURNS STRING AS AI_CLASSIFY(text, ARRAY('系统缺陷：API超时', '账号权限：SSO登录', '计费问题：升级扣费', '功能建议', '通用咨询'), JSON '{"output.behavior": "raw_string", "task.concurrency": "4"}'); CREATE OR REPLACE FUNCTION public.sentiment_score(text STRING) RETURNS STRING AS AI_COMPLETE(CONCAT('请对以下工单情感打分，范围 -1.0 到 +1.0，只返回数字：', text), JSON '{"output.behavior": "raw_string"}'); CREATE OR REPLACE FUNCTION public.mask_pii(text STRING) RETURNS STRING AS AI_MASK(text, ARRAY('姓名', '电话', '邮箱', '地址'), JSON '{"output.behavior": "raw_string"}'); -- 2. 原始数据表 CREATE TABLE raw_tickets ( ticket_id BIGINT, channel STRING, raw_text STRING, created_at TIMESTAMP ); -- 3. 动态表：使用 UDF 自动处理 CREATE DYNAMIC TABLE enriched_tickets REFRESH INTERVAL 30 MINUTE AS SELECT ticket_id, channel, raw_text, created_at, public.classify_ticket(raw_text) AS category, public.sentiment_score(raw_text) AS sentiment_score, public.mask_pii(raw_text) AS masked_text FROM raw_tickets; -- 4. 业务查询（无需了解 AI Function 细节） SELECT category, COUNT(*) AS volume, ROUND(AVG(TRY_CAST(sentiment_score AS DOUBLE)), 2) AS avg_sentiment FROM enriched_tickets GROUP BY category ORDER BY volume DESC;

联系我们