Realtime Compute for Apache Flink:Developing Flink Agents (Beta)

Agent types

You can build Workflow or ReAct agents for different use cases.

Step 1: Download sample job files

Step 2: Upload and deploy the job

1. Log on to the Realtime Compute for Apache Flink console.

2. In the Actions column of the target workspace, click Console.

3. In the left-side navigation pane, click File Management. Then, click Upload Resource to upload the job files that you downloaded.

4. On the Operations > Jobs page, click Deploy Job. Select Python Job or JAR Job based on your job type and fill in the deployment information.

Step 3: Start the job and view the results

1. On the Operations > Jobs page, find your job and click Start in the Actions column.

2. In the Start Job dialog box, select stateless start and click Start.

3. The example uses an in-memory data source, so the job automatically ends after processing is complete. Wait for the status to change to finished, and then click the job name to go to the details page.

4. On the TaskManager tab, view the logs and search for the keyword Review analysis result: to see the output.

Installing Flink Agents

python3 -m venv flink-agents-env
source flink-agents-env/bin/activate
pip install flink-agents       

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-agents-api</artifactId>
    <version>${flink-agents.version}</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-agents-ide-support</artifactId>
    <version>${flink-agents.version}</version>
    <scope>provided</scope>
</dependency>

Writing the agent

The core structure of a Flink Agents job is as follows:

1. Register an LLM connection: Register a ChatModel connection by using the AgentsExecutionEnvironment. Connection information, such as the API key and endpoint URL, is configured through a ResourceDescriptor. For more information, see Chat Models.

2. Define the agent: Implement your custom agent, configure its prompt template, and define the event-handling logic by using the @action (Python) or @Action (Java) annotation.

3. Build the pipeline: Connect the input data stream to the agent for processing and output the analysis results.

For more information about concepts such as agent, prompt, tool, and memory, see the Flink Agents development documentation.

Testing locally

Local testing is handled automatically by a Flink MiniCluster, so you do not need to deploy a separate cluster.

Python

python your_agent_job.py        

Java

mvn exec:java -Dexec.mainClass="com.example.YourAgentJob"       

For more information, see the Flink Agents deployment documentation.

Pre-installed dependencies

The VVR engine image comes pre-installed with the Flink Agents core libraries and some related dependencies.

Python dependencies

Dependencies that are not pre-installed must be uploaded with your job. For more information, see Manage Python dependencies.

Java dependencies

The VVR engine image comes pre-installed with Flink Agents and its built-in LLM integrations. You only need to package third-party dependencies that are not built in into your job JAR.

Managing Java job dependencies

Use the maven-shade-plugin to package your job into a fat JAR. Configure the dependency scope according to the following rules:

• Flink Agents core dependencies (such as flink-agents-api and flink-agents-runtime): Set the scope to <scope>provided</scope> and do not package them into the fat JAR.

• Flink core dependencies (such as flink-streaming-java): Set the scope to <scope>provided</scope>.

• LLM integration dependencies: Handle these based on the following cases.

  • For integrations already supported by the Flink Agents community (such as OpenAI and Anthropic): Set the scope to <scope>provided</scope> and do not package them into the fat JAR. For a complete list of integrations supported in Flink Agents 0.2.1, see Built-in Providers.

  • For integrations not yet supported by the community: Set the scope to <scope>compile</scope> and package them into the fat JAR.

• Libraries that conflict with Flink (such as Jackson): Configure relocation in the maven-shade-plugin.

For a complete pom.xml configuration example, see the quickstart-java-src.zip file.

Python job

The submission process is the same as for a standard PyFlink job. For more information, see Develop a Python job. The key parameters are as follows:

Java job

The submission process is the same as for a standard Flink JAR job. For more information, see Develop a JAR job. The key parameters are as follows:

Common runtime parameters

python.executable: python3.10
python.client.executable: python3.10
containerized.master.env.FLINK_HOME: /flink
containerized.taskmanager.env.FLINK_HOME: /flink

containerized.master.env.<ENV_VAR_NAME>: <value>
containerized.taskmanager.env.<ENV_VAR_NAME>: <value>        

containerized.master.env.OPENAI_MODEL: qwen3.6-flash
containerized.taskmanager.env.OPENAI_MODEL: qwen3.6-flash