Configuration

How to configure Flink Agents

There are two ways to configure Flink Agents, listed in order of priority from high to low:

1. Setting via the AgentsExecutionEnvironment
2. Setting via a Flink YAML configuration file

The AgentsExecutionEnvironment applies to Agents from the AgentsExecutionEnvironment, and the Flink YAML configuration file applies to all Flink Agents Jobs using the same configuration file.

In case of duplicate keys, the value from the highest priority will override those from lower priorities.

Setting via the AgentsExecutionEnvironment

Users can explicitly modify the configuration when defining the AgentsExecutionEnvironment:

python

# Get Flink Agents execution environment
agents_env = AgentsExecutionEnvironment.get_execution_environment()

# Get configuration object from the environment
config = agents_env.get_configuration()

# Set custom configuration using a direct key (string-based key)
# This is suitable for user-defined or non-standardized settings.
config.set_int("kafkaActionStateTopicNumPartitions", 128)

# Set framework-level configuration using a predefined ConfigOption class
# This ensures type safety and better integration with the framework.
config.set(AgentExecutionOptions.ERROR_HANDLING_STRATEGY, ErrorHandlingStrategy.RETRY)

java

// Get Flink Agents execution environment
AgentsExecutionEnvironment agentsEnv = AgentsExecutionEnvironment.getExecutionEnvironment(env);

// Get configuration object
Configuration config = agentsEnv.getConfig();

// Set custom configuration using key (direct string key)
config.setInt("kafkaActionStateTopicNumPartitions", 128);  // Kafka topic partitions count

// Set framework configuration using ConfigOption (predefined option class)
config.set(AgentExecutionOptions.ERROR_HANDLING_STRATEGY, ErrorHandlingStrategy.RETRY);

Setting via the Flink YAML configuration file

Flink Agents allows reading configurations from the Flink YAML configuration file.

Format

As part of the Flink configuration file, the flink agents configuration must follow this format, with all agent-specific settings nested under the agent key:

agent:
  # Agent-specific configurations
  error-handling-strategy: retry
  chat:
    async: true

Loading Behavior

By default, the configuration is automatically loaded from $FLINK_HOME/conf/config.yaml.

Special Condition

In the following two cases, Flink Agents may not locate the corresponding configuration file, necessitating manual configuration. If the files are not set, no configuration files will be loaded, potentially resulting in unexpected behavior or failures.

• For MiniCluster: Manual setup is required — always export the environment variable before running the job:

  export FLINK_CONF_DIR="path/to/your/config.yaml"

  This ensures that Flink can locate and load the configuration file correctly.

• Local mode: When run without flink, use the AgentsExecutionEnvironment.get_configuration() API to load the YAML file directly:

  config = agents_env.get_configuration("path/to/your/config.yaml")

Built-in configuration options

Core Options

Here is the list of all built-in core configuration options.

Key Default Type Description

baseLogDir (none) String Base directory for file-based event logs. If not set, uses java.io.tmpdir/flink-agents.

prettyPrint false boolean Whether to enable pretty-printed JSON format for event logs. When set to true, each event is written as formatted multi-line JSON instead of JSONL (JSON Lines) format.

Note: enabling this option makes the log file no longer valid JSONL format.

error-handling-strategy ErrorHandlingStrategy.FAIL ErrorHandlingStrategy Strategy for handling errors during model requests, include timeout and unexpected output schema.
The option value could be:

• ErrorHandlingStrategy.FAIL
• ErrorHandlingStrategy.RETRY
• ErrorHandlingStrategy.IGNORE

max-retries 3 int Number of retries when using ErrorHandlingStrategy.RETRY.

chat.async true boolean Whether chat asynchronously for built-in chat action.

tool-call.async true boolean Whether process tool call for built-in tool call action.

rag.async true boolean Whether retrieve context asynchronously for built-in context retrieval action.

num-async-threads os cpu count * 2 int The thread pool size for async executor.

job-identifier none String The unique identifier of job, remaining consistent after restoring from a savepoint. If not set, uses flink job id.

Action State Store

Kafka-based Action State Store

Here are the configuration options for Kafka-based Action State Store.