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

Tipos de agente

Você pode criar agentes de Workflow ou ReAct para diferentes casos de uso.

Etapa 1: Baixe os arquivos do job de exemplo

Etapa 2: Envie e implante o job

1. Em seguida, acesse o console do Realtime Compute for Apache Flink.

2. Na coluna Actions do workspace de destino, clique em Console.

3. No painel de navegação à esquerda, clique em File Management e, em seguida, clique em Upload Resource para enviar os arquivos do job baixados.

4. Na página Operations > Jobs, clique em Deploy Job. Selecione Python Job ou JAR Job conforme o tipo do seu job e preencha as informações de implantação.

Etapa 3: Execute o job e visualize os resultados

1. Na página Operations > Jobs, localize o seu job e clique em Start na coluna Actions.

2. Na caixa de diálogo Start Job, selecione stateless start e clique em Start.

3. O exemplo usa uma fonte de dados em memória; portanto, o job termina automaticamente após a conclusão do processamento. Aguarde o status mudar para finished e clique no nome do job para acessar a página de detalhes.

4. Na aba TaskManager, visualize os logs e pesquise a palavra-chave Review analysis result: para conferir a saída.

Instale o 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>

Criando o agente

A estrutura principal de um job do Flink Agents é a seguinte:

1. Registre uma conexão de LLM: registre uma conexão ChatModel usando o AgentsExecutionEnvironment. Configure a classe do service de modelo integrado do Flink com um ResourceDescriptor. Para mais informações, consulte Modelos de Chat. Exemplo:

   @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. Defina o agente: implemente seu agente personalizado, configure o modelo de prompt e defina a lógica de tratamento de eventos com a anotação @action (Python) ou @Action (Java).

3. Construa o pipeline: conecte o fluxo de dados de entrada ao agente para processamento e gere os resultados da análise.

Para mais informações sobre conceitos como agente, prompt, ferramenta e memória, consulte a documentação de desenvolvimento do Flink Agents.

Testes locais

O Flink MiniCluster lida com os testes locais automaticamente, eliminando a necessidade de implantar um cluster separado.

Python

python your_agent_job.py        

Java

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

Para mais informações, consulte a documentação de implantação do Flink Agents.

Dependências pré-instaladas

A imagem do motor VVR já inclui as bibliotecas principais do Flink Agents e algumas dependências relacionadas pré-instaladas.

Dependências de Python

Envie as dependências não pré-instaladas junto com o seu job. Para mais informações, consulte Manage Python dependencies.

Dependências de Java

O motor VVR contém apenas o JAR thin do Flink Agents (código principal). Empacote as dependências de terceiros para todas as integrações de LLM no JAR do job.

Ao implantar o job, além do seu JAR fat de negócios, envie também o flink-agents-dist.jar como dependência adicional. Esse JAR contém o tempo de execução do Flink Agents e todas as integrações suportadas pela comunidade, sendo também necessário para jobs de Agent personalizados.

Gerencie dependências de jobs em Java

Use o maven-shade-plugin para empacotar seu job em um JAR fat. Configure o escopo da dependência de acordo com as seguintes regras:

• Dependências principais do Flink Agents (como flink-agents-api e flink-agents-runtime): defina o escopo como <scope>provided</scope> e não as empacote no JAR fat.

• Dependências principais do Flink (como flink-streaming-java): defina o escopo como <scope>provided</scope>.

• Dependências de integração de LLM: trate-as com base nos seguintes casos.

  • Para integrações já suportadas pela comunidade Flink Agents (como OpenAI e Anthropic): defina o escopo como <scope>provided</scope> e não as empacote no JAR fat. Para a lista completa das integrações suportadas no Flink Agents 0.2.1, consulte Provedores Integrados.

  • Para integrações ainda não suportadas pela comunidade: defina o escopo como <scope>compile</scope> e empacote-as no JAR fat.

• Bibliotecas que entram em conflito com o Flink (como Jackson): configure o relocation no maven-shade-plugin.

Para um exemplo completo de configuração do pom.xml, consulte o arquivo quickstart-java-src.zip.

Job em Python

O processo de envio é o mesmo de um job PyFlink padrão. Para mais informações, consulte Desenvolver um job em Python. Os principais parâmetros são:

Job em Java

O processo de envio é o mesmo de um job JAR do Flink padrão. Para mais informações, consulte Desenvolver um job JAR. Os principais parâmetros são:

Parâmetros de tempo de execução comuns

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