使用AI_EXTRACT函数调用AI大模型提取结构化信息-实时计算Flink版-阿里云

本文为您介绍如何使用AI_EXTRACT通过AI大模型进行结构化信息提取。

使用限制

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

语法

AI_EXTRACT(
  MODEL => MODEL <MODEL NAME>, 
  INPUT => <INPUT COLUMN NAME>,
  EXTRACT_SCHEMA => <EXTRACT SCHEMA>
)

入参

参数	数据类型	说明
MODEL <MODEL NAME>	MODEL	注册的模型服务名字。详情请参见模型设置注册模型服务。注意：目前需要该模型的输出类型为 VARIANT 类型。
<INPUT COLUMN NAME>	STRING	模型待提取信息的原始文本。
<EXTRACT SCHEMA>	STRING	用JSON文本的形式描述提取的schema信息。注意：目前该入参需要是一个常量。

输出

参数	数据类型	说明
extracted_json	STRING	提取出的结构化信息。

示例

测试数据

id	description
1	小明今年18岁住在杭州，他的号码是******。

测试语句

示例SQL创建通义千问Qwen-Plus模型，使用AI_EXTRACT提取用户信息。

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'endpoint'='<YOUR ENDPOINT>',
    'apiKey' = '<YOUR KEY>',
    'model' = 'qwen-plus'
);

CREATE TEMPORARY VIEW infos(id, description)
AS VALUES (1, '小明今年18岁住在杭州，他的号码是******。');

-- Use positional argument to call AI_EXTRACT
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL general_model, 
    description, 
    '{"name":"string","phone":"string","address":"string","age":"int"}'));  
-- Use named argument to call AI_EXTRACT
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL => MODEL general_model, 
    INPUT => description, 
    EXTRACT_SCHEMA => '{"name":"string","phone":"string","address":"string","age":"int"}'));

输出结果

id	extracted_json
1	{"address":"杭州","age":18,"name":"小明","phone":"******"}