AI_EXTRACT uses Flink operators with Large Language Model (LLM) services to extract structured information. It requires Ververica Runtime (VVR) 11.4 or later and is subject to rate limiting by the model platform. - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Pré-requisitos

Antes de usar a função AI_EXTRACT, registre um service de modelo. Para obter detalhes, consulte Model Settings.

O tipo de saída do modelo deve ser VARIANT.

Sintaxe

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

A função AI_EXTRACT aceita argumentos nomeados (mostrados acima) e posicionais. Consulte a seção Exemplo para ver ambos os estilos de chamada.

Parâmetros

Parâmetro	Tipo de dados	Descrição
`MODEL <MODEL NAME>`	MODEL	Nome do service de modelo registrado.
`<INPUT COLUMN NAME>`	STRING	Coluna que contém o texto não estruturado para extração de informações.
`<EXTRACT_SCHEMA>`	STRING	String JSON que define os campos a extrair e seus tipos de dados. Deve ser uma constante.

Formato de EXTRACT_SCHEMA

O parâmetro EXTRACT_SCHEMA aceita um objeto JSON em que cada chave representa o nome do campo a extrair e cada valor indica o tipo de dados de destino:

{"<field_name>": "<data_type>", ...}

Exemplo de schema:

{"name": "string", "age": "int", "address": "string", "phone": "string"}

Saída

Coluna	Tipo de dados	Descrição
`extracted_json`	STRING	Campos extraídos como uma string JSON.

Exemplo

Este exemplo extrai informações pessoais (nome, idade, endereço e telefone) de uma coluna de texto usando um modelo nativo do Flink.

Etapa 1: Registrar o modelo

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

Etapa 2: Chamar AI_EXTRACT

CREATE TEMPORARY VIEW infos(id, description)
AS VALUES (1, 'Xiao Ming is 18 years old and lives in Hangzhou. His phone number is ******.');

-- Named argument style
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"}'));

-- Positional argument style
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL general_model,
    description,
    '{"name":"string","phone":"string","address":"string","age":"int"}'));

Saída:

id	extracted_json
1	`{"address":"Hangzhou","age":18,"name":"Xiao Ming","phone":"******"}`

Limitações

Suportado apenas pelo mecanismo de computação em tempo real Ververica Runtime (VVR) 11,4 ou posterior.
Para usar o Flink AI service (modelos nativos), é necessário ter o VVR 11,7 ou posterior e ativar o Flink AI service. Para obter detalhes, consulte Flink AI service (built-in models).
O throughput está limitado pela plataforma de service de modelo. Quando o tráfego atinge o limite de acesso da plataforma, ocorre backpressure nos operadores do Flink usados nas instruções AI_EXTRACT. Limitações severas de taxa podem causar timeouts e reiniciar o job do Flink.