Frequently Asked Questions about AI with Confluent Cloud for Apache Flink¶

This topic provides answers to frequently asked questions about using large language models (LLMs) and machine learning (ML) with Confluent Cloud for Apache Flink®.

What is AI with Confluent Cloud for Apache Flink?¶

AI with Confluent Cloud for Apache Flink enables you to integrate LLMs and ML models into your streaming data workflows seamlessly.

How do I get started using AI with Confluent Cloud for Apache Flink?¶

Get started by clicking SQL Workspaces in the Confluent Cloud Console. For more information, see Run an AI Model.

How do I register a remote AI model?¶

Use the CREATE MODEL statement to register an AI model that’s hosted outside Confluent Cloud. For more information, see Create a Remote AI Model.

How do I provide tools to an AI model?¶

Use the AI_TOOL_INVOKE function to provide Model Context Protocol (MCP) tools and user-defined functions (UDFs) to an AI model. For more information, see Invoke an AI Model.

How can I forecast and detect anomalies in my data?¶

Use the ML_FORECAST and ML_DETECT_ANOMALIES functions to forecast and detect anomalies in your data. For more information, see Built-in AI/ML Functions.

How can I evaluate the performance of an AI model?¶

Use the ML_EVALUATE function to evaluate the performance of an AI model.

How can I use an AI model to generate text?¶

Use the AI_PREDICT function with an AI model to generate text.

How can I use an AI model to classify data?¶

Use the ML_PREDICT function with an AI model to classify data.

How can I create vector embeddings for retrieval-augmented generation (RAG)?¶

Use the AI_EMBEDDING function to create vector embeddings for RAG. For more information, see Vector search with Pinecone.

How can I do vector searches over external tables?¶

Use the VECTOR_SEARCH_AGG function to do vector searches over external tables.

How can I do key and text searches over external tables?¶

Use the KEY_SEARCH_AGG and TEXT_SEARCH_AGG functions to do key and text searches over external tables.

How can I use an AI model to generate text completions?¶

Use the AI_COMPLETE function with an AI model to generate text completions.

What are Streaming Agents?¶

Streaming Agents are AI-powered streaming applications that can reason over streaming data and take actions through tool invocation. They process events, make decisions, and interact with external systems in real-time.

Streaming Agents provide: - Declarative agent definition - Tool integration (function-based and MCP-based) - Multi-agent workflows

For more information, see Streaming Agents Overview.

How do I create a Streaming Agent?¶

Create a Streaming Agent using the CREATE AGENT statement:

CREATE AGENT customer_support_agent
USING MODEL my_model
USING PROMPT 'You are a helpful customer support agent'
USING TOOLS lookup_tool, support_api_tool
WITH (
  'max_iterations' = '5',
  'request_timeout' = '600'
);

You can also create tools using CREATE TOOL:

CREATE CONNECTION mcp_connection
WITH (
   'type' = 'mcp_server',
   'api-key' = '<api-key>',
   'endpoint' = 'https://mcp.example.com',
);

CREATE TOOL lookup_tool
USING CONNECTION mcp_connection
WITH (
  'type' = 'mcp',
  'description' = 'Lookup customer information'
);

For more information, see CREATE AGENT Statement and CREATE TOOL Statement.

How do I execute a Streaming Agent?¶

Execute a Streaming Agent by using the AI_RUN_AGENT function:

SELECT *
FROM customer_context,
LATERAL TABLE(
  AI_RUN_AGENT('customer_support_agent', `prompt`, `request_id`));

The function processes streaming data through the agent and returns the original data plus agent outputs.

For more information, see AI_RUN_AGENT Function.

What types of tools can Streaming Agents use?¶

Streaming Agents support two types of tools:

Function-based tools: Wrap Flink UDFs and execute locally in the Confluent Cloud for Apache Flink runtime. Best for high-frequency operations and low-latency requirements.

MCP-based tools: Connect to external services by using the Model Context Protocol. Best for accessing external APIs and services.

For more information, see Create and Run Streaming Agents.

For more information, see Agent Runtime Guide.