AI_CLASSIFY uses Flink operators to perform classification with Large Language Models (LLMs). It supports the real-time computing engine Ververica Runtime (VVR) 11.4 and later. Note the rate limiting and input/output requirements of Model Studio. - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

A função AI_CLASSIFY classifica textos usando um large language model (LLM). Use-a no Flink SQL para adicionar uma coluna de classificação baseada em LLM ao seu pipeline de streaming, sem precisar criar uma UDF.

Pré-requisitos

Antes de começar, verifique se você tem:

Ververica Runtime (VVR) 11.4 ou superior
Um modelo registrado com tipo de saída VARIANT. Para mais detalhes, consulte Model settings.

Sintaxe

AI_CLASSIFY(
  MODEL => MODEL <model_name>,
  INPUT => <input_column>,
  LABELS => <labels>
)

Há suporte aos estilos de argumentos posicionais e nomeados. Consulte Exemplos.

Parâmetros

Parâmetro	Tipo	Descrição
`MODEL <model_name>`	MODEL	Modelo registrado a ser usado. O tipo de saída do modelo deve ser VARIANT.
`<input_column>`	STRING	Coluna de texto a ser classificada.
`<labels>`	ARRAY	Rótulos de classificação. Deve ser uma constante.

Valores de retorno

A função AI_CLASSIFY retorna uma linha para cada linha de entrada, com as seguintes colunas:

Coluna	Tipo	Descrição
`category`	STRING	Rótulo atribuído pelo modelo.
`confidence`	DOUBLE	Nível de confiança do rótulo atribuído.

Exemplos

Este exemplo referencia um modelo nativo do Flink e uma view de product, e classifica cada item usando os estilos de argumentos posicionais e nomeados.

Dados de teste

id	content	label
1	Li-Ning Way of Wade 10 Basketball Shoes, Performance Basketball Shoes, Shock Absorption and Rebound, Black/Red	Digital
2	Apple iPhone 15 Pro Max 256GB, Space Black, 5G Phone, A17 Pro Chip, Titanium Frame	Clothing

SQL

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

CREATE TEMPORARY VIEW products(id, content)
AS VALUES
  (1, 'Li-Ning Way of Wade 10 Basketball Shoes, Performance Basketball Shoes, Shock Absorption and Rebound, Black/Red'),
  (2, 'Apple iPhone 15 Pro Max 256GB, Space Black, 5G Phone, A17 Pro Chip, Titanium Frame');

-- Positional argument style
SELECT id, category, confidence
FROM products,
LATERAL TABLE(
  AI_CLASSIFY(MODEL general_model, content, ARRAY['Digital', 'Clothing']));

-- Named argument style
SELECT id, category, confidence
FROM products,
LATERAL TABLE(
  AI_CLASSIFY(
    MODEL  => MODEL general_model,
    INPUT  => content,
    LABELS => ARRAY['Digital', 'Clothing']));

Saída

id	category	confidence
1	Clothing	0,95
2	Digital	0,99

Observações de uso

LABELS deve ser uma constante. Não há suporte a arrays de rótulos dinâmicos ou computados.

Limitações

Requer Ververica Runtime (VVR) 11.4 ou superior.
Para usar o service Flink AI (modelos nativos), é necessário ter o VVR 11.7 ou superior e ativar o service Flink AI. Para mais detalhes, consulte Serviço Flink AI (modelos nativos).
O throughput dos operadores AI_CLASSIFY está sujeito aos limites de taxa do Alibaba Cloud Model Studio. Quando esses limites são atingidos para um modelo, o job do Flink sofre backpressure, com os operadores AI_CLASSIFY como gargalo. Em alguns casos, isso pode causar erros de timeout e reinicialização do job.

Próximos passos

Model settings: Registre e configure um modelo para uso com AI_CLASSIFY.