<a id="cmf-model-inference"></a>

# Run Model Inference in Confluent Manager for Apache Flink

With `CREATE MODEL`, you register a remote AI model in the environment
catalog, and a statement calls it for each row with `ML_PREDICT`.
Confluent Manager for Apache Flink® (CMF) supports the `openai` provider, which works with any
service that has an OpenAI-compatible API.

To use models, you need:

- The environment catalog, enabled. See [Enable the environment catalog](../../../configure/environment-catalog.md#cmf-environment-catalog-enable).
- A compute pool that runs Flink 2.x. CMF rejects model statements on Flink 1.x compute
  pools.
- The model’s endpoint in the list of allowed endpoints. See
  [Allow model endpoints](#cmf-model-inference-allow).

## Create a model

Create the model in the environment catalog. The following example creates the model
`sentiment` in the environment `env-1`:

```bash
confluent flink statement create create-sentiment-model \
  --url http://cmf:8080 \
  --environment env-1 \
  --compute-pool pool \
  --catalog _env_env-1 \
  --database default \
  --sql "CREATE MODEL sentiment INPUT (review STRING) OUTPUT (label STRING) WITH ('provider' = 'openai', 'endpoint' = 'https://api.openai.com/v1/chat/completions', 'model' = 'gpt-4o-mini', 'api-key' = '<api-key>', 'system-prompt' = 'Classify the review as positive or negative. Output only the label.');"
```

| Option                  | Description                                                                                                                                 |
|-------------------------|---------------------------------------------------------------------------------------------------------------------------------------------|
| `provider`              | Required. Must be `openai`.                                                                                                                 |
| `model`                 | Required. The name of the model on the model server.                                                                                        |
| `endpoint`              | The URL of the model server. The path must end with `chat/completions` or<br/>`embeddings`. Required unless `credentials-secret-id` is set. |
| `api-key`               | The key for the model server. Required unless `credentials-secret-id` is set.                                                               |
| `credentials-secret-id` | The name of a secret mapping that holds the endpoint and the key. See<br/>[Keep the key in a secret](#cmf-model-inference-secret).          |
| `system-prompt`         | The instructions that the model gets with each row.                                                                                         |

CMF masks the value of `api-key` in the statement text and in the output of
`SHOW CREATE MODEL`.

<a id="cmf-model-inference-secret"></a>

### Keep the key in a secret

To keep the key out of the SQL text, store the endpoint and the key in a secret, and map the
secret into the environment. The secret’s data must contain exactly the keys `endpoint` and
`api-key`:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Secret",
  "metadata": {
    "name": "openai-secret"
  },
  "spec": {
    "data": {
      "endpoint": "https://api.openai.com/v1/chat/completions",
      "api-key": "<api-key>"
    }
  }
}
```

Create a secret mapping named `openai-credentials` in `env-1` that points to
`openai-secret`. For how to create the secret and the mapping, see [Secrets and secret mappings](../../../configure/catalog.md#secret-mappings).
Then set `credentials-secret-id` to the name of the mapping:

```sql
CREATE MODEL sentiment INPUT (review STRING) OUTPUT (label STRING) WITH (
  'provider' = 'openai',
  'credentials-secret-id' = 'openai-credentials',
  'model' = 'gpt-4o-mini'
);
```

A model with `credentials-secret-id` can’t also set `api-key`. If it sets `endpoint`,
the value must be the endpoint in the secret.

## Call a model

Call the model with `ML_PREDICT`. Pass the input table, the model, and the columns to send
to the model:

```sql
SELECT * FROM ML_PREDICT(TABLE reviews, MODEL sentiment, DESCRIPTOR(review));
```

The result has the columns of the input table and the `OUTPUT` columns of the model. Use
the catalog and database to name a table or model outside the current database, for
example `TABLE `kafka-cat`.`kafka-db`.`reviews``.

`ML_PREDICT` sends the values of the `DESCRIPTOR` columns to the model’s endpoint.

## Manage models

Run these statements in the environment catalog:

```sql
SHOW MODELS;
SHOW CREATE MODEL sentiment;
ALTER MODEL sentiment SET ('model' = 'gpt-4.1-mini');
ALTER MODEL sentiment RENAME TO review_sentiment;
DROP MODEL review_sentiment;
```

`ALTER MODEL ... SET` changes the options you name and keeps the others. A new statement
that calls a dropped model fails. A statement that is already running keeps calling it.

<a id="cmf-model-inference-allow"></a>

## Allow model endpoints

An administrator lists the endpoints that models may call in the CMF Helm values. The list
is empty by default, so no model can be created until you add an endpoint:

```yaml
cmf:
  sql:
    models:
      allowedEndpoints:
        - https://api.openai.com/v1
```

Each entry is an `http` or `https` URL with an optional path. An endpoint is allowed when
its scheme, host, and port match an entry and its path is the entry’s path or below it.
CMF doesn’t start if an entry isn’t a valid URL, or has user info, a query, or a fragment.

CMF checks the endpoint at `CREATE MODEL`, at `ALTER MODEL ... SET`, and each time a
statement that calls the model is compiled. This includes an endpoint from a secret. A
statement that is already running keeps calling its endpoint.

## Limitations

- The only provider is `openai`.
- Models are stored only in the environment catalog, not in Kafka catalogs.
- Model statements run only on compute pools with Flink 2.x.

## Related content

- [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../../configure/environment-catalog.md#cmf-environment-catalog)
- [Secrets and secret mappings](../../../configure/catalog.md#secret-mappings)
- [Create Statements with Confluent Manager for Apache Flink](../create-statements.md#flink-sql-statements-create)