# Run an AI Model with Confluent Cloud AI model inference runs a trained AI or ML model on streaming data to produce predictions in real time. With Confluent Intelligence, you run model inference in Flink SQL by creating a model as a resource, like a table or function, and calling it in streaming queries. The SQL interface is available in Cloud Console and the Flink SQL shell. The [CREATE MODEL](../flink/reference/statements/create-model.md#flink-sql-create-model) statement registers an AI or ML model in your Apache Flink® environment for real-time prediction and inference. You must have the endpoint and API keys of your model before you can use it in a query. The following examples get you started with AI model inference. - [Text generation example with Claude hosted by Anthropic](#flink-sql-ai-model-text-generation) - [Text embedding examples](#flink-sql-ai-model-text-embedding) - [Sentiment analysis example with an OpenAI model](#flink-sql-ai-model-sentiment-analysis) - [Create a managed AI model](#flink-sql-managed-ai-model-example) ## Prerequisites - Access to Confluent Cloud. - Access to a Flink compute pool. - Sufficient permissions to create models. For more information, see [RBAC for model inference](confluent-intelligence/overview.md#ai-model-inference-rbac). ## Get the endpoint and access keys for your model Before you can use a model with Confluent Cloud for Apache Flink®, you must host the model with a supported cloud provider. To connect to your model with Flink SQL, you provide the following details: * The model endpoint * The API keys that authorize Confluent Cloud to invoke your model The following providers are supported: | Provider | Authentication | |---------------|----------------------------------------------------| | Anthropic | API key | | AWS Bedrock | AWS access key and secret (optional session token) | | AWS SageMaker | AWS access key and secret (optional session token) | | Azure OpenAI | API key | | Azure ML | API key | | Fireworks AI | API key | | Google AI | API key | | OpenAI | API key | | Vertex AI | Service account key (no API key) | The following procedures explain how to look up these details from the supported providers. ### Anthropic To connect to a Claude model that’s hosted by Anthropic, you must have the following information for your Flink SQL statements: * The model endpoint: For the Claude inference endpoints, see [API Overview](https://platform.claude.com/docs/en/api/overview) in the Claude API Docs. For conversational interactions, the endpoint is `https://api.anthropic.com/v1/messages`. * Your API key: To create or look up a Claude API key, use the Claude Console. For more information, see [Getting API Keys](https://platform.claude.com/docs/en/api/overview#getting-api-keys) in the Claude API Docs. ### AWS Bedrock 1. In your web browser, navigate to the [AWS Bedrock](https://aws.amazon.com/bedrock/) site. 2. Pick a foundation model. You might need to request access and accept a EULA, if your account hasn’t before. 3. At the bottom of the page, get the `modelId` of the model from the API request example. The model endpoint resembles `https://bedrock-runtime..amazonaws.com/model//invoke`. 4. Get your AWS access key (`AWS_ACCESS_KEY_ID`) and secret (`AWS_SECRET_ACCESS_KEY`). If you’re using temporary credentials, you need an AWS session token (`AWS_SESSION_TOKEN`), which you can get from the AWS access portal. If you use temporary credentials, the query might not work after the credentials expire. 5. Set the INPUT_FORMAT associated with your chosen model from the preceding list. If not set, Confluent Cloud for Apache Flink usually chooses the correct one automatically based on the name of the model. ### AWS SageMaker 1. In your web browser, navigate to the [AWS SageMaker Studio](https://aws.amazon.com/pm/sagemaker/) site. 2. Select **Train a Model** or **Models > Deployable Models > Create** to upload a trained model. Also, you can select **Jumpstart > pick a model > Deploy** to choose a public pretrained model. 3. Click **Deployments > Endpoints > Create endpoint**, or create it by deploying the model. The full endpoint URL is listed on the endpoint page. To authenticate with SageMaker, get your AWS access key (`AWS_ACCESS_KEY_ID`) and secret (`AWS_SECRET_ACCESS_KEY`). If you’re using temporary credentials, you need an AWS session token (`AWS_SESSION_TOKEN`), which you can get from the AWS access portal. If you use temporary credentials, the query might not work after the credentials expire. SageMaker models don’t have standard input or output formats. If you did not train the model yourself, there is no easy way to determine the expected inputs and outputs without testing the model or reading its documentation. You might also need to set the [input_content_type](../flink/reference/statements/create-model.md#flink-sql-create-model-with-input-content-type) and [output_content_type](../flink/reference/statements/create-model.md#flink-sql-create-model-with-input-content-type) parameters. Depending on how the endpoint is deployed, you might have to set additional header parameters, like [inference_component_name](../flink/reference/statements/create-model.md#flink-sql-create-model-with-sagemaker-inference-component-name). ### Azure OpenAI For non-OpenAI LLM models on Azure, see the **Azure ML** tab. These model endpoints are created in [Azure AI Foundry](https://oai.azure.com/portal). You must sign up for AI Foundry in your subscription. 1. On the left sidebar, select **Deployments** under **Management -> Create New Deployment**. The name you choose here is `YOUR_DEPLOYMENT_NAME` in the endpoint URL shown later. 2. Get the API key from the Resource in the [Cognitive Services Hub](https://portal.azure.com/#view/Microsoft_Azure_ProjectOxford/CognitiveServicesHub/~/OpenAI). 3. Select the resource group account from the list. This is `YOUR_RESOURCE_NAME` in the next step. 4. On the resource page, choose **Keys and Endpoints** from the left sidebar. The endpoint URL on that page is not the full endpoint. For chat models, the full URL resembles `https://.openai.azure.com/openai/deployments//chat/completions?api-version=2024-02-01`. For non-chat URL formats, see [Azure OpenAI Service REST API reference](https://learn.microsoft.com/en-us/azure/ai-services/openai/reference). ### Azure ML This provider supports both [Azure AI Foundry](https://ai.azure.com/) generative AI models and [Azure Machine Learning](https://ml.azure.com) for predictive models. #### Azure AI Foundry (Generative AI models) 1. Select a project or create a new one. 2. Navigate to **Components > Deployments**. 3. Navigate to **Create > Pay-as-you-go**. Usually, these models are charged per-use. 4. Navigate to **Create > Realtime Endpoint**. There are more model choices, but they have ongoing costs for as long as the model is deployed. 5. Click the deployment name to get the API key. This page also lists a **Target**, which is not the full endpoint URL. The full endpoint URL resembles `https://..inference.ai.azure.com/`. Usually, the MODEL_PATH resembles `v1/chat/completions`. You can find the whole endpoint in the “How to deploy” documentation for each model. Many of the Azure AI Foundry models use the [OPENAI-CHAT](../flink/reference/statements/create-model.md#flink-sql-create-model-openai-chat-format) input format, even if they are not OpenAI models. #### Azure Machine Learning (Predictive Models) 1. Navigate to **Train a Model** or **Models > Register** to upload one from disk. 2. Navigate to **Endpoints > Create** and select a real-time endpoint. There are also pointers from this page to do the OpenAI or Pay-as-you-go serverless AI models mentioned earlier. 3. After creating the endpoint, navigate to the REST endpoint, which is the endpoint you need for CREATE MODEL. 4. Generate an API key by using the Azure CLI with the following command. ```bash az ml online-endpoint get-credentials ``` ### Fireworks AI To connect to a model that you host with Fireworks AI, you must provide the following information in your Flink SQL statements: * The model endpoint: For the endpoints where Fireworks AI hosts its models, see the Fireworks AI documentation. For example, for the endpoint for text models, see [Text Models](https://docs.fireworks.ai/guides/querying-text-models). For the endpoint for embeddings models, see [Embeddings & Reranking](https://docs.fireworks.ai/guides/querying-embeddings-models). * Your API key: To create an API key, use the [Create API key](https://docs.fireworks.ai/api-reference/create-api-key) operation, or use the Settings page for your account in the Fireworks dashboard. * The model version: You provide a value for the `model_version` parameter in Flink SQL when you register your model. Use the value that Fireworks AI provides for the `baseModel` response parameter when you deploy a model. To look up this value for a model that you’ve deployed, use the [Get Deployment](https://docs.fireworks.ai/api-reference/get-deployment) operation, or use the Fireworks dashboard. ### Google AI To connect to a model that you host with Google AI, you must provide the following information in your Flink SQL statements: * The model endpoint, which is `https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent`. * Your API key. You can create an API key on the [Google AI Studio API keys page](https://aistudio.google.com/app/apikey). Gemini models are also supported through the Vertex AI provider, which you might prefer due to its integrated Google Cloud billing. ### OpenAI To connect to a model that’s hosted by OpenAI, you must have the following information for your Flink SQL statements: 1. The model endpoint. Usually, the endpoint is `https://api.openai.com/v1/chat/completions`. 2. Your API key. You can create an API key on the [OpenAI API keys page](https://platform.openai.com/api-keys). ### Vertex AI 1. Navigate to the [Vertex AI dashboard](https://console.cloud.google.com/vertex-ai/). 2. In the navigation menu, click **Deploy and Use > Online prediction**. 3. Navigate to **Endpoints > Create (Upload or Train Model)**. 4. Choose your endpoint and click **Sample Request** to get the endpoint information. The endpoint resembles `https://-aiplatform.googleapis.com/v1/projects//locations//endpoints/ENDPOINT_ID:predict`. Vertex AI doesn’t use API Keys. Instead, you must have a service account with the `aiplatform.endpoints.predict` IAM permission for the model resource. Also, you must create a service account key for this service account. Because no default IAM role contains only this specific permission, you should create a custom role to maintain least privilege. It is also possible to scope service account permissions such that they have access only to a single model resource. ## Create a connection resource A connection resource stores the credentials and endpoint for a model provider so Flink statements can call the provider securely without exposing secrets. Run the [CREATE CONNECTION](../flink/reference/statements/create-connection.md#flink-sql-create-connection) statement to create a new connection resource. The following examples show how to use connection resources in the CREATE MODEL statements that register models with Confluent Cloud for Apache Flink. - Connections must be created in the same cloud region as the corresponding models. For details on reusable connections, see [Reuse Confluent Cloud Connections With External Services](../integrations/connections/overview.md#connections-overview) and [Manage Connections with External Services in Confluent Cloud](../integrations/connections/manage-connections.md#manage-connections). ## Text generation example with Claude hosted by Anthropic 1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named `anthropic-connection` that uses your Claude API key. ```sql CREATE CONNECTION `anthropic-connection` WITH ( 'type' = 'anthropic', 'endpoint' = 'https://api.anthropic.com/v1/messages', 'api-key' = '' ); ``` 2. Run the following statement to register the Claude model with Confluent Cloud for Apache Flink. ```sql CREATE MODEL `anthropic-model` INPUT (`text` VARCHAR(2147483647)) OUTPUT (`output` VARCHAR(2147483647)) WITH ( 'provider' = 'anthropic', 'anthropic.connection' = 'anthropic-connection', 'anthropic.params.max_tokens' = '2048', 'task' = 'text_generation' ); ``` 3. Run the following statements to create the table for input text. ```sql CREATE TABLE `anthropic-input-table` (`id` INT, `prompt` VARCHAR(1000)); INSERT INTO `anthropic-input-table` VALUES (1, 'What is the capital of France?'), (2, 'What is the capital of Germany?'), (3, 'What is the capital of Italy?'); ``` 4. Run the following statement to invoke the Claude model. ```sql SELECT * from `anthropic-input-table`, LATERAL TABLE(AI_COMPLETE(`anthropic-model`, prompt)); ``` ## Text embedding examples These examples generate vector embeddings from text with AWS Bedrock, Azure OpenAI, Fireworks AI, and Google AI. Each example reuses the `text_input` table that you create next. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create the table for input text. ```sql CREATE TABLE `text_input` ( `id` STRING, `input` STRING ); INSERT INTO text_input VALUES ('1', 'alien'), ('2', 'goldfinger'), ('3', 'license to kill'), ('4', 'aliens'); ``` ### AWS Bedrock For simplicity, this example uses the AWS Bedrock `amazon.titan-embed-text-v1` model. 1. Run the following statement to create a connection resource named `bedrock_connection` that uses your AWS credentials. ```sql CREATE CONNECTION bedrock_connection WITH ( 'type' = 'bedrock', 'endpoint' = 'https://bedrock-runtime..amazonaws.com/model//invoke', 'aws-access-key' = '', 'aws-secret-key' = '', 'aws-session-token' = '' ); ``` 2. Run the following statement to register your model with Confluent Cloud for Apache Flink. ```sql CREATE MODEL bedrock_embed INPUT (text STRING) OUTPUT (response ARRAY) WITH ( 'bedrock.connection'='bedrock_connection', 'bedrock.input_format'='AMAZON-TITAN-EMBED', 'provider'='bedrock', 'task'='embedding' ); ``` 3. Run the following statement to invoke your model. ```sql SELECT * from text_input, LATERAL TABLE(AI_EMBEDDING('bedrock_embed', input)); ``` ### Azure OpenAI 1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = 'https://.openai.azure.com/openai/deployments//embeddings?api-version=2024-06-01', 'api-key' = '' ); ``` 2. Run the following statement to register your model with Confluent Cloud for Apache Flink. ```sql CREATE MODEL azure_embed INPUT (text STRING) OUTPUT (response ARRAY) WITH ( 'azureopenai.connection'='azureopenai_connection', 'provider'='azureopenai', 'task'='embedding' ); ``` 3. Run the following statement to invoke your model. ```sql SELECT * from text_input, LATERAL TABLE(AI_EMBEDDING('azure_embed', input)); ``` ### Fireworks AI 1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named `fireworksai-connection` that uses your Fireworks AI API key. ```sql CREATE CONNECTION `fireworksai-connection` WITH ( 'type' = 'fireworksai', 'endpoint' = 'https://api.fireworks.ai/inference/v1/embeddings', 'api-key' = '' ); ``` 2. Run the following statement to register your model with Confluent Cloud for Apache Flink. ```sql CREATE MODEL `fireworksai-embed` INPUT (`text` VARCHAR(2147483647)) OUTPUT (`embedding` ARRAY) WITH ( 'provider' = 'fireworksai', 'fireworksai.connection' = 'fireworksai-connection', 'task' = 'embedding', 'fireworksai.model_version' = '' ); ``` 3. Run the following statement to invoke your model. ```sql SELECT * from `text_input`, LATERAL TABLE(AI_EMBEDDING(`fireworksai-embed`, input)); ``` ### Google AI 1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named `googleai_connection` that uses your Google Cloud API key. ```sql CREATE CONNECTION googleai_connection WITH ( 'type' = 'googleai', 'endpoint' = 'https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent', 'api-key' = '' ); ``` 2. Run the following statement to register your model with Confluent Cloud for Apache Flink. ```sql CREATE MODEL google_text_cli INPUT (`text` VARCHAR(2147483647)) OUTPUT (`output` VARCHAR(2147483647)) WITH ( 'googleai.connection' = 'googleai_connection', 'googleai.system_prompt' = 'translate text to chinese', 'provider' = 'googleai', 'task' = 'text_generation' ); ``` 3. Run the following statement to invoke your AI model. ```sql SELECT * FROM text_input, LATERAL TABLE(AI_COMPLETE('google_text_cli', text)); ``` ## Sentiment analysis example with an OpenAI model 1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create the table for input text. ```sql CREATE TABLE text_stream ( id BIGINT, text STRING ); INSERT INTO text_stream SELECT 1 id, 'The mitochondria are the powerhouse of the cell' text; INSERT INTO text_stream SELECT 2 id, 'Happy Birthday! You are great!' text; INSERT INTO text_stream SELECT 3 id, 'Today was a bad day in the stock market.' text; ``` 2. Run the following statement to create a connection resource named `openai_connection` that uses your OpenAI API key. ```sql CREATE CONNECTION openai_connection WITH ( 'type' = 'openai', 'endpoint' = 'https://api.openai.com/v1/chat/completions', 'api-key' = '' ); ``` 3. Run the following statement to create the OpenAI model with a system prompt for sentiment analysis. ```sql CREATE MODEL sentimentmodel INPUT(text STRING) OUTPUT(sentiment STRING) COMMENT 'sentiment analysis model' WITH ( 'provider' = 'openai', 'task' = 'classification', 'openai.connection' = 'openai_connection', 'openai.model_version' = 'gpt-3.5-turbo', 'openai.system_prompt' = 'Analyze the sentiment of the text and return only POSITIVE, NEGATIVE, or NEUTRAL.' ); ``` 4. Run the inference statement on the table and model. ```sql SELECT id, text, sentiment FROM text_stream, LATERAL TABLE(ML_PREDICT('sentimentmodel', text)); ``` ## Create a managed AI model Creating a managed AI model using one of the Confluent managed models is very similar to creating a remote AI model. The most significant difference is that you specify the following options in the CREATE MODEL statement: ```properties 'provider' = 'confluent', 'confluent.model' = '' ``` For the supported managed AI models, see [confluent.model](../flink/reference/statements/create-model.md#flink-sql-create-model-with-confluent-model). #### NOTE > Managed AI models are an Early Access Program feature in Confluent Cloud. > > > > > > > > > > An Early Access feature is a component of Confluent Cloud introduced to gain > feedback. This feature should be used only for evaluation and non-production > testing purposes or to provide feedback to Confluent, particularly as it > becomes more widely available in follow-on preview editions. > Early Access Program features are intended for evaluation use in development > and testing environments only, and not for production use. Early Access Program > features are provided: (a) without support; (b) “AS IS”; and (c) without > indemnification, warranty, or condition of any kind. No service level commitment > will apply to Early Access Program features. Early Access Program features are > considered to be a Proof of Concept as defined in the Confluent Cloud Terms of Service. > Confluent may discontinue providing preview releases of the Early Access > Program features at any time in Confluent’s sole discretion. If you want to participate in the Early Access Program, [sign up here](https://events.confluent.io/early-access-flink-features). The following code example shows how to create a managed embedding. ```sql CREATE MODEL `managed_model_embedding` INPUT (text STRING) OUTPUT (embedding ARRAY) WITH ( 'provider' = 'confluent', 'task' = 'embedding', 'confluent.model'='BAAI/bge-large-en-v1.5' ); ``` For more information, see [Run a Managed AI Model](managed-model.md#flink-sql-managed-ai-model). ## Related content - [CREATE MODEL statement](../flink/reference/statements/create-model.md#flink-sql-create-model) - [Reuse Confluent Cloud Connections With External Services](../integrations/connections/overview.md#connections-overview) - [Manage Connections with External Services in Confluent Cloud](../integrations/connections/manage-connections.md#manage-connections) #### 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).