# AI Model Inference and Machine Learning Functions in Confluent Cloud for Apache Flink
Confluent Cloud for Apache Flink® provides built-in functions for invoking remote AI/ML models in
Flink SQL queries. These functions simplify developing and deploying AI by
providing a unified platform for data processing and AI/ML tasks.
- [AI_COMPLETE](#flink-sql-ai-complete-function): Generate text
completions.
- [AI_EMBEDDING](#flink-sql-ai-embedding-function): Create embeddings.
- [AI_RUN_AGENT](#flink-sql-ai-run-agent-function): Execute a streaming
agent.
- [AI_TOOL_INVOKE](#flink-sql-ai-tool-invoke-function): Invoke model
context protocol (MCP) tools.
- [AI_DETECT_PII](#flink-sql-ai-detect-pii-function): Detect and redact
personally identifiable information (PII) in text.
- [AI_DETECT_ANOMALIES](#flink-sql-ai-detect-anomalies-function): Detect
anomalies in time-series data using foundation models.
- [AI_FORECAST](#flink-sql-ai-forecast-function): Forecast future values
in time-series data using foundation models.
- [AI_SENTIMENT](#flink-sql-ai-sentiment-function): Analyze sentiment
toward specific aspects in text using aspect-based sentiment analysis.
- [ML_DETECT_ANOMALIES](#flink-sql-ml-anomaly-detect-function): Detect
anomalies in your data.
- [ML_DETECT_ANOMALIES_ROBUST](#flink-sql-ml-anomaly-detect-robust-function):
Detect multivariate anomalies in your data using a robust approach.
- [ML_EVALUATE](#flink-sql-ml-evaluate-function): Evaluate the performance
of an AI/ML model.
- [ML_FORECAST](#flink-sql-ml-forecast-function): Forecast trends.
- [ML_PREDICT](#flink-sql-ml-predict-function): Run a remote AI/ML model
for tasks like predicting outcomes, generating text, and classification.
For enriching streams with lookups against key-value stores, full-text
indexes, or vector databases, see [Search Functions](search-functions.md#flink-sql-search-functions).
For machine-language preprocessing utilities, see
[ML Preprocessing Functions](ml-preprocessing-functions.md#flink-sql-ml-preprocessing-functions).
## ML_PREDICT
Run a remote AI/ML model for tasks like predicting outcomes, generating text,
and classification.
Syntax
: ```sql
ML_PREDICT(`model_name[$version_id]`, column);
-- map settings are optional
ML_PREDICT(`model_name[$version_id]`, column, map['async_enabled', [boolean], 'client_timeout', [int], 'max_parallelism', [int], 'retry_count', [int]]);
```
Description
: The ML_PREDICT function performs predictions using pre-trained machine
learning models.
The first argument to the ML_PREDICT table function is the model name. The
other arguments are the columns used for prediction. They are defined in the
model resource INPUT for AI models and can vary in length or type.
Before using ML_PREDICT, you must register the model by using the
[CREATE MODEL](../statements/create-model.md#flink-sql-create-model) statement.
For more information, see [Run an AI Model](../../../ai/ai-model-inference.md#flink-sql-ai-model).
Configuration
: You can control how calls to the remote model execute with these optional
parameters.
- `async_enabled`: Calls to remote models are asynchronous and don’t block.
The default is `true`.
- `client_timeout`: Time, in seconds, after which the request to the model
endpoint times out. The default is 30 seconds.
- `debug`: Return a detailed stack trace in the API response. The default
is `false`. Confluent Cloud for Apache Flink implements data masking for error messages to
remove any secrets or customer input, but the stack trace can contain the
prompt itself or some part of the response string.
- `retry_count`: Maximum number of times the remote model request is
retried if the request to the model fails. The default is 2.
- `max_parallelism`: Maximum number of parallel requests that the function
can make. Can be used only when `async_enabled` is `true`. The default
is the number of available CPU cores.
Example
: After you have registered the AI model by using the
[CREATE MODEL](../statements/create-model.md#flink-sql-create-model) statement, run the model by
using the ML_PREDICT function in a SQL query.
The following example runs a model named `embeddingmodel` on the data in
a table named `text_stream`.
```sql
SELECT id, text, embedding FROM text_stream, LATERAL TABLE(ML_PREDICT('embeddingmodel', text));
```
The following examples call the ML_PREDICT function with different configurations.
```sql
-- Specify the timeout.
SELECT * FROM `db1`.`tb1`, LATERAL TABLE(ML_PREDICT('md1', key, map['client_timeout', 60 ]));
-- Specify all configuration parameters.
SELECT * FROM `db1`.`tb1`, LATERAL TABLE(ML_PREDICT('md1', key, map['async_enabled', true, 'client_timeout', 60, 'max_parallelism', 20, 'retry_count', 5]));
```
## ML_DETECT_ANOMALIES
Identify outliers in a data stream.
Syntax
: ```sql
ML_DETECT_ANOMALIES(
data_column,
timestamp_column,
JSON_OBJECT('p' VALUE 1, 'q' VALUE 1, 'd' VALUE 1, 'minTrainingSize' VALUE 10));
```
Description
: The ML_DETECT_ANOMALIES function uses an
[ARIMA model](../../../ai/builtin-functions/detect-anomalies.md#flink-sql-detect-anomalies-arima-model) to identify
outliers in time-series data.
Your data must include:
- A timestamp column.
- A target column representing some quantity of interest at each timestamp.
For more information, see
[Detect Anomalies in Data](../../../ai/builtin-functions/detect-anomalies.md#flink-sql-detect-anomalies).
Parameters
: For anomaly detection parameters,
see [ARIMA model parameters](../../../ai/builtin-functions/detect-anomalies.md#flink-sql-detect-anomalies-arima-model-parameters).
Example
: ```sql
SELECT
ML_DETECT_ANOMALIES(
total_orderunits,
summed_ts,
JSON_OBJECT('p' VALUE 1, 'q' VALUE 1, 'd' VALUE 1, 'minTrainingSize' VALUE 10))
OVER (
ORDER BY summed_ts
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS anomalies
FROM test_table;
```
## ML_DETECT_ANOMALIES_ROBUST
Identify outliers in a data stream by using univariate or multivariate
detection.
Syntax
: ```sql
ML_DETECT_ANOMALIES_ROBUST(
value_or_vector,
event_time,
JSON_OBJECT(
'window' VALUE 60,
'threshold' VALUE 3.0,
'imputeOutliers' VALUE TRUE,
'majorityRule' VALUE TRUE,
'majorityThreshold' VALUE 0.6,
'expertWindows' VALUE '[30,60,120]'
)
)
-- Univariate detection
ML_DETECT_ANOMALIES_ROBUST(latency_ms, ts, config_json)
-- Multivariate detection
ML_DETECT_ANOMALIES_ROBUST(ROW(cpu, heap_used, gc_pause_ms), ts, config_json)
```
Description
: The ML_DETECT_ANOMALIES_ROBUST function is a built-in machine learning
function that detects anomalies in one or more metrics over time by using
robust statistics instead of forecasting models.
Your input data must include:
- A timestamp column, for ordering the time series.
- A single numeric column, for univariate detection, or
- A vector of numeric columns, passed as a `ROW(...)` expression, for
multivariate detection.
If your stream has multiple records with the same timestamp or irregular
spacing, consider first aggregating into regular intervals with TUMBLE and
then applying ML_DETECT_ANOMALIES_ROBUST on the windowed results.
For more information, see
[Detect Anomalies in Data with MAD](../../../ai/builtin-functions/detect-anomalies.md#flink-sql-detect-anomalies-ml-detect-anomalies-robust).
Configuration
: - `value_or_vector` (DOUBLE or ROW): The latency of the data stream.
- `event_time` (TIMESTAMP): The timestamp of the data stream.
- `config_json` (JSON): A JSON object that has the following fields:
- `window` (INT): The window size in number of observations.
- `threshold` (DOUBLE): The threshold for the anomaly detection.
- `imputeOutliers` (BOOLEAN): Whether to impute outliers.
Expert mode
: For advanced scenarios, ML_DETECT_ANOMALIES_ROBUST supports an *expert mode*,
in which a panel of MAD detectors with different window sizes vote on each
point, and you decide how many must agree before an anomaly is raised.
In expert mode, the primary expert still uses `window`, and all experts
share the same `threshold`. For multivariate inputs, previously flagged
anomalies are excluded automatically from future context for each expert, to
keep the panel robust.
Expert mode is controlled by the following parameters.
- `majorityRule` (BOOLEAN): Whether to use the majority rule in expert
mode. When TRUE, enables expert-panel / majority-vote mode. The default
is FALSE.
- When FALSE: a single MAD detector with window `window` drives all
anomaly decisions.
- When TRUE: a panel of experts is built, each with its own window size.
- `majorityThreshold` (DOUBLE): The threshold for the majority rule in
expert mode. The supported range of values is `0.0` to `1.0`. The
default is `0.5`.
The `majorityThreshold` parameter defines the fraction of ready experts
that must vote “anomaly” for the point to be flagged.
If `ready` is the number of experts that have enough history to evaluate
the current point, and `votes` is the number of those experts that say it
is an anomaly, the point is flagged if `votes / ready > majorityThreshold`.
With the default value of `0.5`, strictly more than half of ready experts
must agree.
- `expertWindows` (ARRAY): Sizes, in number of observations, for each
expert’s sliding window. The default, when `majorityRule` is TRUE,
is a panel derived from the `window` parameter, roughly:
`window / 4, window / 2, window, 2 * window, 5 * window`.
All values are clamped to a reasonable range.
The type is either:
- A JSON array, for example, `JSON_ARRAY(20, 40, 80)`
- A string representation, like “[20,40,80]” or “20,40,80”
Include both short windows, for fast response, and long windows, for
stability.
Keep values positive and reasonably small compared to your history size.
Output
: The ML_DETECT_ANOMALIES_ROBUST function returns a single ROW field named
`anomaly`, with the following schema, which matches the schema of the
ML_DETECT_ANOMALIES function.
- `timestamp` (TIMESTAMP): The event time of the evaluated point.
- `actual_value` (DOUBLE): The original metric value.
- Scalar mode: the original metric value.
- Vector mode: NULL, because the input is multidimensional.
- `forecast_value` (DOUBLE): The robust center (median) of the metric over
the current window.
- Scalar mode: the robust center (median) of the metric over the current
window.
- Vector mode: the distance of the current vector from the robust center
vector.
- `lower_bound` (DOUBLE): The lower bound on allowed distance.
- Scalar mode: computed with the formula `forecast_value - threshold * robust_sigma`.
- Vector mode: `0.0`, representing the lower bound on allowed distance.
- `upper_bound` (DOUBLE): The distance threshold; values above this indicate an anomaly.
- Scalar mode: computed with the formula `forecast_value + threshold * robust_sigma`.
- Vector mode: the distance threshold. Values above this indicate an anomaly.
- `is_anomaly` (BOOLEAN): TRUE if the current point exceeds the robust
threshold; otherwise FALSE. During warm-up, this field is NULL.
- `rmse` (DOUBLE): Reserved for future use. Always NULL.
- `aic` (DOUBLE): Reserved for future use. Always NULL.
In multivariate mode, the anomaly score you typically examine is:
- `anomaly.forecast_value` for robust distance
- `anomaly.upper_bound` for distance threshold
- `anomaly.is_anomaly` for final decision
Example
: ```sql
SELECT
ML_DETECT_ANOMALIES_ROBUST(latency_ms, ts, config_json)
OVER (
ORDER BY ts
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS anomaly
FROM test_table;
```
## ML_EVALUATE
Aggregate a table and return model evaluation metrics.
Syntax
: ```sql
ML_EVALUATE(`model_name`, label, col1, col2, ...) FROM 'eval_data_table';
```
### Description
The ML_EVALUATE function is a table aggregation function that takes an entire
table and returns a single row of model evaluation metrics. If run on all
versions of a model, the function returns one row for each model version.
After comparing the metrics for different versions, you can update the
default version for deployment with the model that has the best evaluation
metrics.
Internally, the ML_EVALUATE function runs [ML_PREDICT](#flink-sql-ml-predict-function)
and processes the results.
Before using ML_EVALUATE, you must register the model by using the
[CREATE MODEL](../statements/create-model.md#flink-sql-create-model) statement.
The first argument to the ML_EVALUATE table function is the model name.
The second argument is the true label that the output of the model should be
evaluated against. Its type depends on the model OUTPUT type and the model
task. The other arguments are the columns used for prediction. They are
defined in the model resource INPUT for AI models and can vary in length
or type.
The return type of the ML_EVALUATE function is `Map` for
all types of tasks. Each task type has different metrics keys in the map,
depending on the task type.
### Metrics
The metric columns returned by ML_EVALUATE depend on the
[task type](../statements/create-model.md#flink-sql-create-model-task-types) of the specified model.
#### Classification
Classification models choose a group to place their inputs in and return one of
N possible values. A classification model that returns only 2 possible values
is called a *binary classifier*. If it returns more than 2 values, it is
referred to as *multi-class*.
Classification models return these metrics:
- [Accuracy](https://en.wikipedia.org/wiki/Accuracy_and_precision): Total
Fraction of correct predictions across all classes.
- [F1 Score](https://en.wikipedia.org/wiki/F1_score): Harmonic mean of
precision and recall.
- [Precision](https://en.wikipedia.org/wiki/Precision_and_recall): (Class X
Correctly Predicted) / (# of Class X Predicted)
- [Recall](https://en.wikipedia.org/wiki/Precision_and_recall): (Class X
Correctly Predicted) / (# of actual Class X)
#### Clustering
Clustering models group the model examples into K groups. Metrics are a measure
of how compact the clusters are.
Clustering models return these metrics:
- [Davies Bouldin Index](https://en.wikipedia.org/wiki/Davies%E2%80%93Bouldin_index):
A measure of how separated clusters are and how compact they are.
- Intra-Cluster Variance (Mean Squared Distance): Average Squared distance of
each training point to the centroid of the cluster it was assigned to.
- [Silhouette Score](https://en.wikipedia.org/wiki/Silhouette_(clustering)):
Compares how similar each point is to its own cluster with how dissimilar it
is to other clusters.
#### Embedding
Embedding models return these metrics:
- [Mean Cosine Similarity](https://en.wikipedia.org/wiki/Cosine_similarity):
A measure of how similar two vectors are.
- [Mean Jaccard Similarity](https://en.wikipedia.org/wiki/Jaccard_index):
A measure of how similar two sets are.
- [Mean Euclidean Distance](https://en.wikipedia.org/wiki/Euclidean_distance):
A measure of how similar two vectors are.
#### Regression
Regression models predict a continuous output variable based on one or more
input features.
Regression models return these metrics:
- [Mean Absolute Error](https://en.wikipedia.org/wiki/Mean_absolute_error):
The average of the absolute differences between the predicted and actual values.
- [Mean Squared Error](https://en.wikipedia.org/wiki/Mean_squared_error):
The average of the squared differences between the predicted and actual values.
#### Text generation
Text generation models generate text based on a prompt. Text generation models
return these metrics:
- [Mean BLEU](https://en.wikipedia.org/wiki/BLEU):
A measure of how similar two texts are.
- [Mean ROUGE](https://en.wikipedia.org/wiki/ROUGE_(metric)):
A measure of how similar two texts are.
- [Mean Semantic Similarity](https://en.wikipedia.org/wiki/Semantic_similarity):
A measure of how similar two texts are.
#### Example metrics
The following table shows example metrics for different task types.
| Task type | Example metrics |
|-----------------|----------------------------------------------------------------------------------------------------------------------------------------|
| Classification | {Accuracy=0.9999991465990892, Precision=0.9996998081063332, Recall=0.0013025368892873059, F1=0.0013025368892873059} |
| Clustering | {Mean Davies-Bouldin Index=0.9999991465990892} |
| Embedding | {Mean Cosine Similarity=0.9999991465990892, Mean Jaccard Similarity=0.9996998081063332, Mean Euclidean Distance=0.0013025368892873059} |
| Regression | {MAE=0.9999991465990892, MSE=0.9996998081063332, RMSE=0.0013025368892873059, MAPE=0.0013025368892873059, R²=0.0043025368892873059} |
| Text generation | {Mean BLEU=0.9999991465990892, Mean ROUGE=0.9996998081063332, Mean Semantic Similarity=0.0013025368892873059} |
### Example
After you have registered the AI model by using the
[CREATE MODEL](../statements/create-model.md#flink-sql-create-model) statement, run the model by
using the ML_EVALUATE function in a SQL query.
The following example statement registers a remote OpenAI model for a
classification task.
```sql
CREATE MODEL `my_remote_model`
INPUT (f1 INT, f2 STRING)
OUTPUT (output_label STRING)
WITH(
'task' = 'classification',
'type' = 'remote',
'provider' = 'openai',
'openai.endpoint' = 'https://api.openai.com/v1/llm/v1/chat',
'openai.api_key' = ''
);
```
The following statements show how to run the ML_EVALUATE function on
various versions of `my_remote_model` using data in a table named
`eval_data`.
```sql
-- Model evaluation with all versions
SELECT ML_EVALUATE(`my_remote_model$all`, label, f1, f2) FROM `eval_data`;
-- Model evaluation with default version
SELECT ML_EVALUATE(`my_remote_model`, label, f1, f2) FROM `eval_data`;
-- Model evaluation with specific version 2
SELECT ML_EVALUATE(`my_remote_model$2`, label, f1, f2) FROM `eval_data`;
```
## ML_FORECAST
Perform continuous forecasting on a table.
Syntax
: ```sql
ML_FORECAST(
data_column,
timestamp_column,
JSON_OBJECT('p' VALUE 1, 'q' VALUE 1, 'd' VALUE 1, 'minTrainingSize' VALUE 10));
```
Description
: The ML_FORECAST function uses an
[ARIMA model](../../../ai/builtin-functions/forecast.md#flink-sql-forecast-arima-model) to perform time-series
forecasting.
Your data must include:
- A timestamp column.
- A target column representing some quantity of interest at each timestamp.
For more information, see [Forecast Data Trends](../../../ai/builtin-functions/forecast.md#flink-sql-forecast).
Parameters
: For forecasting parameters, see
[ARIMA model parameters](../../../ai/builtin-functions/forecast.md#flink-sql-forecast-arima-model-parameters).
Example
: ```sql
SELECT
ML_FORECAST(
total_orderunits,
summed_ts,
JSON_OBJECT('p' VALUE 1, 'q' VALUE 1, 'd' VALUE 1, 'minTrainingSize' VALUE 10))
OVER (
ORDER BY summed_ts
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS forecast
FROM test_table;
```
## AI_COMPLETE
Invoke a large language model (LLM) to generate text completions, summaries, or answers.
Syntax
: ```sql
AI_COMPLETE(model_name, input_prompt [, invocation_config]);
```
Description
: The AI_COMPLETE function provides a streamlined approach for generating text,
taking a single string as input and returning a single string as output. This
functionality enables you to leverage LLMs to produce text based on any
given prompt.
Configuration
: - `model_name`: Name of the model entity to call to for prediction [STRING].
- `input_prompt`: Input prompt to pass to the LLM for prediction [STRING].
- `invocation_config[optional]`: Map to pass the configuration to manage
function behavior, for example, `MAP['debug', true]`.
Example
: The following example shows how to invoke an LLM to generate text completions.
```sql
-- Create an OpenAI connection.
CREATE CONNECTION openai_connection
WITH (
'type' = 'openai',
'endpoint' = 'https://api.openai.com/v1/chat/completions',
'api-key' = ''
);
CREATE MODEL description_extractor
INPUT (input STRING)
OUTPUT (output_json STRING)
WITH(
'provider' = 'openai',
'openai.connection' = 'openai_connection',
'openai.system_prompt' = 'Extract json from input free text',
'task' = 'text_generation'
);
CREATE TABLE claims_with_structured_description(id INT, customer_id INT, output_json STRING);
INSERT INTO claims_with_structured_description
SELECT id, customer_id, output_json FROM claims_submitted, LATERAL TABLE(AI_COMPLETE('description_extractor', description));
```
## AI_DETECT_PII
Detect and redact personally identifiable information (PII) in text.
Syntax
: ```sql
AI_DETECT_PII(
text,
JSON_OBJECT(
'mode' VALUE 'redact',
'redaction_strategy' VALUE 'mask'
)
)
```
Description
: You can use the AI_DETECT_PII function to scan text for personally
identifiable information. The function uses Piiranha by default, a
model that is managed by Confluent and hosted in Confluent Cloud to
identify PII in your data. AI_DETECT_PII supports detection mode
(default), which returns structured results with PII types and
locations, and redaction mode, which replaces detected PII with
redacted values.
For more information, see
[Detect PII in Text](../../../ai/builtin-functions/pii-detection.md#flink-sql-pii-detection).
Parameters
: - `text` (STRING): The text to scan for personally identifiable
information. The text must not be null.
- `config` (STRING): Optional JSON object with configuration
parameters:
- `mode` (STRING, default: `'detect'`): The processing mode.
Set to `'detect'` or `'redact'`.
- `model` (STRING, default: `'piiranha-v1-pii'`): The model
to use for PII detection.
- `pii_types` (ARRAY, default: all types): The PII
types to detect.
- `redaction_strategy` (STRING, default: `'mask'`): The
redaction strategy. Set to `'mask'`, `'hash'`, or
`'remove'`.
- `score_threshold` (DOUBLE, default: 0.0): The minimum
confidence score for detections. Must be between 0.0 and 1.0.
Output
: The AI_DETECT_PII function returns a ROW data type with the following
fields:
- `has_pii` (BOOLEAN): TRUE if the text contains detected PII.
- `entities` (ARRAY>): An array of detected PII entities.
- `redacted_text` (STRING): The input text with detected PII
replaced. Null in detection mode.
- `metadata` (STRING): A JSON string containing model metadata.
Example
: ```sql
SELECT
message_id,
message,
AI_DETECT_PII(message) AS pii_result
FROM customer_messages;
```
## AI_DETECT_ANOMALIES
Detect anomalies in time-series data using a foundation model.
Syntax
: ```sql
AI_DETECT_ANOMALIES(
value,
timestamp,
JSON_OBJECT(
'model' VALUE 'timesfm-2.5',
'minContextSize' VALUE 10,
'maxContextSize' VALUE 200,
'confidencePercentage' VALUE 80.0
)
) OVER (
ORDER BY timestamp
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
)
```
Description
: The AI_DETECT_ANOMALIES function monitors a stream of time-series data and
flags data points that fall outside expected bounds. The function uses
IBM Granite TTM by default, a lightweight model that’s a fast,
general-purpose starting point, and also supports Google TimesFM 2.5 and
two other IBM Granite time-series models (FlowState and PatchTST-FM). All
models are managed by Confluent and hosted in Confluent Cloud, and learn
patterns from historical data to produce prediction intervals.
Your data must include:
- A timestamp column for ordering the time series.
- A numeric value column representing the time-series data to monitor.
For more information, see [Detect Anomalies in Data](../../../ai/builtin-functions/detect-anomalies.md#flink-sql-detect-anomalies).
Parameters
: - `config` (STRING): Optional JSON object with configuration parameters:
- `confidencePercentage` (DOUBLE, default: 99.0): The symmetric
confidence level for the prediction interval.
- `lowerBoundConfidencePercentage` (DOUBLE, optional): The asymmetric
lower bound confidence.
- `maxContextSize` (INTEGER, default: 512): The maximum number of
historical points used for prediction. Older data beyond the limit
is truncated, regardless of model. Behavior varies by model:
- `timesfm-2.5`: Capped at the model’s context limit and rounded
down to a multiple of 32.
- `ttm`: Automatically selects the model variant that best matches
the requested context size, from 52 to 512. No manual variant
selection needed.
- `flowstate`: Supports a context size up to 4,096.
- `patchtstfm`: Supports a context size from 16 to 8,064.
- `minContextSize` (INTEGER, default: 20): The minimum number of data
points before anomaly detection begins.
- `model` (STRING, default: `'ttm'`): The model to use for
anomaly detection. Supported values: `timesfm-2.5`, `ttm`
(default), `flowstate`, `patchtstfm`.
- `rmseWindowSize` (INTEGER, default: 5): The rolling window size for
tracking RMSE.
- `upperBoundConfidencePercentage` (DOUBLE, optional): The asymmetric
upper bound confidence.
- `timestamp` (TIMESTAMP(3)): The event time associated with the value.
- `value` (DOUBLE): The numeric time-series value to monitor.
Output
: The AI_DETECT_ANOMALIES function returns a ROW data type with the following
fields:
- `timestamp` (TIMESTAMP(3)): The timestamp of the evaluated data point.
- `actual_value` (DOUBLE): The observed value at this timestamp.
- `forecast_value` (DOUBLE): The model’s predicted value.
- `lower_bound` (DOUBLE): The lower bound of the prediction interval.
- `upper_bound` (DOUBLE): The upper bound of the prediction interval.
- `is_anomaly` (BOOLEAN): TRUE if the value falls outside the bounds.
- `rmse` (DOUBLE): The rolling RMSE of prediction errors.
- `aic` (DOUBLE): Reserved for future use. Always null.
Example
: ```sql
SELECT
event_time,
response_ms,
AI_DETECT_ANOMALIES(
response_ms,
event_time,
JSON_OBJECT(
'model' VALUE 'timesfm-2.5',
'minContextSize' VALUE 10,
'maxContextSize' VALUE 200,
'confidencePercentage' VALUE 80.0
)
) OVER (
ORDER BY event_time
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS anomaly_result
FROM server_metrics;
```
## AI_EMBEDDING
Generate vector embeddings for text or other data using a registered embedding model.
```sql
AI_EMBEDDING(model_name, input_text [, invocation_config]);
```
Description
: The AI_EMBEDDING function provides a straightforward interface, accepting a
single string input and returning an array of floats as the embedding
response. This functionality enables you to leverage large language models
(LLMs) to generate embeddings for text efficiently.
Configuration
: - `model_name`: Name of the model entity to call to for embeddings [STRING].
- `input_text`: Input text to pass to the LLM for embeddings [STRING].
- `invocation_config[optional]`: Map to pass the configuration to manage
function behavior, for example, `MAP['debug', true]`.
Example
: The following example shows how to generate vector embeddings for text or
other data using a registered embedding model.
```sql
-- Create an OpenAI connection.
CREATE CONNECTION openai_embedding_connection
WITH (
'type' = 'openai',
'endpoint' = 'https://api.openai.com/v1/embeddings',
'api-key' = ''
);
CREATE MODEL description_embedding
INPUT (input STRING)
OUTPUT (embeddings ARRAY)
WITH(
'provider' = 'openai',
'openai.connection' = 'openai_embedding_connection',
'task' = 'embedding'
);
CREATE TABLE claims_embeddings(id INT, customer_id INT, embeddings ARRAY);
INSERT INTO claims_embeddings
SELECT id, customer_id, embeddings FROM claims_submitted, LATERAL TABLE(AI_EMBEDDING('description_embedding', description));
```
## AI_FORECAST
Forecast future values in time-series data using a foundation model.
Syntax
: ```sql
AI_FORECAST(
value,
timestamp,
JSON_OBJECT(
'model' VALUE 'timesfm-2.5',
'minContextSize' VALUE 10,
'maxContextSize' VALUE 200,
'horizon' VALUE 5
)
) OVER (
ORDER BY timestamp
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
)
```
Description
: The AI_FORECAST function analyzes a stream of time-series data and produces
multi-step-ahead forecasts with uncertainty quantiles. The function uses
IBM Granite TTM by default, a lightweight model that’s a fast,
general-purpose starting point, and also supports Google TimesFM 2.5 and
two other IBM Granite time-series models (FlowState and PatchTST-FM). All
models are managed by Confluent and hosted in Confluent Cloud, and learn
patterns from historical data to predict future values.
Your data must include:
- A timestamp column for ordering the time series.
- A numeric value column representing the time-series data to forecast.
For more information, see [Forecast Data Trends](../../../ai/builtin-functions/forecast.md#flink-sql-forecast).
Parameters
: - `value` (DOUBLE): The numeric time-series value to forecast.
- `timestamp` (TIMESTAMP(3)): The event time associated with the value.
- `config` (STRING): Optional JSON object with configuration parameters:
- `horizon` (INTEGER, default: 5): The number of future time steps to
forecast. Limits vary by model:
- `timesfm-2.5`: Capped at the model’s configured limit. Requests
beyond it are rejected.
- `ttm`: Automatically selects the model variant that best matches
the requested horizon, up to 96. No manual variant selection
needed.
- `flowstate`: Supports a horizon greater than six. Combined context
size and horizon must be less than 4,102; context size must be at
least the horizon.
- `patchtstfm`: Supports a horizon up to 8,176. Combined context
size and horizon must not exceed 8,192.
- `maxContextSize` (INTEGER, default: 512): The maximum number of
historical data points used for prediction. Historical data beyond
this limit is truncated to the most recent points, regardless of
model. Behavior varies by model:
- `timesfm-2.5`: Capped at the model’s context limit and rounded
down to a multiple of 32.
- `ttm`: Automatically selects the model variant that best matches
the requested context size, from 52 to 512. No manual variant
selection needed.
- `flowstate`: Supports a context size that is at least the
value of the `horizon` parameter. Combined context size and
horizon must be less than 4,102.
- `patchtstfm`: Supports a context size from 16 to 8,064. Combined
context size and horizon must not exceed 8,192.
- `minContextSize` (INTEGER, default: 20): The minimum number of data
points required before forecasting begins.
- `model` (STRING, default: `'ttm'`): The model to use for
forecasting. Supported values: `timesfm-2.5`, `ttm` (default),
`flowstate`, `patchtstfm`.
- `rmseWindowSize` (INTEGER, default: 5): The rolling window size for
tracking RMSE.
Output
: The AI_FORECAST function returns a ROW data type with the following fields:
- `forecast` (ARRAY>): An array of forecasted data points, one per
horizon step. Each entry contains the predicted timestamp, mean forecast,
and three quantiles (10th, 50th, and 90th percentile).
- `metadata` (STRING): A JSON string containing model metadata, including
the model name, requested horizon, frequency, and input lengths.
Example
: ```sql
SELECT
event_time,
cpu_percent,
AI_FORECAST(
cpu_percent,
event_time,
JSON_OBJECT(
'model' VALUE 'timesfm-2.5',
'minContextSize' VALUE 10,
'maxContextSize' VALUE 200,
'horizon' VALUE 5
)
) OVER (
ORDER BY event_time
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) AS forecast_result
FROM cpu_metrics;
```
## AI_SENTIMENT
Analyze sentiment toward specific aspects in text using aspect-based sentiment
analysis.
Syntax
: ```sql
AI_SENTIMENT(
text,
ARRAY['aspect1', 'aspect2'],
JSON_OBJECT(
'batchSize' VALUE 32,
'model' VALUE 'deberta'
)
)
```
Description
: You can use the AI_SENTIMENT function to analyze text and determine sentiment
toward specific aspects. The function uses DeBERTa, a transformer model
designed for aspect-based sentiment analysis that is managed by Confluent
and hosted in Confluent Cloud, to classify sentiment for each aspect you specify as
positive, negative, or neutral with a confidence score.
For more information, see [Analyze Sentiment in Text](../../../ai/builtin-functions/sentiment.md#flink-sql-sentiment).
Parameters
: - `aspects` (ARRAY): The aspects to evaluate sentiment for. The
array must not be null or empty.
- `config` (STRING): Optional JSON object with configuration parameters:
- `batchSize` (INTEGER, default: 32): The number of aspect-text pairs to
process per batch. Must be greater than 0.
- `model` (STRING, default: `'deberta'`): The model to use for sentiment
analysis.
- `text` (STRING): The text to analyze for sentiment. The text must not be
null or blank.
Output
: The AI_SENTIMENT function returns a ROW data type with the following fields:
- `sentiment` (ARRAY>): An
array of sentiment results, one per aspect. Each entry contains the aspect
name, sentiment label as a string value (`'positive'`, `'negative'`, or
`'neutral'`), and confidence score as a double value from 0.0 to 1.0.
- `metadata` (STRING): A JSON string containing model metadata.
Example
: ```sql
SELECT
review_id,
review_text,
AI_SENTIMENT(
review_text,
ARRAY['battery', 'screen', 'performance'],
JSON_OBJECT('model' VALUE 'deberta')
) AS sentiment_result
FROM reviews;
```
## AI_RUN_AGENT
Execute a streaming agent against real-time context data, enabling dynamic
reasoning and tool invocation within SQL queries.
Syntax
: ```sql
AI_RUN_AGENT(agent_name, prompt [, request_id]
[, system_table] [, invocation_config]);
```
Configuration
: - `agent_name`: Name of the registered agent to execute
[STRING].
- `prompt`: Task prompt for the agent [STRING].
- `request_id`: Optional. Request ID for the request [STRING].
- `system_table`: Optional. Name of a system log table to
capture agent workflow logs. The table must use the required
schema. For more information, see
[Monitor Streaming Agents](../../../ai/streaming-agents/monitor-streaming-agents.md#monitor-streaming-agents)
[STRING].
- `invocation_config`: Optional. Configuration map for agent
execution, for example, `MAP['debug', 'true']` to enable
more descriptive errors [MAP].
Return Type
: The function returns a ROW.
- `status`: Execution status, either SUCCESS or FAILED.
- `response`: The final response from the agent.
Example
: ```sql
SELECT * FROM ml_catalog.db.my_table,
LATERAL TABLE(AI_RUN_AGENT('claim_intake_agent', 'Extract structured data from this claim description', request_id));
```
Execute an agent with a system log table:
```sql
SELECT `prompt`, `status`, `response`
FROM `customer_queries`,
LATERAL TABLE(
AI_RUN_AGENT(
'my_agent', `prompt`, `id`,
'agent_system_table'));
```
## AI_TOOL_INVOKE
Invoke a registered tool, either externally by using an MCP server or locally
by using a [UDF](../../concepts/user-defined-functions.md#flink-sql-udfs), as part of an AI workflow.
Syntax
: ```sql
AI_TOOL_INVOKE(model_name, input_prompt, remote_udf_descriptor, mcp_tool_descriptor [, invocation_config]);
```
Description
: The AI_TOOL_INVOKE function enables large language models (LLMs) to access
various tools. The LLM decides which tools should be accessed, then the
AI_TOOL_INVOKE function invokes the tools, gets the responses, and returns the
responses to the LLM. The function returns a map that includes all the tools
that were accessed, along with their responses and the status of the call,
indicating whether it was a SUCCESS or FAILURE.
The following models are supported:
- Anthropic
- AzureOpenAI
- Gemini
- OpenAI
#### NOTE
The AI_TOOL_INVOKE function is available for preview.
A Preview feature is a Confluent Cloud component that is being introduced to gain
early feedback from developers. Preview features can be used for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent do not apply to Preview features. Confluent may discontinue
providing preview releases of the Preview features at any time in
Confluent’s’ sole discretion.
Configuration
: - `model_name`: Name of the model entity to call [STRING].
- `input_prompt`: Input prompt to pass to the LLM [STRING].
- `remote_udf_descriptor`: Map to pass UDF names as key and function
description as value [MAP]. A maximum of 3 UDFs can be
passed.
- `mcp_tool_descriptor`: Map to pass MCP tool names as key and tool
description as value [MAP]. A maximum of 5 tools can be
passed. This additional description is passed to the LLM as
“Additional description”. If the MCP server already has a description, and
if the server doesn’t have a description, `mcp_tool_descriptor` is added as
the description. You can leave it empty, in which case no changes are
made to the description provided by the server.
- `invocation_config[optional]`: Map to pass the config to manage function
behavior, for example, `MAP['debug', true, 'on_error', 'continue']`.
Example
: The following example shows how to invoke a UDF and a registered external
tool or API as part of an AI workflow.
When you create an MCP server connection, specify the following options:
- `endpoint`: Defines the base URL for all non-SSE communications with the
MCP server, including other http calls and general data exchange.
- `sse_endpoint`: Specifies the explicit URL endpoint used to establish a
Server-Sent Events (SSE) connection with the MCP server. If omitted, the
client defaults to constructing the SSE endpoint by appending `/sse` to
the domain specified in `endpoint`.
- `transport-type`: Specifies the transport type to use for the connection.
Valid values are `SSE` and `STREAMABLE_HTTP`. The default is `SSE`.
```sql
-- Create an MCP server connection.
CREATE CONNECTION claims_mcp_server
WITH (
'type' = 'mcp_server',
'endpoint' = 'https://mcp.deepwiki.com',
'sse-endpoint' = 'https://mcp.deepwiki.com/sse',
'api-key' = 'api_key'
);
```
```sql
-- Create a model that uses the MCP server connection.
CREATE MODEL tool_invoker
INPUT (input_message STRING)
OUTPUT (tool_calls STRING)
WITH(
'provider' = 'openai',
'task' = 'text_generation',
'openai.connection' = 'openai_connection',
'openai.system_prompt' = 'Select the best tools to complete the task',
'mcp.connection' = 'claims_mcp_server'
);
-- Create a table that contains the input prompts.
CREATE TABLE claims_verified (
id int,
customer_id int
);
-- Run the AI_TOOL_INVOKE function.
SELECT
id,
customer_id,
AI_TOOL_INVOKE(
'tool_invoker',
customer_id,
MAP['udf_1', 'udf_1 description', 'udf_2', 'udf_2 description'],
MAP['tool_1', 'tool_1_description', 'tool_2', 'tool_2_description']
) AS verified_result
FROM claims_verified;
```
## Other built-in functions
- [Aggregate Functions](aggregate-functions.md#flink-sql-aggregate-functions)
- [Changelog Conversion Functions](changelog-conversion.md#flink-sql-changelog-conversion-functions)
- [Collection Functions](collection-functions.md#flink-sql-collection-functions)
- [Comparison Functions](comparison-functions.md#flink-sql-comparison-functions)
- [Conditional Functions](conditional-functions.md#flink-sql-conditional-functions)
- [Datetime Functions](datetime-functions.md#flink-sql-datetime-functions)
- [Hash Functions](hash-functions.md#flink-sql-hash-functions)
- [JSON Functions](json-functions.md#flink-sql-json-functions)
- [ML Preprocessing Functions](ml-preprocessing-functions.md#flink-sql-ml-preprocessing-functions)
- [Model Inference Functions](#flink-sql-model-inference-functions)
- [Numeric Functions](numeric-functions.md#flink-sql-numeric-functions)
- [Search Functions](search-functions.md#flink-sql-search-functions)
- [String Functions](string-functions.md#flink-sql-string-functions)
- [Table API Functions](table-api-functions.md#flink-table-api-functions)
## Related content
- [Build AI with Flink SQL](../../../ai/overview.md#ai-overview)
- [CREATE MODEL](../statements/create-model.md#flink-sql-create-model)
- [Flink SQL Queries](../queries/overview.md#flink-sql-queries)
- [ML Preprocessing Functions](ml-preprocessing-functions.md#flink-sql-ml-preprocessing-functions)
- [Search Functions](search-functions.md#flink-sql-search-functions)
#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).