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

Agent types

Anda dapat membangun Workflow Agent atau ReAct Agent untuk berbagai kasus penggunaan.

Langkah 1: Unduh file pekerjaan contoh

Langkah 2: Unggah dan terapkan pekerjaan

1. Login ke Konsol Realtime Compute for Apache Flink.

2. Pada kolom Actions ruang kerja target, klik Console.

3. Di panel navigasi sebelah kiri, klik File Management. Kemudian, klik Upload Resource untuk mengunggah file pekerjaan yang telah Anda unduh.

4. Pada halaman Operations > Jobs, klik Deploy Job. Pilih Python Job atau JAR Job sesuai jenis pekerjaan Anda dan isi informasi penerapan.

Langkah 3: Mulai pekerjaan dan lihat hasilnya

1. Pada halaman Operations > Jobs, temukan pekerjaan Anda dan klik Start pada kolom Actions.

2. Pada kotak dialog Start Job, pilih stateless start dan klik Start.

3. Contoh ini menggunakan sumber data in-memory, sehingga pekerjaan secara otomatis berhenti setelah pemrosesan selesai. Tunggu hingga status berubah menjadi finished, lalu klik nama pekerjaan untuk membuka halaman detail.

4. Pada tab TaskManager, lihat log dan cari kata kunci Review analysis result: untuk melihat output.

Menginstal 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>

Menulis agent

Struktur inti pekerjaan Flink Agents adalah sebagai berikut:

1. Daftarkan koneksi LLM: Daftarkan koneksi ChatModel menggunakan AgentsExecutionEnvironment. Konfigurasikan clazz layanan model bawaan Flink menggunakan ResourceDescriptor. Untuk informasi lebih lanjut, lihat Chat Models. Contoh:

   @chat_model_connection
   @staticmethod
   def openai_connection() -> ResourceDescriptor:
       return ResourceDescriptor(
           clazz="flink_agents.integrations.chat_models.openai.openai_chat_model.OpenAIChatModelConnection"
       )
   
   @chat_model_setup
   @staticmethod
   def openai_chat_model() -> ResourceDescriptor:
       return ResourceDescriptor(
           clazz="flink_agents.integrations.chat_models.openai.openai_chat_model.OpenAIChatModelSetup",
           connection="openai_connection",
           model="qwen3.7-flash",
           temperature=0.7
       )

2. Definisikan agent: Implementasikan agent kustom Anda, konfigurasikan templat prompt-nya, dan definisikan logika penanganan event menggunakan anotasi @action (Python) atau @Action (Java).

3. Bangun pipeline: Hubungkan aliran data masukan ke agent untuk diproses dan keluarkan hasil analisis.

Untuk informasi lebih lanjut tentang konsep seperti agent, prompt, tool, dan memori, lihat dokumentasi pengembangan Flink Agents.

Pengujian lokal

Pengujian lokal ditangani secara otomatis oleh Flink MiniCluster, sehingga Anda tidak perlu menerapkan kluster terpisah.

Python

python your_agent_job.py        

Java

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

Untuk informasi lebih lanjut, lihat dokumentasi penerapan Flink Agents.

Dependensi yang telah dipra-instal

Image engine VVR telah dipra-instal dengan pustaka inti Flink Agents dan beberapa dependensi terkait.

Dependensi Python

Dependensi yang tidak dipra-instal harus diunggah bersama pekerjaan Anda. Untuk informasi lebih lanjut, lihat Kelola dependensi Python.

Dependensi Java

Engine VVR hanya berisi thin JAR Flink Agents (kode inti). Dependensi pihak ketiga untuk semua integrasi LLM harus dikemas ke dalam JAR pekerjaan.

Saat menerapkan pekerjaan, selain fat JAR bisnis Anda, Anda juga harus mengunggah flink-agents-dist.jar sebagai dependensi tambahan. JAR ini berisi runtime Flink Agents beserta semua integrasi yang didukung komunitas, dan diperlukan juga untuk pekerjaan Agent kustom.

Mengelola dependensi pekerjaan Java

Gunakan maven-shade-plugin untuk mengemas pekerjaan Anda ke dalam fat JAR. Konfigurasikan cakupan dependensi sesuai aturan berikut:

• Dependensi inti Flink Agents (seperti flink-agents-api dan flink-agents-runtime): Tetapkan cakupan ke <scope>provided</scope> dan jangan masukkan ke dalam fat JAR.

• Dependensi inti Flink (seperti flink-streaming-java): Tetapkan cakupan ke <scope>provided</scope>.

• Dependensi integrasi LLM: Tangani berdasarkan kasus berikut.

  • Untuk integrasi yang sudah didukung komunitas Flink Agents (seperti OpenAI dan Anthropic): Tetapkan cakupan ke <scope>provided</scope> dan jangan masukkan ke dalam fat JAR. Untuk daftar lengkap integrasi yang didukung di Flink Agents 0.2.1, lihat Built-in Providers.

  • Untuk integrasi yang belum didukung komunitas: Tetapkan cakupan ke <scope>compile</scope> dan masukkan ke dalam fat JAR.

• Pustaka yang bentrok dengan Flink (seperti Jackson): Konfigurasikan relocation di maven-shade-plugin.

Untuk contoh konfigurasi lengkap file pom.xml, lihat file quickstart-java-src.zip.

Python job

Proses pengiriman sama seperti pekerjaan PyFlink standar. Untuk informasi lebih lanjut, lihat Mengembangkan pekerjaan Python. Parameter utamanya adalah sebagai berikut:

Java job

Proses pengiriman sama seperti pekerjaan JAR Flink standar. Untuk informasi lebih lanjut, lihat Mengembangkan pekerjaan JAR. Parameter utamanya adalah sebagai berikut:

Parameter waktu proses umum

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