Realtime Compute for Apache Flink:Développement d'agents Flink (bêta)

Types d'agents

Créez des agents Workflow ou ReAct selon vos cas d'utilisation.

Étape 1 : Télécharger les fichiers d'exemple de la tâche

Étape 2 : Téléverser et déployer la tâche

1. Connectez-vous à la console Realtime Compute for Apache Flink.

2. Dans la colonne Actions de l'espace de travail cible, cliquez sur Console.

3. Dans le volet de navigation de gauche, cliquez sur File Management. Puis, cliquez sur Upload Resource pour téléverser les fichiers de tâche que vous avez téléchargés.

4. Sur la page Operations > Jobs, cliquez sur Deploy Job. Sélectionnez Python Job ou JAR Job selon le type de votre tâche et renseignez les informations de déploiement.

Étape 3 : Démarrer la tâche et afficher les résultats

1. Sur la page Operations > Jobs, localisez votre tâche et cliquez sur Start dans la colonne Actions.

2. Dans la boîte de dialogue Start Job, sélectionnez stateless start et cliquez sur Start.

3. L'exemple utilise une source de données en mémoire, donc la tâche se termine automatiquement une fois le traitement terminé. Attendez que le statut passe à finished, puis cliquez sur le nom de la tâche pour accéder à la page des détails.

4. Sous l'onglet TaskManager, consultez les journaux et recherchez le mot-clé Review analysis result: pour voir la sortie.

Installation de 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>

Écriture de l'agent

La structure centrale d'une tâche Flink Agents est la suivante :

1. Enregistrer une connexion LLM : Enregistrez une connexion ChatModel en utilisant AgentsExecutionEnvironment. Configurez les informations de connexion, telles que la clé API et l'URL de l'endpoint, via un ResourceDescriptor. Pour plus d'informations, consultez Chat Models.

2. Définir l'agent : Implémentez votre agent personnalisé, configurez son modèle d'invite et définissez la logique de gestion des événements en utilisant l'annotation @action (Python) ou @Action (Java).

3. Construire le pipeline : Connectez le flux de données d'entrée à l'agent pour le traitement et produisez les résultats d'analyse.

Pour plus d'informations sur les concepts tels que l'agent, l'invite, l'outil et la mémoire, consultez la documentation de développement Flink Agents.

Tests locaux

Un MiniCluster Flink gère automatiquement les tests locaux ; vous n'avez donc pas besoin de déployer un cluster séparé.

Python

python your_agent_job.py        

Java

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

Pour plus d'informations, consultez la documentation de déploiement Flink Agents.

Dépendances préinstallées

L'image du moteur VVR inclut les bibliothèques principales de Flink Agents et certaines dépendances associées préinstallées.

Dépendances Python

Téléversez les dépendances non préinstallées avec votre tâche. Pour plus d'informations, consultez Gérer les dépendances Python.

Dépendances Java

L'image du moteur VVR inclut Flink Agents et ses intégrations LLM intégrées. Empaquetez uniquement les dépendances tierces non intégrées dans le JAR de votre tâche.

Gestion des dépendances des tâches Java

Utilisez le plugin maven-shade-plugin pour empaqueter votre tâche dans un JAR fat. Configurez la portée des dépendances selon les règles suivantes :

• Dépendances principales de Flink Agents (telles que flink-agents-api et flink-agents-runtime) : Définissez la portée sur <scope>provided</scope> et ne les incluez pas dans le JAR fat.

• Dépendances principales de Flink (telles que flink-streaming-java) : Définissez la portée sur <scope>provided</scope>.

• Dépendances d'intégration LLM : Traitez-les selon les cas suivants.

  • Pour les intégrations déjà prises en charge par la communauté Flink Agents (telles que OpenAI et Anthropic) : Définissez la portée sur <scope>provided</scope> et ne les incluez pas dans le JAR fat. Pour la liste complète des intégrations prises en charge dans Flink Agents 0,2.1, consultez Built-in Providers.

  • Pour les intégrations non encore prises en charge par la communauté : Définissez la portée sur <scope>compile</scope> et incluez-les dans le JAR fat.

• Bibliothèques en conflit avec Flink (telles que Jackson) : Configurez la relocation dans le plugin maven-shade-plugin.

Pour un exemple complet de configuration pom.xml, consultez le fichier quickstart-java-src.zip.

Tâche Python

Le processus de soumission est identique à celui d'une tâche PyFlink standard. Pour plus d'informations, consultez Développer une tâche Python. Les paramètres clés sont les suivants :

Tâche Java

Le processus de soumission est identique à celui d'une tâche JAR Flink standard. Pour plus d'informations, consultez Développer une tâche JAR. Les paramètres clés sont les suivants :

Paramètres d'exécution courants

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